Allow not type-erased producers.

Telegram/SourceFiles/info/profile/info_profile_cover.cpp

@@ -185,11 +185,12 @@ SectionWithToggle *SectionWithToggle::setToggleShown(
 }
 
 rpl::producer<bool> SectionWithToggle::toggledValue() const {
-	return _toggle
+	if (_toggle) {
-		? (rpl::single(_toggle->checked())
+		return rpl::single(_toggle->checked())
 			| rpl::then(
-				base::ObservableViewer(_toggle->checkedChanged)))
+				base::ObservableViewer(_toggle->checkedChanged));
-		: rpl::never<bool>();
+	}
+	return rpl::never<bool>();
 }
 
 rpl::producer<bool> SectionWithToggle::toggleShownValue() const {

Telegram/SourceFiles/info/profile/info_profile_inner_widget.cpp

@@ -432,7 +432,7 @@ object_ptr<Ui::RpWidget> InnerWidget::setupUserActions(
 			st::infoBlockButtonSkip));
 
 		auto text = PeerUpdateValue(user, Notify::PeerUpdate::Flag::UserIsBlocked)
-			| rpl::map([user](auto&&) {
+			| rpl::map([user](auto&&) -> rpl::producer<QString> {
 				switch (user->blockStatus()) {
 				case UserData::BlockStatus::Blocked:
 					return Lang::Viewer(lng_profile_unblock_user);

Telegram/SourceFiles/rpl/after_next.h

@@ -33,10 +33,9 @@ public:
 		: _method(std::forward<OtherSideEffect>(method)) {
 	}
 
-	template <typename Value, typename Error>
+	template <typename Value, typename Error, typename Generator>
-	rpl::producer<Value, Error> operator()(
+	auto operator()(producer<Value, Error, Generator> &&initial) {
-			rpl::producer<Value, Error> &&initial) {
+		return make_producer<Value, Error>([
-		return [
 			initial = std::move(initial),
 			method = std::move(_method)
 		](const consumer<Value, Error> &consumer) mutable {
@@ -52,7 +51,7 @@ public:
 			}, [consumer] {
 				consumer.put_done();
 			});
-		};
+		});
 	}
 
 private:

Telegram/SourceFiles/rpl/combine.h

@@ -58,8 +58,8 @@ public:
 	, _state(state) {
 	}
 
-	template <typename Value, typename Error>
+	template <typename Value, typename Error, typename Generator>
-	void subscribe(producer<Value, Error> &&producer) {
+	void subscribe(producer<Value, Error, Generator> &&producer) {
 		_consumer.add_lifetime(std::move(producer).start(
 			[consumer = _consumer, state = _state](Value &&value) {
 			if (!state->accumulated) {
@@ -101,12 +101,13 @@ template <
 	typename consumer_type,
 	typename ...Values,
 	typename ...Errors,
+	typename ...Generators,
 	std::size_t ...I>
 inline void combine_subscribe(
 		const consumer_type &consumer,
 		combine_state<Values...> *state,
 		std::index_sequence<I...>,
-		producer<Values, Errors> &&...producers) {
+		producer<Values, Errors, Generators> &&...producers) {
 	auto consume = { (
 		details::combine_subscribe_one<
 			I,
@@ -119,17 +120,18 @@ inline void combine_subscribe(
 	(void)consume;
 }
 
-template <typename ...Values, typename ...Errors>
+template <
+	typename ...Values,
+	typename ...Errors,
+	typename ...Generators>
 inline auto combine_implementation(
-		producer<Values, Errors> &&...producers) {
+		producer<Values, Errors, Generators> &&...producers) {
+	using CombinedValue = std::tuple<Values...>;
 	using CombinedError = details::normalized_variant_t<Errors...>;
-	using Result = producer<
+	using consumer_type = consumer<CombinedValue, CombinedError>;
-		std::tuple<Values...>,
-		CombinedError>;
-	using consumer_type = typename Result::consumer_type;
 	auto result = [](
 			const consumer_type &consumer,
-			producer<Values, Errors> &...producers) {
+			producer<Values, Errors, Generators> &...producers) {
 		auto state = consumer.template make_state<
 			details::combine_state<Values...>>();
 
@@ -142,7 +144,7 @@ inline auto combine_implementation(
 
 		return lifetime();
 	};
-	return Result(std::bind(
+	return make_producer<CombinedValue, CombinedError>(std::bind(
 		result,
 		std::placeholders::_1,
 		std::move(producers)...));
@@ -156,8 +158,12 @@ template <typename ...Args>
 constexpr bool combine_just_producers_v
 	= combine_just_producers<Args...>::value;
 
-template <typename ...Values, typename ...Errors>
+template <
-struct combine_just_producers<producer<Values, Errors>...>
+	typename ...Values,
+	typename ...Errors,
+	typename ...Generators>
+struct combine_just_producers<
+		producer<Values, Errors, Generators>...>
 	: std::true_type {
 };
 
@@ -173,8 +179,11 @@ template <typename ...Args>
 using combine_result_type_t
 	= typename combine_result_type<Args...>::type;
 
-template <typename ...Values, typename ...Errors>
+template <
-struct combine_result_type<producer<Values, Errors>...> {
+	typename ...Values,
+	typename ...Errors,
+	typename ...Generators>
+struct combine_result_type<producer<Values, Errors, Generators>...> {
 	using type = std::tuple<Values...>;
 };
 
@@ -222,10 +231,13 @@ template <typename ...Args>
 constexpr bool combine_producers_with_mapper_v
 	 = combine_producers_with_mapper<Args...>::value;
 
-template <typename ...Values, typename ...Errors>
+template <
+	typename ...Values,
+	typename ...Errors,
+	typename ...Generators>
 inline decltype(auto) combine_helper(
 		std::true_type,
-		producer<Values, Errors> &&...producers) {
+		producer<Values, Errors, Generators> &&...producers) {
 	return combine_implementation(std::move(producers)...);
 }
 
@@ -274,17 +286,14 @@ struct combine_vector_state {
 
 } // namespace details
 
-template <typename Value, typename Error>
+template <typename Value, typename Error, typename Generator>
-inline producer<std::vector<Value>, Error> combine(
+inline auto combine(
-		std::vector<producer<Value, Error>> &&producers) {
+		std::vector<producer<Value, Error, Generator>> &&producers) {
-	if (producers.empty()) {
-		return complete<std::vector<Value>, Error>();
-	}
-
 	using state_type = details::combine_vector_state<Value>;
 	using consumer_type = consumer<std::vector<Value>, Error>;
-	return [producers = std::move(producers)](
+	return make_producer<std::vector<Value>, Error>([
-			const consumer_type &consumer) mutable {
+		producers = std::move(producers)
+	](const consumer_type &consumer) mutable {
 		auto count = producers.size();
 		auto state = consumer.template make_state<state_type>();
 		state->accumulated.resize(count);
@@ -320,13 +329,20 @@ inline producer<std::vector<Value>, Error> combine(
 				}
 			}));
 		}
+		if (!count) {
+			consumer.put_done();
+		}
 		return lifetime();
-	};
+	});
 }
 
-template <typename Value, typename Error, typename Mapper>
+template <
+	typename Value,
+	typename Error,
+	typename Generator,
+	typename Mapper>
 inline auto combine(
-		std::vector<producer<Value, Error>> &&producers,
+		std::vector<producer<Value, Error, Generator>> &&producers,
 		Mapper &&mapper) {
 	return combine(std::move(producers))
 		| map(std::forward<Mapper>(mapper));

Telegram/SourceFiles/rpl/combine_previous.h

@@ -28,14 +28,15 @@ namespace details {
 
 class combine_previous_helper {
 public:
-	template <typename Value, typename Error>
+	template <typename Value, typename Error, typename Generator>
-	rpl::producer<std::tuple<Value, Value>, Error> operator()(
+	auto operator()(
-			rpl::producer<Value, Error> &&initial) const {
+			producer<Value, Error, Generator> &&initial) const {
 		using consumer_type = consumer<
 			std::tuple<Value, Value>,
 			Error>;
-		return [initial = std::move(initial)](
+		return make_producer<std::tuple<Value, Value>, Error>([
-				const consumer_type &consumer) mutable {
+			initial = std::move(initial)
+		](const consumer_type &consumer) mutable {
 			auto previous = consumer.template make_state<
 				base::optional<Value>
 			>();
@@ -58,7 +59,7 @@ public:
 				}, [consumer] {
 					consumer.put_done();
 				});
-		};
+		});
 	}
 
 };
@@ -71,13 +72,12 @@ public:
 		: _value(std::forward<OtherValue>(value)) {
 	}
 
-	template <typename Value, typename Error>
+	template <typename Value, typename Error, typename Generator>
-	rpl::producer<std::tuple<Value, Value>, Error> operator()(
+	auto operator()(producer<Value, Error, Generator> &&initial) {
-			rpl::producer<Value, Error> &&initial) {
 		using consumer_type = consumer<
 			std::tuple<Value, Value>,
 			Error>;
-		return [
+		return make_producer<std::tuple<Value, Value>, Error>([
 			initial = std::move(initial),
 			value = Value(std::move(_value))
 		](const consumer_type &consumer) mutable {
@@ -96,7 +96,7 @@ public:
 				}, [consumer] {
 					consumer.put_done();
 				});
-		};
+		});
 	}
 
 private:

Telegram/SourceFiles/rpl/complete.h

@@ -25,11 +25,12 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 namespace rpl {
 
 template <typename Value = empty_value, typename Error = no_error>
-inline producer<Value, Error> complete() {
+inline auto complete() {
-	return [](const consumer<Value, Error> &consumer) mutable {
+	return make_producer<Value, Error>([](
+			const consumer<Value, Error> &consumer) {
 		consumer.put_done();
 		return lifetime();
-	};
+	});
 }
 
 } // namespace rpl

Telegram/SourceFiles/rpl/deferred.h

@@ -28,11 +28,12 @@ template <
 	typename Creator,
 	typename Value = typename decltype(std::declval<Creator>()())::value_type,
 	typename Error = typename decltype(std::declval<Creator>()())::error_type>
-inline producer<Value, Error> deferred(Creator &&creator) {
+inline auto deferred(Creator &&creator) {
-	return [creator = std::forward<Creator>(creator)](
+	return make_producer<Value, Error>([
-			const consumer<Value, Error> &consumer) mutable {
+		creator = std::forward<Creator>(creator)
+	](const consumer<Value, Error> &consumer) mutable {
 		return std::move(creator)().start_existing(consumer);
-	};
+	});
 }
 
 } // namespace rpl

Telegram/SourceFiles/rpl/details/superset_type.h

@@ -0,0 +1,36 @@
+/*
+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
+
+namespace rpl {
+
+template <typename Value1, typename Value2>
+struct superset_type;
+
+template <typename Value1, typename Value2>
+using superset_type_t = typename superset_type<Value1, Value2>::type;
+
+template <typename Value>
+struct superset_type<Value, Value> {
+	using type = Value;
+};
+
+} // namespace rpl

Telegram/SourceFiles/rpl/distinct_until_changed.h

@@ -28,11 +28,12 @@ namespace details {
 
 class distinct_until_changed_helper {
 public:
-	template <typename Value, typename Error>
+	template <typename Value, typename Error, typename Generator>
-	rpl::producer<Value, Error> operator()(
+	auto operator()(
-			rpl::producer<Value, Error> &&initial) const {
+			producer<Value, Error, Generator> &&initial) const {
-		return [initial = std::move(initial)](
+		return make_producer<Value, Error>([
-				const consumer<Value, Error> &consumer) mutable {
+			initial = std::move(initial)
+		](const consumer<Value, Error> &consumer) mutable {
 			auto previous = consumer.template make_state<
 				base::optional<Value>
 			>();
@@ -47,7 +48,7 @@ public:
 				}, [consumer] {
 					consumer.put_done();
 				});
-		};
+		});
 	}
 
 };

Telegram/SourceFiles/rpl/event_stream.h

@@ -46,13 +46,29 @@ public:
 	void fire_copy(const Value &value) const {
 		return fire_forward(value);
 	}
-	producer<Value, no_error> events() const;
+	auto events() const {
-	producer<Value, no_error> events_starting_with(
+		using consumer_type = consumer<Value, no_error>;
-			Value &&value) const {
+		return make_producer<Value>([weak = weak()](
+			const consumer_type &consumer) {
+			if (auto strong = weak.lock()) {
+				auto result = [weak, consumer] {
+					if (auto strong = weak.lock()) {
+						auto it = base::find(*strong, consumer);
+						if (it != strong->end()) {
+							it->terminate();
+						}
+					}
+				};
+				strong->push_back(std::move(consumer));
+				return lifetime(std::move(result));
+			}
+			return lifetime();
+		});
+	}
+	auto events_starting_with(Value &&value) const {
 		return single(std::move(value)) | then(events());
 	}
-	producer<Value, no_error> events_starting_with_copy(
+	auto events_starting_with_copy(const Value &value) const {
-			const Value &value) const {
 		return single(value) | then(events());
 	}
 
@@ -117,26 +133,6 @@ inline void event_stream<Value>::fire_forward(
 	}
 }
 
-template <typename Value>
-inline producer<Value, no_error> event_stream<Value>::events() const {
-	return producer<Value, no_error>([weak = weak()](
-			const consumer<Value, no_error> &consumer) {
-		if (auto strong = weak.lock()) {
-			auto result = [weak, consumer] {
-				if (auto strong = weak.lock()) {
-					auto it = base::find(*strong, consumer);
-					if (it != strong->end()) {
-						it->terminate();
-					}
-				}
-			};
-			strong->push_back(std::move(consumer));
-			return lifetime(std::move(result));
-		}
-		return lifetime();
-	});
-}
-
 template <typename Value>
 inline std::weak_ptr<std::vector<consumer<Value, no_error>>> event_stream<Value>::weak() const {
 	if (!_consumers) {
@@ -171,9 +167,9 @@ public:
 	start_spawning_helper(lifetime &alive_while)
 	: _lifetime(alive_while) {
 	}
-	template <typename Value, typename Error>
+
-	producer<Value, Error> operator()(
+	template <typename Value, typename Error, typename Generator>
-			producer<Value, Error> &&initial) {
+	auto operator()(producer<Value, Error, Generator> &&initial) {
 		auto stream = _lifetime.make_state<event_stream<Value>>();
 		auto collected = std::vector<Value>();
 		{
@@ -185,9 +181,7 @@ public:
 				[] {});
 			std::move(initial) | start_to_stream(*stream, _lifetime);
 		}
-		return collected.empty()
+		return vector(std::move(collected))
-			? stream->events()
-			: vector(std::move(collected))
 				| then(stream->events());
 	}
 

Telegram/SourceFiles/rpl/fail.h

@@ -25,13 +25,14 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 namespace rpl {
 
 template <typename Value, typename Error>
-inline producer<Value, std::decay_t<Error>> fail(Error &&error) {
+inline auto fail(Error &&error) {
 	using consumer_t = consumer<Value, std::decay_t<Error>>;
-	return [error = std::forward<Error>(error)](
+	return make_producer<Value, std::decay_t<Error>>([
-			const consumer_t &consumer) mutable {
+		error = std::forward<Error>(error)
+	](const consumer_t &consumer) mutable {
 		consumer.put_error(std::move(error));
 		return lifetime();
-	};
+	});
 }
 
 } // namespace rpl

Telegram/SourceFiles/rpl/filter.h

@@ -22,6 +22,7 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 
 #include <rpl/producer.h>
 #include <rpl/combine.h>
+#include <rpl/mappers.h>
 #include "base/optional.h"
 
 namespace rpl {
@@ -38,15 +39,15 @@ public:
 	template <
 		typename Value,
 		typename Error,
+		typename Generator,
 		typename = std::enable_if_t<
 			details::is_callable_v<Predicate, Value>>>
-	rpl::producer<Value, Error> operator()(
+	auto operator()(producer<Value, Error, Generator> &&initial) {
-			rpl::producer<Value, Error> &&initial) {
+		using consumer_type = consumer<Value, Error>;
-		return [
+		return make_producer<Value, Error>([
-				initial = std::move(initial),
+			initial = std::move(initial),
-				predicate = std::move(_predicate)
+			predicate = std::move(_predicate)
-			](
+		](const consumer_type &consumer) mutable {
-				const consumer<Value, Error> &consumer) mutable {
 			return std::move(initial).start(
 				[
 					consumer,
@@ -66,7 +67,7 @@ public:
 				}, [consumer] {
 					consumer.put_done();
 				});
-		};
+		});
 	}
 
 private:
@@ -85,27 +86,24 @@ inline auto filter(Predicate &&predicate)
 
 namespace details {
 
-template <>
+template <typename FilterError, typename FilterGenerator>
-class filter_helper<producer<bool>> {
+class filter_helper<producer<bool, FilterError, FilterGenerator>> {
 public:
-	filter_helper(producer<bool> &&filterer)
+	filter_helper(
-		: _filterer(std::move(filterer)) {
+		producer<bool, FilterError, FilterGenerator> &&filterer)
+	: _filterer(std::move(filterer)) {
 	}
 
-	template <
+	template <typename Value, typename Error, typename Generator>
-		typename Value,
+	auto operator()(producer<Value, Error, Generator> &&initial) {
-		typename Error>
+		using namespace mappers;
-	rpl::producer<Value, Error> operator()(
-			rpl::producer<Value, Error> &&initial) {
 		return combine(std::move(initial), std::move(_filterer))
-			| filter([](auto &&value, bool let) { return let; })
+			| filter($2)
-			| map([](auto &&value, bool) {
+			| map($1_of_two);
-				return std::forward<decltype(value)>(value);
-			});
 	}
 
 private:
-	producer<bool> _filterer;
+	producer<bool, FilterError, FilterGenerator> _filterer;
 
 };
 
@@ -123,12 +121,15 @@ inline Value &&deref_optional_helper(
 
 class filter_optional_helper {
 public:
-	template <typename Value, typename Error>
+	template <typename Value, typename Error, typename Generator>
-	rpl::producer<Value, Error> operator()(
+	auto operator()(producer<
-		rpl::producer<base::optional<Value>, Error> &&initial
+			base::optional<Value>,
-	) const {
+			Error,
-		return [initial = std::move(initial)](
+			Generator> &&initial) const {
-				const consumer<Value, Error> &consumer) mutable {
+		using consumer_type = consumer<Value, Error>;
+		return make_producer<Value, Error>([
+			initial = std::move(initial)
+		](const consumer_type &consumer) mutable {
 			return std::move(initial).start(
 				[consumer](auto &&value) {
 					if (value) {
@@ -143,7 +144,7 @@ public:
 				}, [consumer] {
 					consumer.put_done();
 				});
-		};
+		});
 	}
 
 };

Telegram/SourceFiles/rpl/flatten_latest.h

@@ -27,17 +27,22 @@ namespace details {
 
 class flatten_latest_helper {
 public:
-	template <typename Value, typename Error>
+	template <
-	rpl::producer<Value, Error> operator()(
+		typename Value,
-			rpl::producer<
+		typename Error,
-				rpl::producer<Value, Error>,
+		typename Generator,
-				Error
+		typename MetaGenerator>
-			> &&initial) const {
+	auto operator()(producer<
-		return [initial = std::move(initial)](
+			producer<Value, Error, Generator>,
-				const consumer<Value, Error> &consumer) mutable {
+			Error,
+			MetaGenerator> &&initial) const {
+		using consumer_type = consumer<Value, Error>;
+		return make_producer<Value, Error>([
+			initial = std::move(initial)
+		](const consumer_type &consumer) mutable {
 			auto state = std::make_shared<State>();
 			return std::move(initial).start(
-			[consumer, state](rpl::producer<Value, Error> &&inner) {
+			[consumer, state](producer<Value, Error> &&inner) {
 				state->finished = false;
 				state->alive = std::move(inner).start(
 				[consumer](auto &&value) {
@@ -60,7 +65,7 @@ public:
 					state->finished = true;
 				}
 			});
-		};
+		});
 	}
 
 private:

Telegram/SourceFiles/rpl/map.h

@@ -84,12 +84,12 @@ public:
 	template <
 		typename Value,
 		typename Error,
+		typename Generator,
 		typename NewValue = details::callable_result<
 			Transform,
 			Value>>
-	rpl::producer<NewValue, Error> operator()(
+	auto operator()(producer<Value, Error, Generator> &&initial) {
-			rpl::producer<Value, Error> &&initial) {
+		return make_producer<NewValue, Error>([
-		return [
 			initial = std::move(initial),
 			transform = std::move(_transform)
 		](const consumer<NewValue, Error> &consumer) mutable {
@@ -103,7 +103,7 @@ public:
 			}, [consumer] {
 				consumer.put_done();
 			});
-		};
+		});
 	}
 
 private:
@@ -181,12 +181,12 @@ public:
 	template <
 		typename Value,
 		typename Error,
+		typename Generator,
 		typename NewError = details::callable_result<
 			Transform,
 			Error>>
-	rpl::producer<Value, NewError> operator()(
+	auto operator()(producer<Value, Error, Generator> &&initial) {
-			rpl::producer<Value, Error> &&initial) {
+		return make_producer<Value, NewError>([
-		return [
 			initial = std::move(initial),
 			transform = std::move(_transform)
 		](const consumer<Value, NewError> &consumer) mutable {
@@ -200,7 +200,7 @@ public:
 			), [consumer] {
 				consumer.put_done();
 			});
-		};
+		});
 	}
 
 private:

Telegram/SourceFiles/rpl/mappers.h

@@ -501,5 +501,7 @@ inline auto $val(Type &&value) {
 	return details::make_value_mapper(std::forward<Type>(value));
 }
 
+constexpr const auto $1_of_two = ((void)$2, $1);
+
 } // namespace mappers
 } // namespace rpl

Telegram/SourceFiles/rpl/never.h

@@ -25,10 +25,12 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 namespace rpl {
 
 template <typename Value = empty_value, typename Error = no_error>
-inline producer<Value, Error> never() {
+inline auto never() {
-	return [](const consumer<Value, Error> &consumer) mutable {
+	using consumer_type = consumer<Value, Error>;
+	return make_producer<Value, Error>([](
+			const consumer_type &consumer) {
 		return lifetime();
-	};
+	});
 }
 
 } // namespace rpl

Telegram/SourceFiles/rpl/operators_tests.cpp

@@ -123,7 +123,7 @@ TEST_CASE("basic operators tests", "[rpl::operators]") {
 		auto copyCount = std::make_shared<int>(0);
 		auto moveCount = std::make_shared<int>(0);
 		{
-			auto testing = complete<InvokeCounter>();
+			auto testing = complete<InvokeCounter>() | type_erased();
 			for (auto i = 0; i != 5; ++i) {
 				InvokeCounter counter(copyCount, moveCount);
 				testing = std::move(testing)
@@ -248,8 +248,18 @@ TEST_CASE("basic operators tests", "[rpl::operators]") {
 				| start_with_next([=](std::string &&value) {
 					*sum += std::move(value) + ' ';
 				}, lifetime);
+			single(single(1))
+				| then(single(single(2) | then(single(3))))
+				| then(single(single(4) | then(single(5)) | then(single(6))))
+				| flatten_latest()
+				| map([](int value) {
+					return std::to_string(value);
+				})
+				| start_with_next([=](std::string &&value) {
+					*sum += std::move(value) + ' ';
+				}, lifetime);
 		}
-		REQUIRE(*sum == "1 2 3 4 5 6 ");
+		REQUIRE(*sum == "1 2 3 4 5 6 1 2 3 4 5 6 ");
 	}
 
 	SECTION("combine vector test") {
@@ -363,4 +373,19 @@ TEST_CASE("basic operators tests", "[rpl::operators]") {
 		}
 		REQUIRE(*sum == "16192225");
 	}
+
+	SECTION("after_next test") {
+		auto sum = std::make_shared<std::string>("");
+		{
+			rpl::lifetime lifetime;
+			rpl::ints(3)
+				| after_next([=](int value) {
+					*sum += std::to_string(-value-1);
+				})
+				| start_with_next([=](int value) {
+					*sum += std::to_string(value);
+				}, lifetime);
+		}
+		REQUIRE(*sum == "0-11-22-3");
+	}
 }

Telegram/SourceFiles/rpl/producer.h

@@ -23,16 +23,27 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #include "base/lambda.h"
 #include <rpl/consumer.h>
 #include <rpl/lifetime.h>
+#include <rpl/details/superset_type.h>
+#include <rpl/details/callable.h>
 
 namespace rpl {
 namespace details {
 
+template <typename Value, typename Error>
+const consumer<Value, Error> &const_ref_consumer();
+
 template <typename Lambda>
 class mutable_lambda_wrap {
 public:
 	mutable_lambda_wrap(Lambda &&lambda)
 		: _lambda(std::move(lambda)) {
 	}
+	mutable_lambda_wrap(const mutable_lambda_wrap &other) = default;
+	mutable_lambda_wrap(mutable_lambda_wrap &&other) = default;
+	mutable_lambda_wrap &operator=(
+		const mutable_lambda_wrap &other) = default;
+	mutable_lambda_wrap &operator=(
+		mutable_lambda_wrap &&other) = default;
 
 	template <typename... Args>
 	auto operator()(Args&&... args) const {
@@ -46,62 +57,134 @@ private:
 };
 
 // Type-erased copyable mutable lambda using base::lambda.
-template <typename Function> class mutable_lambda;
+template <typename Value, typename Error>
-
+class type_erased_generator final {
-template <typename Return, typename ...Args>
-class mutable_lambda<Return(Args...)> {
 public:
+	using value_type = Value;
+	using error_type = Error;
+	using consumer_type = consumer<Value, Error>;
+
+	type_erased_generator(
+		const type_erased_generator &other) = default;
+	type_erased_generator(
+		type_erased_generator &&other) = default;
+	type_erased_generator &operator=(
+		const type_erased_generator &other) = default;
+	type_erased_generator &operator=(
+		type_erased_generator &&other) = default;
 
-	// Copy / move construct / assign from an arbitrary type.
 	template <
-		typename Lambda,
+		typename Generator,
-		typename = std::enable_if_t<std::is_convertible<
+		typename = std::enable_if_t<
-			decltype(std::declval<Lambda>()(
+			std::is_convertible_v<
-				std::declval<Args>()...)),
+				decltype(std::declval<Generator>()(
-			Return
+					const_ref_consumer<Value, Error>())),
-		>::value>>
+				lifetime> &&
-	mutable_lambda(Lambda other) : _implementation(
+			!std::is_same_v<
-		mutable_lambda_wrap<Lambda>(std::move(other))) {
+				std::decay_t<Generator>,
+				type_erased_generator>>>
+	type_erased_generator(Generator other) : _implementation(
+		mutable_lambda_wrap<Generator>(std::move(other))) {
+	}
+	template <
+		typename Generator,
+		typename = std::enable_if_t<
+			std::is_convertible_v<
+				decltype(std::declval<Generator>()(
+					const_ref_consumer<Value, Error>())),
+				lifetime> &&
+			!std::is_same_v<
+				std::decay_t<Generator>,
+				type_erased_generator>>>
+	type_erased_generator &operator=(Generator other) {
+		_implementation = mutable_lambda_wrap<Generator>(
+			std::move(other));
+		return *this;
 	}
 
-	template <
+	lifetime operator()(const consumer_type &consumer) {
-		typename ...OtherArgs,
+		return _implementation(consumer);
-		typename = std::enable_if_t<
-			(sizeof...(Args) == sizeof...(OtherArgs))>>
-	Return operator()(OtherArgs&&... args) {
-		return _implementation(std::forward<OtherArgs>(args)...);
 	}
 
 private:
-	base::lambda<Return(Args...)> _implementation;
+	base::lambda<lifetime(const consumer_type &)> _implementation;
 
 };
 
 } // namespace details
 
-template <typename Value = empty_value, typename Error = no_error>
+template <
-class producer {
+	typename Value = empty_value,
+	typename Error = no_error,
+	typename Generator = details::type_erased_generator<
+		Value,
+		Error>>
+class producer;
+
+template <
+	typename Value1,
+	typename Value2,
+	typename Error1,
+	typename Error2,
+	typename Generator>
+struct superset_type<
+		producer<Value1, Error1, Generator>,
+		producer<Value2, Error2, Generator>> {
+	using type = producer<
+		superset_type_t<Value1, Value2>,
+		superset_type_t<Error1, Error2>,
+		Generator>;
+};
+
+template <
+	typename Value,
+	typename Error,
+	typename Generator1,
+	typename Generator2>
+struct superset_type<
+		producer<Value, Error, Generator1>,
+		producer<Value, Error, Generator2>> {
+	using type = producer<Value, Error>;
+};
+
+template <
+	typename Value,
+	typename Error,
+	typename Generator>
+struct superset_type<
+		producer<Value, Error, Generator>,
+		producer<Value, Error, Generator>> {
+	using type = producer<Value, Error, Generator>;
+};
+
+namespace details {
+
+template <typename Value, typename Error, typename Generator>
+class producer_base {
 public:
 	using value_type = Value;
 	using error_type = Error;
 	using consumer_type = consumer<Value, Error>;
 
 	template <
-		typename Generator,
+		typename OtherGenerator,
-		typename = std::enable_if<std::is_convertible<
+		typename = std::enable_if_t<
-			decltype(std::declval<Generator>()(
+			std::is_constructible_v<Generator, OtherGenerator&&>>>
-				std::declval<consumer_type>())),
+	producer_base(OtherGenerator &&generator);
-			lifetime>::value>>
+
-	producer(Generator &&generator);
+	producer_base(const producer_base &other) = default;
+	producer_base(producer_base &&other) = default;
+	producer_base &operator=(const producer_base &other) = default;
+	producer_base &operator=(producer_base &&other) = default;
 
 	template <
 		typename OnNext,
 		typename OnError,
 		typename OnDone,
 		typename = std::enable_if_t<
-			details::is_callable_v<OnNext, Value>
+			is_callable_v<OnNext, Value>
-			&& details::is_callable_v<OnError, Error>
+			&& is_callable_v<OnError, Error>
-			&& details::is_callable_v<OnDone>>>
+			&& is_callable_v<OnDone>>>
 	lifetime start(
 		OnNext &&next,
 		OnError &&error,
@@ -112,9 +195,9 @@ public:
 		typename OnError,
 		typename OnDone,
 		typename = std::enable_if_t<
-			details::is_callable_v<OnNext, Value>
+			is_callable_v<OnNext, Value>
-			&& details::is_callable_v<OnError, Error>
+			&& is_callable_v<OnError, Error>
-			&& details::is_callable_v<OnDone>>>
+			&& is_callable_v<OnDone>>>
 	lifetime start_copy(
 		OnNext &&next,
 		OnError &&error,
@@ -123,24 +206,30 @@ public:
 	lifetime start_existing(const consumer_type &consumer) &&;
 
 private:
-	details::mutable_lambda<
+	Generator _generator;
-		lifetime(const consumer_type &)> _generator;
+
+	template <
+		typename OtherValue,
+		typename OtherError,
+		typename OtherGenerator>
+	friend class ::rpl::producer;
 
 };
 
-template <typename Value, typename Error>
+template <typename Value, typename Error, typename Generator>
-template <typename Generator, typename>
+template <typename OtherGenerator, typename>
-inline producer<Value, Error>::producer(Generator &&generator)
+inline producer_base<Value, Error, Generator>::producer_base(
-: _generator(std::forward<Generator>(generator)) {
+	OtherGenerator &&generator)
+: _generator(std::forward<OtherGenerator>(generator)) {
 }
 
-template <typename Value, typename Error>
+template <typename Value, typename Error, typename Generator>
 template <
 	typename OnNext,
 	typename OnError,
 	typename OnDone,
 	typename>
-inline lifetime producer<Value, Error>::start(
+inline lifetime producer_base<Value, Error, Generator>::start(
 		OnNext &&next,
 		OnError &&error,
 		OnDone &&done) && {
@@ -150,13 +239,13 @@ inline lifetime producer<Value, Error>::start(
 		std::forward<OnDone>(done)));
 }
 
-template <typename Value, typename Error>
+template <typename Value, typename Error, typename Generator>
 template <
 	typename OnNext,
 	typename OnError,
 	typename OnDone,
 	typename>
-inline lifetime producer<Value, Error>::start_copy(
+inline lifetime producer_base<Value, Error, Generator>::start_copy(
 		OnNext &&next,
 		OnError &&error,
 		OnDone &&done) const & {
@@ -167,26 +256,122 @@ inline lifetime producer<Value, Error>::start_copy(
 		std::forward<OnDone>(done));
 }
 
-template <typename Value, typename Error>
+template <typename Value, typename Error, typename Generator>
-inline lifetime producer<Value, Error>::start_existing(
+inline lifetime producer_base<Value, Error, Generator>::start_existing(
 		const consumer_type &consumer) && {
 	consumer.add_lifetime(std::move(_generator)(consumer));
 	return [consumer] { consumer.terminate(); };
 }
 
 template <typename Value, typename Error>
-inline producer<Value, Error> duplicate(
+using producer_base_type_erased = producer_base<
-		const producer<Value, Error> &value) {
+	Value,
+	Error,
+	type_erased_generator<Value, Error>>;
+
+} // namespace details
+
+template <typename Value, typename Error, typename Generator>
+class producer final
+: public details::producer_base<Value, Error, Generator> {
+	using parent_type = details::producer_base<
+		Value,
+		Error,
+		Generator>;
+public:
+	using parent_type::parent_type;
+
+};
+
+template <typename Value, typename Error>
+class producer<
+	Value,
+	Error,
+	details::type_erased_generator<Value, Error>> final
+: public details::producer_base_type_erased<Value, Error> {
+	using parent_type = details::producer_base_type_erased<
+		Value,
+		Error>;
+public:
+	using parent_type::parent_type;;
+
+	producer(const producer &other) = default;
+	producer(producer &&other) = default;
+	producer &operator=(const producer &other) = default;
+	producer &operator=(producer &&other) = default;
+
+	template <
+		typename Generic,
+		typename = std::enable_if_t<!std::is_same_v<
+			Generic,
+			details::type_erased_generator<Value, Error>>>>
+	producer(const details::producer_base<Value, Error, Generic> &other)
+		: parent_type(other._generator) {
+	}
+
+	template <
+		typename Generic,
+		typename = std::enable_if_t<!std::is_same_v<
+			Generic,
+			details::type_erased_generator<Value, Error>>>>
+	producer(details::producer_base<Value, Error, Generic> &&other)
+		: parent_type(std::move(other._generator)) {
+	}
+
+	template <
+		typename Generic,
+		typename = std::enable_if_t<!std::is_same_v<
+			Generic,
+			details::type_erased_generator<Value, Error>>>>
+	producer &operator=(
+			const details::producer_base<Value, Error, Generic> &other) {
+		this->_generator = other._generator;
+		return *this;
+	}
+
+	template <
+		typename Generic,
+		typename = std::enable_if_t<!std::is_same_v<
+			Generic,
+			details::type_erased_generator<Value, Error>>>>
+	producer &operator=(
+			details::producer_base<Value, Error, Generic> &&other) {
+		this->_generator = std::move(other._generator);
+		return *this;
+	}
+
+};
+
+template <
+	typename Value = empty_value,
+	typename Error = no_error,
+	typename Generator,
+	typename = std::enable_if_t<
+		std::is_convertible_v<
+			decltype(std::declval<Generator>()(
+				details::const_ref_consumer<Value, Error>())),
+			lifetime>>>
+inline auto make_producer(Generator &&generator)
+-> producer<Value, Error, std::decay_t<Generator>> {
+	return std::forward<Generator>(generator);
+}
+
+template <typename Value, typename Error, typename Generator>
+inline producer<Value, Error, Generator> duplicate(
+		const producer<Value, Error, Generator> &value) {
 	return value;
 }
 
 template <
 	typename Value,
 	typename Error,
+	typename Generator,
 	typename Method,
 	typename = decltype(std::declval<Method>()(
-		std::declval<producer<Value, Error>>()))>
+		std::declval<producer<Value, Error, Generator>>()))>
-inline auto operator|(producer<Value, Error> &&value, Method &&method) {
+inline auto operator|(
+		producer<Value, Error, Generator> &&value,
+		Method &&method) {
 	return std::forward<Method>(method)(std::move(value));
 }
 
@@ -333,9 +518,9 @@ inline auto start_with_next_error_done(
 
 namespace details {
 
-template <typename Value, typename Error>
+template <typename Value, typename Error, typename Generator>
 inline void operator|(
-		producer<Value, Error> &&value,
+		producer<Value, Error, Generator> &&value,
 		lifetime_with_none &&lifetime) {
 	lifetime.alive_while.add(
 		std::move(value).start(
@@ -347,10 +532,11 @@ inline void operator|(
 template <
 	typename Value,
 	typename Error,
+	typename Generator,
 	typename OnNext,
 	typename = std::enable_if_t<is_callable_v<OnNext, Value>>>
 inline void operator|(
-		producer<Value, Error> &&value,
+		producer<Value, Error, Generator> &&value,
 		lifetime_with_next<OnNext> &&lifetime) {
 	lifetime.alive_while.add(
 		std::move(value).start(
@@ -362,10 +548,11 @@ inline void operator|(
 template <
 	typename Value,
 	typename Error,
+	typename Generator,
 	typename OnError,
 	typename = std::enable_if_t<is_callable_v<OnError, Error>>>
 inline void operator|(
-		producer<Value, Error> &&value,
+		producer<Value, Error, Generator> &&value,
 		lifetime_with_error<OnError> &&lifetime) {
 	lifetime.alive_while.add(
 		std::move(value).start(
@@ -377,10 +564,11 @@ inline void operator|(
 template <
 	typename Value,
 	typename Error,
+	typename Generator,
 	typename OnDone,
 	typename = std::enable_if_t<is_callable_v<OnDone>>>
 inline void operator|(
-		producer<Value, Error> &&value,
+		producer<Value, Error, Generator> &&value,
 		lifetime_with_done<OnDone> &&lifetime) {
 	lifetime.alive_while.add(
 		std::move(value).start(
@@ -392,13 +580,14 @@ inline void operator|(
 template <
 	typename Value,
 	typename Error,
+	typename Generator,
 	typename OnNext,
 	typename OnError,
 	typename = std::enable_if_t<
 		is_callable_v<OnNext, Value> &&
 		is_callable_v<OnError, Error>>>
 inline void operator|(
-		producer<Value, Error> &&value,
+		producer<Value, Error, Generator> &&value,
 		lifetime_with_next_error<OnNext, OnError> &&lifetime) {
 	lifetime.alive_while.add(
 		std::move(value).start(
@@ -410,13 +599,14 @@ inline void operator|(
 template <
 	typename Value,
 	typename Error,
+	typename Generator,
 	typename OnError,
 	typename OnDone,
 	typename = std::enable_if_t<
 		is_callable_v<OnError, Error> &&
 		is_callable_v<OnDone>>>
 inline void operator|(
-		producer<Value, Error> &&value,
+		producer<Value, Error, Generator> &&value,
 		lifetime_with_error_done<OnError, OnDone> &&lifetime) {
 	lifetime.alive_while.add(
 		std::move(value).start(
@@ -428,13 +618,14 @@ inline void operator|(
 template <
 	typename Value,
 	typename Error,
+	typename Generator,
 	typename OnNext,
 	typename OnDone,
 	typename = std::enable_if_t<
 		is_callable_v<OnNext, Value> &&
 		is_callable_v<OnDone>>>
 inline void operator|(
-		producer<Value, Error> &&value,
+		producer<Value, Error, Generator> &&value,
 		lifetime_with_next_done<OnNext, OnDone> &&lifetime) {
 	lifetime.alive_while.add(
 		std::move(value).start(
@@ -446,6 +637,7 @@ inline void operator|(
 template <
 	typename Value,
 	typename Error,
+	typename Generator,
 	typename OnNext,
 	typename OnError,
 	typename OnDone,
@@ -454,7 +646,7 @@ template <
 		is_callable_v<OnError, Error> &&
 		is_callable_v<OnDone>>>
 inline void operator|(
-		producer<Value, Error> &&value,
+		producer<Value, Error, Generator> &&value,
 		lifetime_with_next_error_done<
 			OnNext,
 			OnError,

Telegram/SourceFiles/rpl/producer_tests.cpp

@@ -52,7 +52,7 @@ TEST_CASE("basic producer tests", "[rpl::producer]") {
 				*destroyed = true;
 			});
 			{
-				producer<int, no_error>([=](auto &&consumer) {
+				make_producer<int>([=](auto &&consumer) {
 					(void)destroyCaller;
 					consumer.put_next(1);
 					consumer.put_next(2);
@@ -82,7 +82,7 @@ TEST_CASE("basic producer tests", "[rpl::producer]") {
 	SECTION("producer error test") {
 		auto errorGenerated = std::make_shared<bool>(false);
 		{
-			producer<no_value, bool>([=](auto &&consumer) {
+			make_producer<no_value, bool>([=](auto &&consumer) {
 				consumer.put_error(true);
 				return lifetime();
 			}).start([=](no_value) {
@@ -99,7 +99,7 @@ TEST_CASE("basic producer tests", "[rpl::producer]") {
 		{
 			auto lifetimes = lifetime();
 			{
-				auto testProducer = producer<no_value, no_error>([=](auto &&consumer) {
+				auto testProducer = make_producer<no_value>([=](auto &&consumer) {
 					return [=] {
 						++*lifetimeEndCount;
 					};
@@ -123,8 +123,8 @@ TEST_CASE("basic producer tests", "[rpl::producer]") {
 		auto lifetimeEndCount = std::make_shared<int>(0);
 		auto saved = lifetime();
 		{
-			saved = producer<int, no_error>([=](auto &&consumer) {
+			saved = make_producer<int>([=](auto &&consumer) {
-				auto inner = producer<int, no_error>([=](auto &&consumer) {
+				auto inner = make_producer<int>([=](auto &&consumer) {
 					consumer.put_next(1);
 					consumer.put_next(2);
 					consumer.put_next(3);
@@ -161,7 +161,8 @@ TEST_CASE("basic producer tests", "[rpl::producer]") {
 	SECTION("tuple producer test") {
 		auto result = std::make_shared<int>(0);
 		{
-			producer<std::tuple<int, double>>([=](auto &&consumer) {
+			make_producer<std::tuple<int, double>>([=](
+					auto &&consumer) {
 				consumer.put_next(std::make_tuple(1, 2.));
 				return lifetime();
 			}).start([=](int a, double b) {
@@ -345,7 +346,7 @@ TEST_CASE("basic piping tests", "[rpl::producer]") {
 		auto dones = std::make_shared<int>(0);
 		{
 			auto alive = lifetime();
-			producer<int, int>([=](auto &&consumer) {
+			make_producer<int, int>([=](auto &&consumer) {
 				consumer.put_next(1);
 				consumer.put_done();
 				return lifetime();
@@ -353,7 +354,7 @@ TEST_CASE("basic piping tests", "[rpl::producer]") {
 				*sum += value;
 			}, alive);
 
-			producer<int, int>([=](auto &&consumer) {
+			make_producer<int, int>([=](auto &&consumer) {
 				consumer.put_next(11);
 				consumer.put_error(111);
 				return lifetime();
@@ -361,7 +362,7 @@ TEST_CASE("basic piping tests", "[rpl::producer]") {
 				*sum += value;
 			}, alive);
 
-			producer<int, int>([=](auto &&consumer) {
+			make_producer<int, int>([=](auto &&consumer) {
 				consumer.put_next(1111);
 				consumer.put_done();
 				return lifetime();
@@ -369,7 +370,7 @@ TEST_CASE("basic piping tests", "[rpl::producer]") {
 				*dones += 1;
 			}, alive);
 
-			producer<int, int>([=](auto &&consumer) {
+			make_producer<int, int>([=](auto &&consumer) {
 				consumer.put_next(11111);
 				consumer.put_next(11112);
 				consumer.put_next(11113);
@@ -383,7 +384,7 @@ TEST_CASE("basic piping tests", "[rpl::producer]") {
 		}
 
 		auto alive = lifetime();
-		producer<int, int>([=](auto &&consumer) {
+		make_producer<int, int>([=](auto &&consumer) {
 			consumer.put_next(111111);
 			consumer.put_next(111112);
 			consumer.put_next(111113);
@@ -395,7 +396,7 @@ TEST_CASE("basic piping tests", "[rpl::producer]") {
 			*dones += 11;
 		}, alive);
 
-		producer<int, int>([=](auto &&consumer) {
+		make_producer<int, int>([=](auto &&consumer) {
 			consumer.put_error(1111111);
 			return lifetime();
 		}) | start_with_error_done([=](int value) {
@@ -404,7 +405,7 @@ TEST_CASE("basic piping tests", "[rpl::producer]") {
 			*dones = 0;
 		}, alive);
 
-		producer<int, int>([=](auto &&consumer) {
+		make_producer<int, int>([=](auto &&consumer) {
 			consumer.put_next(11111111);
 			consumer.put_next(11111112);
 			consumer.put_next(11111113);
@@ -438,7 +439,7 @@ TEST_CASE("basic piping tests", "[rpl::producer]") {
 
 			for (int i = 0; i != 3; ++i) {
 				auto alive = lifetime();
-				producer<int, int>([=](auto &&consumer) {
+				make_producer<int, int>([=](auto &&consumer) {
 					consumer.put_next(1);
 					consumer.put_done();
 					return lifetime();

Telegram/SourceFiles/rpl/range.h

@@ -24,69 +24,78 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 
 namespace rpl {
 
-template <typename Value, typename Error = no_error>
+template <typename Value>
-inline producer<std::decay_t<Value>, Error> single(Value &&value) {
+inline auto single(Value &&value) {
-	using consumer_t = consumer<std::decay_t<Value>, Error>;
+	using consumer_type = consumer<std::decay_t<Value>, no_error>;
-	return [value = std::forward<Value>(value)](
+	return make_producer<std::decay_t<Value>>([
-			const consumer_t &consumer) mutable {
+		value = std::forward<Value>(value)
+	](const consumer_type &consumer) mutable {
 		consumer.put_next(std::move(value));
 		consumer.put_done();
 		return lifetime();
-	};
+	});
 }
 
-template <typename Error = no_error>
+inline auto single() {
-inline producer<empty_value, Error> single() {
+	using consumer_type = consumer<empty_value, no_error>;
-	return [](const consumer<empty_value, Error> &consumer) {
+	return make_producer<>([](
+			const consumer_type &consumer) {
 		consumer.put_next({});
 		consumer.put_done();
 		return lifetime();
-	};
+	});
 }
 
-template <typename Value, typename Error = no_error>
+template <typename Value>
-inline producer<Value, Error> vector(std::vector<Value> &&values) {
+inline auto vector(std::vector<Value> &&values) {
-	return [values = std::move(values)](
+	using consumer_type = consumer<Value, no_error>;
-			const consumer<Value, Error> &consumer) mutable {
+	return make_producer<Value>([
+		values = std::move(values)
+	](const consumer_type &consumer) mutable {
 		for (auto &value : values) {
 			consumer.put_next(std::move(value));
 		}
 		consumer.put_done();
 		return lifetime();
-	};
+	});
 }
 
-template <typename Error = no_error>
+inline auto vector(std::vector<bool> &&values) {
-inline producer<bool, Error> vector(std::vector<bool> &&values) {
+	using consumer_type = consumer<bool, no_error>;
-	return [values = std::move(values)](
+	return make_producer<bool>([
-			const consumer<bool, Error> &consumer) mutable {
+		values = std::move(values)
+	](const consumer_type &consumer) {
 		for (auto value : values) {
 			consumer.put_next_copy(value);
 		}
 		consumer.put_done();
 		return lifetime();
-	};
+	});
 }
 
-template <typename Value, typename Error = no_error, typename Range>
+template <
-inline producer<Value, Error> range(Range &&range) {
+	typename Range,
+	typename Value = std::decay_t<
+		decltype(*std::begin(std::declval<Range>()))>>
+inline auto range(Range &&range) {
 	return vector(std::vector<Value>(
 		std::begin(range),
 		std::end(range)));
 }
 
-inline producer<int> ints(int from, int till) {
+inline auto ints(int from, int till) {
 	Expects(from <= till);
-	return [from, till](const consumer<int> &consumer) {
+	return make_producer<int>([from, till](
+			const consumer<int> &consumer) {
 		for (auto i = from; i != till; ++i) {
 			consumer.put_next_copy(i);
 		}
 		consumer.put_done();
 		return lifetime();
-	};
+	});
 }
 
-inline producer<int> ints(int count) {
+inline auto ints(int count) {
 	return ints(0, count);
 }
 

Telegram/SourceFiles/rpl/rpl.h

@@ -36,6 +36,7 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #include <rpl/mappers.h>
 #include <rpl/filter.h>
 #include <rpl/distinct_until_changed.h>
+#include <rpl/type_erased.h>
 #include <rpl/flatten_latest.h>
 #include <rpl/combine.h>
 #include <rpl/combine_previous.h>

Telegram/SourceFiles/rpl/then.h

@@ -23,36 +23,64 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #include <rpl/producer.h>
 
 namespace rpl {
+namespace details {
 
-template <typename Value, typename Error>
+template <typename Value, typename Error, typename Generator>
-inline auto then(producer<Value, Error> &&following) {
+class then_helper {
-	return [following = std::move(following)](
+public:
-			producer<Value, Error> &&initial) mutable
+	then_helper(producer<Value, Error, Generator> &&following)
-			-> producer<Value, Error> {
+	: _following(std::move(following)) {
-		return [
+	}
+
+	template <
+		typename OtherValue,
+		typename OtherError,
+		typename OtherGenerator,
+		typename NewValue = superset_type_t<Value, OtherValue>,
+		typename NewError = superset_type_t<Error, OtherError>>
+	auto operator()(
+		producer<OtherValue, OtherError, OtherGenerator> &&initial
+	) {
+		using consumer_type = consumer<NewValue, NewError>;
+		return make_producer<NewValue, NewError>([
 			initial = std::move(initial),
-			following = std::move(following)
+			following = std::move(_following)
-		](const consumer<Value, Error> &consumer) mutable {
+		](const consumer_type &consumer) mutable {
 			return std::move(initial).start(
 			[consumer](auto &&value) {
-				consumer.put_next_forward(std::forward<decltype(value)>(value));
+				consumer.put_next_forward(
+					std::forward<decltype(value)>(value));
 			}, [consumer](auto &&error) {
-				consumer.put_error_forward(std::forward<decltype(error)>(error));
+				consumer.put_error_forward(
+					std::forward<decltype(error)>(error));
 			}, [
 				consumer,
 				following = std::move(following)
 			]() mutable {
 				consumer.add_lifetime(std::move(following).start(
 				[consumer](auto &&value) {
-					consumer.put_next_forward(std::forward<decltype(value)>(value));
+					consumer.put_next_forward(
+						std::forward<decltype(value)>(value));
 				}, [consumer](auto &&error) {
-					consumer.put_error_forward(std::forward<decltype(error)>(error));
+					consumer.put_error_forward(
+						std::forward<decltype(error)>(error));
 				}, [consumer] {
 					consumer.put_done();
 				}));
 			});
-		};
+		});
-	};
+	}
+
+private:
+	producer<Value, Error, Generator> _following;
+
+};
+
+} // namespace details
+template <typename Value, typename Error, typename Generator>
+inline auto then(producer<Value, Error, Generator> &&following)
+-> details::then_helper<Value, Error, Generator> {
+	return { std::move(following) };
 }
 
 } // namespace rpl

Telegram/SourceFiles/rpl/type_erased.h

@@ -0,0 +1,45 @@
+/*
+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 <rpl/producer.h>
+
+namespace rpl {
+namespace details {
+
+class type_erased_helper {
+public:
+	template <typename Value, typename Error, typename Generator>
+	producer<Value, Error> operator()(
+			producer<Value, Error, Generator> &&initial) const {
+		return std::move(initial);
+	}
+
+};
+
+} // namespace details
+
+inline auto type_erased()
+-> details::type_erased_helper {
+	return details::type_erased_helper();
+}
+
+} // namespace rpl

Telegram/SourceFiles/rpl/variable.h

@@ -62,9 +62,11 @@ public:
 
 	template <
 		typename OtherType,
+		typename Error,
+		typename Generator,
 		typename = std::enable_if_t<
 			std::is_assignable_v<Type, OtherType>>>
-	variable(rpl::producer<OtherType> &&stream) {
+	variable(producer<OtherType, Error, Generator> &&stream) {
 		std::move(stream)
 			| start_with_next([this](auto &&data) {
 				*this = std::forward<decltype(data)>(data);
@@ -73,9 +75,12 @@ public:
 
 	template <
 		typename OtherType,
+		typename Error,
+		typename Generator,
 		typename = std::enable_if_t<
 			std::is_assignable_v<Type, OtherType>>>
-	variable &operator=(rpl::producer<OtherType> &&stream) {
+	variable &operator=(
+			producer<OtherType, Error, Generator> &&stream) {
 		_lifetime.destroy();
 		std::move(stream)
 			| start_with_next([this](auto &&data) {
@@ -86,7 +91,7 @@ public:
 	Type current() const {
 		return _data;
 	}
-	rpl::producer<Type> value() const {
+	auto value() const {
 		return _changes.events_starting_with_copy(_data);
 	}
 
@@ -100,8 +105,8 @@ private:
 	}
 
 	Type _data;
-	rpl::event_stream<Type> _changes;
+	event_stream<Type> _changes;
-	rpl::lifetime _lifetime;
+	lifetime _lifetime;
 
 };
 

Telegram/gyp/tests/tests.gyp

@@ -98,6 +98,7 @@
     ],
     'sources': [
       '<(src_loc)/rpl/details/callable.h',
+      '<(src_loc)/rpl/details/superset_type.h',
       '<(src_loc)/rpl/details/type_list.h',
       '<(src_loc)/rpl/after_next.h',
       '<(src_loc)/rpl/before_next.h',
@@ -121,6 +122,7 @@
       '<(src_loc)/rpl/range.h',
       '<(src_loc)/rpl/rpl.h',
       '<(src_loc)/rpl/then.h',
+      '<(src_loc)/rpl/type_erased.h',
       '<(src_loc)/rpl/variable.h',
     ],
   }],