Use rvalue references in rpl next / error.

Telegram/SourceFiles/base/algorithm.h

@@ -32,38 +32,6 @@ inline constexpr size_t array_size(const Type(&)[Size]) {
 	return Size;
 }
 
-// This version of remove_if allows predicate to push_back() items.
-// The added items won't be tested for predicate but just left in the container.
-template <typename Container, typename Predicate>
-void push_back_safe_remove_if(
-	Container &&container,
-	Predicate &&predicate) {
-	auto first = size_t(0);
-	auto count = container.size();
-	auto moveFrom = first;
-	for (; moveFrom != count; ++moveFrom) {
-		if (predicate(container[moveFrom])) {
-			break;
-		}
-	}
-	if (moveFrom != count) {
-		auto moveTo = moveFrom;
-		for (++moveFrom; moveFrom != count; ++moveFrom) {
-			if (!predicate(container[moveFrom])) {
-				container[moveTo++] = std::move(container[moveFrom]);
-			}
-		}
-
-		// Move items that we've added while checking the initial items.
-		count = container.size();
-		for (; moveFrom != count; ++moveFrom) {
-			container[moveTo++] = std::move(container[moveFrom]);
-		}
-
-		container.erase(container.begin() + moveTo, container.end());
-	}
-}
-
 template <typename Range, typename Method>
 decltype(auto) for_each(Range &&range, Method &&method) {
 	return std::for_each(

Telegram/SourceFiles/base/algorithm_tests.cpp

@@ -20,20 +20,29 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 */
 #include "catch.hpp"
 
-#include "base/algorithm.h"
+#include "base/index_based_iterator.h"
 
-TEST_CASE("push_back_safe_remove_if tests", "[base::algorithm]") {
+TEST_CASE("index_based_iterator tests", "[base::algorithm]") {
 	auto v = std::vector<int>();
 
-	SECTION("doesn't change an empty vector") {
-		base::push_back_safe_remove_if(v, [](int) { return true; });
-		REQUIRE(v.empty());
-	}
-
 	v.insert(v.end(), { 1, 2, 3, 4, 5, 4, 3, 2, 1 });
-
+	auto push_back_safe_remove_if = [](auto &v, auto predicate) {
+		auto begin = base::index_based_begin(v);
+		auto end = base::index_based_end(v);
+		auto from = std::remove_if(begin, end, predicate);
+		if (from != end) {
+			auto newEnd = base::index_based_end(v);
+			if (newEnd != end) {
+				REQUIRE(newEnd > end);
+				while (end != newEnd) {
+					*from++ = *end++;
+				}
+			}
+			v.erase(from.base(), newEnd.base());
+		}
+	};
 	SECTION("allows to push_back from predicate") {
-		base::push_back_safe_remove_if(v, [&v](int value) {
+		push_back_safe_remove_if(v, [&v](int value) {
 			v.push_back(value);
 			return (value % 2) == 1;
 		});
@@ -42,7 +51,7 @@ TEST_CASE("push_back_safe_remove_if tests", "[base::algorithm]") {
 	}
 
 	SECTION("allows to push_back while removing all") {
-		base::push_back_safe_remove_if(v, [&v](int value) {
+		push_back_safe_remove_if(v, [&v](int value) {
 			if (value == 5) {
 				v.push_back(value);
 			}

Telegram/SourceFiles/base/index_based_iterator.h

@@ -0,0 +1,135 @@
+/*
+This file is part of Telegram Desktop,
+the official desktop version of Telegram messaging app, see https://telegram.org
+
+Telegram Desktop is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+It is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+GNU General Public License for more details.
+
+In addition, as a special exception, the copyright holders give permission
+to link the code of portions of this program with the OpenSSL library.
+
+Full license: https://github.com/telegramdesktop/tdesktop/blob/master/LICENSE
+Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
+*/
+#pragma once
+
+#include "base/assertion.h"
+
+namespace base {
+
+template <typename Container>
+class index_based_iterator {
+public:
+	using iterator_category = std::random_access_iterator_tag;
+
+	using value_type = typename Container::value_type;
+	using difference_type = typename Container::difference_type;
+	using pointer = typename Container::pointer;
+	using reference = typename Container::reference;
+
+	index_based_iterator(
+		Container *container,
+		typename Container::iterator impl)
+		: _container(container)
+		, _index(impl - _container->begin()) {
+	}
+
+	reference operator*() const {
+		return *(_container->begin() + _index);
+	}
+	pointer operator->() const {
+		return std::addressof(**this);
+	}
+	index_based_iterator &operator++() {
+		++_index;
+		return *this;
+	}
+	index_based_iterator operator++(int) {
+		auto copy = *this;
+		++*this;
+		return copy;
+	}
+	index_based_iterator &operator--() {
+		--_index;
+		return *this;
+	}
+	index_based_iterator operator--(int) {
+		auto copy = *this;
+		--*this;
+		return copy;
+	}
+	index_based_iterator &operator+=(difference_type offset) {
+		_index += offset;
+		return *this;
+	}
+	index_based_iterator operator+(difference_type offset) const {
+		auto copy = *this;
+		copy += offset;
+		return copy;
+	}
+	index_based_iterator &operator-=(difference_type offset) {
+		_index -= offset;
+		return *this;
+	}
+	index_based_iterator operator-(difference_type offset) const {
+		auto copy = *this;
+		copy -= offset;
+		return copy;
+	}
+	difference_type operator-(const index_based_iterator &other) const {
+		return _index - other._index;
+	}
+	reference operator[](difference_type offset) const {
+		return *(*this + offset);
+	}
+
+	bool operator==(const index_based_iterator &other) const {
+		Expects(_container == other._container);
+		return _index == other._index;
+	}
+	bool operator!=(const index_based_iterator &other) const {
+		Expects(_container == other._container);
+		return _index != other._index;
+	}
+	bool operator<(const index_based_iterator &other) const {
+		Expects(_container == other._container);
+		return _index < other._index;
+	}
+	bool operator>(const index_based_iterator &other) const {
+		return other < *this;
+	}
+	bool operator<=(const index_based_iterator &other) const {
+		return !(other < *this);
+	}
+	bool operator>=(const index_based_iterator &other) const {
+		return !(*this < other);
+	}
+
+	typename Container::iterator base() const {
+		return _container->begin() + _index;
+	}
+
+private:
+	Container *_container = nullptr;
+	difference_type _index = 0;
+
+};
+
+template <typename Container>
+index_based_iterator<Container> index_based_begin(Container &container) {
+	return { &container, std::begin(container) };
+}
+
+template <typename Container>
+index_based_iterator<Container> index_based_end(Container &container) {
+	return { &container, std::end(container) };
+}
+
+} // namespace base

Telegram/SourceFiles/rpl/consumer.h

@@ -25,12 +25,13 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 
 namespace rpl {
 
-template <char Tag>
-struct no_type {
-	no_type() = delete;
+struct no_value {
+	no_value() = delete;
+};
+
+struct no_error {
+	no_error() = delete;
 };
-using no_value = no_type<'V'>;
-using no_error = no_type<'E'>;
 
 struct empty_value {
 };
@@ -50,8 +51,10 @@ public:
 		OnError &&error,
 		OnDone &&done);
 
-	bool put_next(Value value) const;
-	void put_error(Error error) const;
+	bool put_next(Value &&value) const;
+	bool put_next_copy(const Value &value) const;
+	void put_error(Error &&error) const;
+	void put_error_copy(const Error &error) const;
 	void put_done() const;
 
 	void set_lifetime(lifetime &&lifetime) const;
@@ -96,8 +99,8 @@ private:
 template <typename Value, typename Error>
 class consumer<Value, Error>::abstract_consumer_instance {
 public:
-	virtual bool put_next(Value value) = 0;
-	virtual void put_error(Error error) = 0;
+	virtual bool put_next(Value &&value) = 0;
+	virtual void put_error(Error &&error) = 0;
 	virtual void put_done() = 0;
 
 	void set_lifetime(lifetime &&lifetime);
@@ -125,8 +128,8 @@ public:
 		, _done(std::forward<OnDoneImpl>(done)) {
 	}
 
-	bool put_next(Value value) override;
-	void put_error(Error error) override;
+	bool put_next(Value &&value) override;
+	void put_error(Error &&error) override;
 	void put_done() override;
 
 private:
@@ -170,7 +173,7 @@ consumer<Value, Error>::consumer(
 }
 
 template <typename Value, typename Error>
-bool consumer<Value, Error>::put_next(Value value) const {
+bool consumer<Value, Error>::put_next(Value &&value) const {
 	if (_instance) {
 		if (_instance->put_next(std::move(value))) {
 			return true;
@@ -181,12 +184,24 @@ bool consumer<Value, Error>::put_next(Value value) const {
 }
 
 template <typename Value, typename Error>
-void consumer<Value, Error>::put_error(Error error) const {
+bool consumer<Value, Error>::put_next_copy(const Value &value) const {
+	auto copy = value;
+	return put_next(std::move(copy));
+}
+
+template <typename Value, typename Error>
+void consumer<Value, Error>::put_error(Error &&error) const {
 	if (_instance) {
 		std::exchange(_instance, nullptr)->put_error(std::move(error));
 	}
 }
 
+template <typename Value, typename Error>
+void consumer<Value, Error>::put_error_copy(const Error &error) const {
+	auto copy = error;
+	return put_error(std::move(error));
+}
+
 template <typename Value, typename Error>
 void consumer<Value, Error>::put_done() const {
 	if (_instance) {
@@ -238,7 +253,7 @@ void consumer<Value, Error>::abstract_consumer_instance::terminate() {
 template <typename Value, typename Error>
 template <typename OnNext, typename OnError, typename OnDone>
 bool consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_next(
-		Value value) {
+		Value &&value) {
 	std::unique_lock<std::mutex> lock(this->_mutex);
 	if (this->_terminated) {
 		return false;
@@ -253,7 +268,7 @@ bool consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_nex
 template <typename Value, typename Error>
 template <typename OnNext, typename OnError, typename OnDone>
 void consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::put_error(
-		Error error) {
+		Error &&error) {
 	std::unique_lock<std::mutex> lock(this->_mutex);
 	if (!this->_terminated) {
 		auto handler = std::move(this->_error);

Telegram/SourceFiles/rpl/event_stream.h

@@ -23,6 +23,7 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #include "producer.h"
 #include "base/algorithm.h"
 #include "base/assertion.h"
+#include "base/index_based_iterator.h"
 
 namespace rpl {
 
@@ -32,7 +33,7 @@ public:
 	event_stream();
 	event_stream(event_stream &&other);
 
-	void fire(Value value);
+	void fire(Value &&value);
 	producer<Value, no_error> events() const;
 
 	~event_stream();
@@ -57,11 +58,41 @@ event_stream<Value>::event_stream(event_stream &&other)
 }
 
 template <typename Value>
-void event_stream<Value>::fire(Value value) {
+void event_stream<Value>::fire(Value &&value) {
 	Expects(_consumers != nullptr);
-	base::push_back_safe_remove_if(*_consumers, [&](auto &consumer) {
-		return !consumer.put_next(value);
-	});
+	auto &consumers = *_consumers;
+	auto begin = base::index_based_begin(consumers);
+	auto end = base::index_based_end(consumers);
+	if (begin != end) {
+		// Copy value for every consumer except the last.
+		auto prev = end - 1;
+		auto removeFrom = std::remove_if(begin, prev, [&](auto &consumer) {
+			return !consumer.put_next_copy(value);
+		});
+
+		// Move value for the last consumer.
+		if (prev->put_next(std::move(value))) {
+			if (removeFrom != prev) {
+				*removeFrom++ = std::move(*prev);
+			} else {
+				++removeFrom;
+			}
+		}
+
+		if (removeFrom != end) {
+			// Move new consumers.
+			auto newEnd = base::index_based_end(consumers);
+			if (newEnd != end) {
+				Assert(newEnd > end);
+				while (end != newEnd) {
+					*removeFrom++ = *end++;
+				}
+			}
+
+			// Erase stale consumers.
+			consumers.erase(removeFrom.base(), consumers.end());
+		}
+	}
 }
 
 template <typename Value>

Telegram/SourceFiles/rpl/producer.h

@@ -105,10 +105,10 @@ inline decltype(auto) bind_on_next(OnNext &&handler) {
 			handler = std::forward<OnNext>(handler)
 		](consumer<no_value, error_type> consumer) {
 			return existing.start([handler = std::decay_t<OnNext>(handler)](
-				value_type value) {
-				handler(value);
-			}, [consumer](error_type error) {
-				consumer.put_error(error);
+				value_type &&value) {
+				handler(std::move(value));
+			}, [consumer](error_type &&error) {
+				consumer.put_error(std::move(error));
 			}, [consumer] {
 				consumer.put_done();
 			});
@@ -125,10 +125,10 @@ inline decltype(auto) bind_on_error(OnError &&handler) {
 			existing = std::move(existing),
 			handler = std::forward<OnError>(handler)
 		](consumer<value_type, no_error> consumer) {
-			return existing.start([consumer](value_type value) {
-				consumer.put_next(value);
-			}, [handler = std::decay_t<OnError>(handler)](error_type value) {
-				handler(value);
+			return existing.start([consumer](value_type &&value) {
+				consumer.put_next(std::move(value));
+			}, [handler = std::decay_t<OnError>(handler)](error_type &&error) {
+				handler(std::move(error));
 			}, [consumer] {
 				consumer.put_done();
 			});
@@ -145,10 +145,10 @@ inline decltype(auto) bind_on_done(OnDone &&handler) {
 			existing = std::move(existing),
 			handler = std::forward<OnDone>(handler)
 		](consumer<value_type, error_type> consumer) {
-			return existing.start([consumer](value_type value) {
-				consumer.put_next(value);
-			}, [consumer](error_type value) {
-				consumer.put_error(value);
+			return existing.start([consumer](value_type &&value) {
+				consumer.put_next(std::move(value));
+			}, [consumer](error_type &&value) {
+				consumer.put_error(std::move(value));
 			}, [handler = std::decay_t<OnDone>(handler)] {
 				handler();
 			});
@@ -511,8 +511,8 @@ inline void operator|(
 		producer<Value, Error> &&producer,
 		lifetime_holder &&start_with_lifetime) {
 	return std::move(producer)
-		| on_next([](Value) {})
-		| on_error([](Error) {})
+		| on_next([](Value&&) {})
+		| on_error([](Error&&) {})
 		| on_done([] {})
 		| std::move(start_with_lifetime);
 }
@@ -522,7 +522,7 @@ inline void operator|(
 		producer_with_next<Value, Error, OnNext> &&producer_with_next,
 		lifetime_holder &&start_with_lifetime) {
 	return std::move(producer_with_next)
-		| on_error([](Error) {})
+		| on_error([](Error&&) {})
 		| on_done([] {})
 		| std::move(start_with_lifetime);
 }
@@ -532,7 +532,7 @@ inline void operator|(
 		producer_with_error<Value, Error, OnError> &&producer_with_error,
 		lifetime_holder &&start_with_lifetime) {
 	return std::move(producer_with_error)
-		| on_next([](Value) {})
+		| on_next([](Value&&) {})
 		| on_done([] {})
 		| std::move(start_with_lifetime);
 }
@@ -542,8 +542,8 @@ inline void operator|(
 		producer_with_done<Value, Error, OnDone> &&producer_with_done,
 		lifetime_holder &&start_with_lifetime) {
 	return std::move(producer_with_done)
-		| on_next([](Value) {})
-		| on_error([](Error) {})
+		| on_next([](Value&&) {})
+		| on_error([](Error&&) {})
 		| std::move(start_with_lifetime);
 }
 
@@ -569,7 +569,7 @@ inline void operator|(
 			OnDone> &&producer_with_next_done,
 		lifetime_holder &&start_with_lifetime) {
 	return std::move(producer_with_next_done)
-		| on_error([](Error) {})
+		| on_error([](Error&&) {})
 		| std::move(start_with_lifetime);
 }
 
@@ -582,7 +582,7 @@ inline void operator|(
 			OnDone> &&producer_with_error_done,
 		lifetime_holder &&start_with_lifetime) {
 	return std::move(producer_with_error_done)
-		| on_next([](Value) {})
+		| on_next([](Value&&) {})
 		| std::move(start_with_lifetime);
 }
 

Telegram/SourceFiles/rpl/producer_tests.cpp

@@ -136,12 +136,12 @@ TEST_CASE("basic producer tests", "[rpl::producer]") {
 					++*lifetimeEndCount;
 				});
 				result.add(inner.start([=](int value) {
-					consumer.put_next(value);
+					consumer.put_next_copy(value);
 				}, [=](no_error) {
 				}, [=] {
 				}));
 				result.add(inner.start([=](int value) {
-					consumer.put_next(value);
+					consumer.put_next_copy(value);
 				}, [=](no_error) {
 				}, [=] {
 				}));

Telegram/gyp/telegram_sources.txt

@@ -6,6 +6,7 @@
 <(src_loc)/base/flat_set.h
 <(src_loc)/base/lambda.h
 <(src_loc)/base/lambda_guard.h
+<(src_loc)/base/index_based_iterator.h
 <(src_loc)/base/observer.cpp
 <(src_loc)/base/observer.h
 <(src_loc)/base/ordered_set.h