Try to use const-ref better in rpl.

Telegram/SourceFiles/base/observer.h

@@ -470,7 +470,8 @@ inline rpl::producer<Type> ObservableViewer(
 		auto lifetime = rpl::lifetime();
 		lifetime.make_state<base::Subscription>(
 			observable.add_subscription([consumer](auto &&update) {
-				consumer.put_next_copy(update);
+				consumer.put_next_forward(
+					std::forward<decltype(update)>(update));
 			}));
 		return lifetime;
 	};

Telegram/SourceFiles/history/history_shared_media.cpp

@@ -434,14 +434,14 @@ rpl::producer<SharedMediaSlice> SharedMediaViewer(
 			limitBefore,
 			limitAfter);
 		auto applyUpdate = [=](auto &&update) {
-			if (builder->applyUpdate(std::move(update))) {
+			if (builder->applyUpdate(std::forward<decltype(update)>(update))) {
 				consumer.put_next(builder->snapshot());
 			}
 		};
 		auto requestMediaAround = [
 			peer = App::peer(key.peerId),
 			type = key.type
-		](SharedMediaSliceBuilder::AroundData data) {
+		](const SharedMediaSliceBuilder::AroundData &data) {
 			Auth().api().requestSharedMedia(
 				peer,
 				type,
@@ -465,8 +465,7 @@ rpl::producer<SharedMediaSlice> SharedMediaViewer(
 		Auth().storage().query(Storage::SharedMediaQuery(
 			key,
 			limitBefore,
-			limitAfter
+			limitAfter))
-		))
 			| rpl::on_next(applyUpdate)
 			| rpl::on_done([=] { builder->checkInsufficientMedia(); })
 			| rpl::start(lifetime);

Telegram/SourceFiles/history/history_user_photos.cpp

@@ -222,11 +222,12 @@ rpl::producer<UserPhotosSlice> UserPhotosViewer(
 			limitBefore,
 			limitAfter);
 		auto applyUpdate = [=](auto &&update) {
-			if (builder->applyUpdate(std::move(update))) {
+			if (builder->applyUpdate(std::forward<decltype(update)>(update))) {
 				consumer.put_next(builder->snapshot());
 			}
 		};
-		auto requestPhotosAround = [user = App::user(key.userId)](PhotoId photoId) {
+		auto requestPhotosAround = [user = App::user(key.userId)](
+				PhotoId photoId) {
 			Auth().api().requestUserPhotos(user, photoId);
 		};
 		builder->insufficientPhotosAround()

Telegram/SourceFiles/rpl/consumer.h

@@ -25,6 +25,28 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #include <rpl/lifetime.h>
 
 namespace rpl {
+namespace details {
+
+template <typename Arg>
+const Arg &const_ref_val();
+
+template <
+	typename Func,
+	typename Arg,
+	typename = decltype(std::declval<Func>()(const_ref_val<Arg>()))>
+void const_ref_call_helper(Func &handler, const Arg &value, int) {
+	handler(value);
+}
+
+template <
+	typename Func,
+	typename Arg>
+void const_ref_call_helper(Func &handler, const Arg &value, double) {
+	auto copy = value;
+	handler(std::move(copy));
+}
+
+} // namespace details
 
 struct no_value {
 	no_value() = delete;
@@ -57,8 +79,20 @@ public:
 
 	bool put_next(Value &&value) const;
 	bool put_next_copy(const Value &value) const;
+	bool put_next_forward(Value &&value) const {
+		return put_next(std::move(value));
+	}
+	bool put_next_forward(const Value &value) const {
+		return put_next_copy(value);
+	}
 	void put_error(Error &&error) const;
 	void put_error_copy(const Error &error) const;
+	void put_error_forward(Error &&error) const {
+		return put_error(std::move(error));
+	}
+	void put_error_forward(const Error &error) const {
+		return put_error_copy(error);
+	}
 	void put_done() const;
 
 	void add_lifetime(lifetime &&lifetime) const;
@@ -108,7 +142,9 @@ template <typename Value, typename Error>
 class consumer<Value, Error>::abstract_consumer_instance {
 public:
 	virtual bool put_next(Value &&value) = 0;
+	virtual bool put_next_copy(const Value &value) = 0;
 	virtual void put_error(Error &&error) = 0;
+	virtual void put_error_copy(const Error &error) = 0;
 	virtual void put_done() = 0;
 
 	void add_lifetime(lifetime &&lifetime);
@@ -141,7 +177,9 @@ public:
 	}
 
 	bool put_next(Value &&value) override;
+	bool put_next_copy(const Value &value) override;
 	void put_error(Error &&error) override;
+	void put_error_copy(const Error &error) override;
 	void put_done() override;
 
 private:
@@ -197,8 +235,13 @@ bool consumer<Value, Error>::put_next(Value &&value) const {
 
 template <typename Value, typename Error>
 bool consumer<Value, Error>::put_next_copy(const Value &value) const {
-	auto copy = value;
+	if (_instance) {
-	return put_next(std::move(copy));
+		if (_instance->put_next_copy(value)) {
+			return true;
+		}
+		_instance = nullptr;
+	}
+	return false;
 }
 
 template <typename Value, typename Error>
@@ -210,8 +253,9 @@ void consumer<Value, Error>::put_error(Error &&error) const {
 
 template <typename Value, typename Error>
 void consumer<Value, Error>::put_error_copy(const Error &error) const {
-	auto copy = error;
+	if (_instance) {
-	return put_error(std::move(error));
+		std::exchange(_instance, nullptr)->put_error_copy(error);
+	}
 }
 
 template <typename Value, typename Error>
@@ -293,6 +337,21 @@ bool consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_nex
 	return true;
 }
 
+template <typename Value, typename Error>
+template <typename OnNext, typename OnError, typename OnDone>
+bool consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_next_copy(
+		const Value &value) {
+	std::unique_lock<std::mutex> lock(this->_mutex);
+	if (this->_terminated) {
+		return false;
+	}
+	auto handler = this->_next;
+	lock.unlock();
+
+	details::const_ref_call_helper(handler, value, 0);
+	return true;
+}
+
 template <typename Value, typename Error>
 template <typename OnNext, typename OnError, typename OnDone>
 void consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_error(
@@ -307,6 +366,20 @@ void consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_err
 	}
 }
 
+template <typename Value, typename Error>
+template <typename OnNext, typename OnError, typename OnDone>
+void consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_error_copy(
+		const Error &error) {
+	std::unique_lock<std::mutex> lock(this->_mutex);
+	if (!this->_terminated) {
+		auto handler = std::move(this->_error);
+		lock.unlock();
+
+		details::const_ref_call_helper(handler, error, 0);
+		this->terminate();
+	}
+}
+
 template <typename Value, typename Error>
 template <typename OnNext, typename OnError, typename OnDone>
 void consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_done() {

Telegram/SourceFiles/rpl/distinct_until_changed.h

@@ -37,13 +37,13 @@ public:
 				base::optional<Value>
 			>();
 			return std::move(initial).start(
-				[consumer, previous](Value &&value) {
+				[consumer, previous](auto &&value) {
 					if (!(*previous) || (**previous) != value) {
 						*previous = value;
-						consumer.put_next(std::move(value));
+						consumer.put_next_forward(std::forward<decltype(value)>(value));
 					}
-				}, [consumer](Error &&error) {
+				}, [consumer](auto &&error) {
-					consumer.put_error(std::move(error));
+					consumer.put_error_forward(std::forward<decltype(error)>(error));
 				}, [consumer] {
 					consumer.put_done();
 				});

Telegram/SourceFiles/rpl/event_stream.h

@@ -37,10 +37,13 @@ public:
 	event_stream();
 	event_stream(event_stream &&other);
 
-	void fire(Value &&value);
+	template <typename OtherValue>
+	void fire_forward(OtherValue &&value);
+	void fire(Value &&value) {
+		return fire_forward(std::move(value));
+	}
 	void fire_copy(const Value &value) {
-		auto copy = value;
+		return fire_forward(value);
-		fire(std::move(copy));
 	}
 	producer<Value, no_error> events() const;
 	producer<Value, no_error> events_starting_with(
@@ -49,8 +52,7 @@ public:
 	}
 	producer<Value, no_error> events_starting_with_copy(
 			const Value &value) const {
-		auto copy = value;
+		return single(value) | then(events());
-		return events_starting_with(std::move(copy));
 	}
 
 	~event_stream();
@@ -72,7 +74,8 @@ event_stream<Value>::event_stream(event_stream &&other)
 }
 
 template <typename Value>
-void event_stream<Value>::fire(Value &&value) {
+template <typename OtherValue>
+void event_stream<Value>::fire_forward(OtherValue &&value) {
 	if (!_consumers) {
 		return;
 	}
@@ -87,8 +90,8 @@ void event_stream<Value>::fire(Value &&value) {
 			return !consumer.put_next_copy(value);
 		});
 
-		// Move value for the last consumer.
+		// Perhaps move value for the last consumer.
-		if (prev->put_next(std::move(value))) {
+		if (prev->put_next_forward(std::forward<OtherValue>(value))) {
 			if (removeFrom != prev) {
 				*removeFrom++ = std::move(*prev);
 			} else {
@@ -152,8 +155,8 @@ event_stream<Value>::~event_stream() {
 
 template <typename Value>
 inline auto to_stream(event_stream<Value> &stream) {
-	return on_next([&stream](Value &&value) {
+	return on_next([&stream](auto &&value) {
-		stream.fire(std::move(value));
+		stream.fire_forward(std::forward<decltype(value)>(value));
 	});
 }
 

Telegram/SourceFiles/rpl/filter.h

@@ -45,13 +45,13 @@ public:
 				[
 					consumer,
 					predicate = std::move(predicate)
-				](Value &&value) {
+				](auto &&value) {
 					const auto &immutable = value;
 					if (predicate(immutable)) {
-						consumer.put_next(std::move(value));
+						consumer.put_next_forward(std::forward<decltype(value)>(value));
 					}
-				}, [consumer](Error &&error) {
+				}, [consumer](auto &&error) {
-					consumer.put_error(std::move(error));
+					consumer.put_error_forward(std::forward<decltype(error)>(error));
 				}, [consumer] {
 					consumer.put_done();
 				});

Telegram/SourceFiles/rpl/flatten_latest.h

@@ -40,10 +40,10 @@ public:
 			[consumer, state](rpl::producer<Value, Error> &&inner) {
 				state->finished = false;
 				state->alive = std::move(inner).start(
-				[consumer](Value &&value) {
+				[consumer](auto &&value) {
-					consumer.put_next(std::move(value));
+					consumer.put_next_forward(std::forward<decltype(value)>(value));
-				}, [consumer](Error &&error) {
+				}, [consumer](auto &&error) {
-					consumer.put_error(std::move(error));
+					consumer.put_error_forward(std::forward<decltype(error)>(error));
 				}, [consumer, state] {
 					if (state->finished) {
 						consumer.put_done();
@@ -51,8 +51,8 @@ public:
 						state->finished = true;
 					}
 				});
-			}, [consumer](Error &&error) {
+			}, [consumer](auto &&error) {
-				consumer.put_error(std::move(error));
+				consumer.put_error_forward(std::forward<decltype(error)>(error));
 			}, [consumer, state] {
 				if (state->finished) {
 					consumer.put_done();

Telegram/SourceFiles/rpl/map.h

@@ -25,6 +25,40 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 namespace rpl {
 namespace details {
 
+template <
+	typename Transform,
+	typename NewValue,
+	typename Error,
+	typename Value>
+class map_transform_helper {
+public:
+	map_transform_helper(
+		const consumer<NewValue, Error> &consumer,
+		Transform &&transform)
+		: _transform(std::move(transform))
+		, _consumer(consumer) {
+	}
+	template <
+		typename OtherValue,
+		typename = std::enable_if_t<
+			std::is_rvalue_reference_v<OtherValue&&>>>
+	void operator()(OtherValue &&value) const {
+		_consumer.put_next_forward(_transform(std::move(value)));
+	}
+	template <
+		typename OtherValue,
+		typename = decltype(
+			std::declval<Transform>()(const_ref_val<OtherValue>()))>
+	void operator()(const OtherValue &value) const {
+		_consumer.put_next_forward(_transform(value));
+	}
+
+private:
+	consumer<NewValue, Error> _consumer;
+	Transform _transform;
+
+};
+
 template <typename Transform>
 class map_helper {
 public:
@@ -46,13 +80,11 @@ public:
 			transform = std::move(_transform)
 		](const consumer<NewValue, Error> &consumer) mutable {
 			return std::move(initial).start(
-				[
+			map_transform_helper<Transform, NewValue, Error, Value>(
-					consumer,
+				consumer,
-					transform = std::move(transform)
+				std::move(transform)
-				](Value &&value) {
+			), [consumer](auto &&error) {
-				consumer.put_next(transform(std::move(value)));
+				consumer.put_error_forward(std::forward<decltype(error)>(error));
-			}, [consumer](Error &&error) {
-				consumer.put_error(std::move(error));
 			}, [consumer] {
 				consumer.put_done();
 			});

Telegram/SourceFiles/rpl/then.h

@@ -34,19 +34,19 @@ auto then(producer<Value, Error> &&following) {
 			following = std::move(following)
 		](const consumer<Value, Error> &consumer) mutable {
 			return std::move(initial).start(
-			[consumer](Value &&value) {
+			[consumer](auto &&value) {
-				consumer.put_next(std::move(value));
+				consumer.put_next_forward(std::forward<decltype(value)>(value));
-			}, [consumer](Error &&error) {
+			}, [consumer](auto &&error) {
-				consumer.put_error(std::move(error));
+				consumer.put_error_forward(std::forward<decltype(error)>(error));
 			}, [
 				consumer,
 				following = std::move(following)
 			]() mutable {
 				consumer.add_lifetime(std::move(following).start(
-				[consumer](Value &&value) {
+				[consumer](auto &&value) {
-					consumer.put_next(std::move(value));
+					consumer.put_next_forward(std::forward<decltype(value)>(value));
-				}, [consumer](Error &&error) {
+				}, [consumer](auto &&error) {
-					consumer.put_error(std::move(error));
+					consumer.put_error_forward(std::forward<decltype(error)>(error));
 				}, [consumer] {
 					consumer.put_done();
 				}));

Telegram/SourceFiles/storage/storage_shared_media.cpp

@@ -234,7 +234,7 @@ std::map<PeerId, SharedMedia::Lists>::iterator
 		auto type = static_cast<SharedMediaType>(index);
 
 		list.sliceUpdated()
-			| rpl::on_next([this, peer, type](SliceUpdate &&update) {
+			| rpl::on_next([this, peer, type](const SliceUpdate &update) {
 				_sliceUpdated.fire(SharedMediaSliceUpdate(
 					peer,
 					type,

Telegram/SourceFiles/storage/storage_user_photos.cpp

@@ -122,7 +122,7 @@ UserPhotos::enforceLists(UserId user) {
 	}
 	result = _lists.emplace(user, List {}).first;
 	result->second.sliceUpdated(
-	) | rpl::on_next([this, user](SliceUpdate &&update) {
+	) | rpl::on_next([this, user](const SliceUpdate &update) {
 		_sliceUpdated.fire(UserPhotosSliceUpdate(
 			user,
 			update.photoIds,

Telegram/SourceFiles/ui/widgets/labels.cpp

@@ -171,8 +171,8 @@ FlatLabel::FlatLabel(
 , _st(st)
 , _contextCopyText(lang(lng_context_copy_text)) {
 	std::move(text)
-		| rpl::on_next([this](QString &&value) {
+		| rpl::on_next([this](const QString &value) {
-			setText(std::move(value));
+			setText(value);
 		})
 		| rpl::start(lifetime());
 }
@@ -186,8 +186,8 @@ FlatLabel::FlatLabel(
 , _st(st)
 , _contextCopyText(lang(lng_context_copy_text)) {
 	std::move(text)
-		| rpl::on_next([this](TextWithEntities &&value) {
+		| rpl::on_next([this](const TextWithEntities &value) {
-			setMarkedText(std::move(value));
+			setMarkedText(value);
 		})
 		| rpl::start(lifetime());
 }

Telegram/SourceFiles/ui/wrap/wrap.h

@@ -126,7 +126,7 @@ Wrap<Widget, RpWidget>::Wrap(QWidget *parent, object_ptr<Widget> child)
 , _wrapped(std::move(child)) {
 	if (_wrapped) {
 		_wrapped->sizeValue()
-			| rpl::on_next([this](QSize &&value) {
+			| rpl::on_next([this](const QSize &value) {
 				wrappedSizeUpdated(value);
 			})
 			| rpl::start(lifetime());

Telegram/SourceFiles/window/player_wrap_widget.cpp

@@ -27,13 +27,13 @@ namespace Window {
 PlayerWrapWidget::PlayerWrapWidget(QWidget *parent)
 : Parent(parent, object_ptr<Media::Player::Widget>(parent)) {
 	sizeValue()
-		| rpl::on_next([this](QSize &&size) {
+		| rpl::on_next([this](const QSize &size) {
 			updateShadowGeometry(size);
 		})
 		| rpl::start(lifetime());
 }
 
-void PlayerWrapWidget::updateShadowGeometry(QSize size) {
+void PlayerWrapWidget::updateShadowGeometry(const QSize &size) {
 	auto skip = Adaptive::OneColumn() ? 0 : st::lineWidth;
 	entity()->setShadowGeometryToLeft(
 		skip,

Telegram/SourceFiles/window/player_wrap_widget.h

@@ -29,7 +29,7 @@ public:
 	}
 
 private:
-	void updateShadowGeometry(QSize size);
+	void updateShadowGeometry(const QSize &size);
 
 };
 

Telegram/gyp/tests/tests.gyp

@@ -98,6 +98,7 @@
     ],
     'sources': [
       '<(src_loc)/rpl/before_next.h',
+      '<(src_loc)/rpl/combine_latest.h',
       '<(src_loc)/rpl/complete.h',
       '<(src_loc)/rpl/consumer.h',
       '<(src_loc)/rpl/deferred.h',