Add rpl::variable, improve filter / combine.

Telegram/SourceFiles/rpl/after_next.h

@@ -0,0 +1,72 @@
+/*
+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 {
+
+template <typename SideEffect>
+class after_next_helper {
+public:
+	template <typename OtherSideEffect>
+	after_next_helper(OtherSideEffect &&method)
+		: _method(std::forward<OtherSideEffect>(method)) {
+	}
+
+	template <typename Value, typename Error>
+	rpl::producer<Value, Error> operator()(
+			rpl::producer<Value, Error> &&initial) {
+		return [
+			initial = std::move(initial),
+			method = std::move(_method)
+		](const consumer<Value, Error> &consumer) mutable {
+			return std::move(initial).start(
+			[method = std::move(method), consumer](auto &&value) {
+				auto copy = method;
+				consumer.put_next_copy(value);
+				std::move(copy)(
+					std::forward<decltype(value)>(value));
+			}, [consumer](auto &&error) {
+				consumer.put_error_forward(
+					std::forward<decltype(error)>(error));
+			}, [consumer] {
+				consumer.put_done();
+			});
+		};
+	}
+
+private:
+	SideEffect _method;
+
+};
+
+} // namespace details
+
+template <typename SideEffect>
+inline auto after_next(SideEffect &&method)
+-> details::after_next_helper<std::decay_t<SideEffect>> {
+	return details::after_next_helper<std::decay_t<SideEffect>>(
+		std::forward<SideEffect>(method));
+}
+
+} // namespace rpl

Telegram/SourceFiles/rpl/combine.h

@@ -20,6 +20,7 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 */
 #pragma once
 
+#include "base/optional.h"
 #include <rpl/producer.h>
 #include <rpl/details/type_list.h>
 #include <rpl/details/callable.h>

Telegram/SourceFiles/rpl/filter.h

@@ -21,6 +21,7 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #pragma once
 
 #include <rpl/producer.h>
+#include <rpl/combine.h>
 
 namespace rpl {
 namespace details {
@@ -33,7 +34,11 @@ public:
 		: _predicate(std::forward<OtherPredicate>(predicate)) {
 	}
 
-	template <typename Value, typename Error>
+	template <
+		typename Value,
+		typename Error,
+		typename = std::enable_if_t<
+			details::is_callable_v<Predicate, Value>>>
 	rpl::producer<Value, Error> operator()(
 			rpl::producer<Value, Error> &&initial) {
 		return [
@@ -47,7 +52,7 @@ public:
 					predicate = std::move(predicate)
 				](auto &&value) {
 					const auto &immutable = value;
-					if (predicate(immutable)) {
+					if (details::callable_invoke(predicate, immutable)) {
 						consumer.put_next_forward(std::forward<decltype(value)>(value));
 					}
 				}, [consumer](auto &&error) {
@@ -72,4 +77,31 @@ inline auto filter(Predicate &&predicate)
 		std::forward<Predicate>(predicate));
 }
 
+namespace details {
+
+template <>
+class filter_helper<producer<bool>> {
+public:
+	filter_helper(producer<bool> &&filterer)
+		: _filterer(std::move(filterer)) {
+	}
+
+	template <
+		typename Value,
+		typename Error>
+	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; })
+			| map([](auto &&value, bool) {
+				return std::forward<decltype(value)>(value);
+			});
+	}
+
+private:
+	producer<bool> _filterer;
+
+};
+
+} // namespace details
 } // namespace rpl

Telegram/SourceFiles/rpl/rpl.h

@@ -38,5 +38,8 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #include <rpl/distinct_until_changed.h>
 #include <rpl/flatten_latest.h>
 #include <rpl/combine.h>
+#include <rpl/combine_previous.h>
+#include <rpl/variable.h>
 
 #include <rpl/before_next.h>
+#include <rpl/after_next.h>

Telegram/SourceFiles/rpl/variable.h

@@ -0,0 +1,108 @@
+/*
+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>
+#include <rpl/event_stream.h>
+
+namespace rpl {
+
+template <typename Type>
+class variable {
+public:
+	variable() : _data{} {
+	}
+	variable(const variable &other) = default;
+	variable(variable &&other) = default;
+	variable &operator=(const variable &other) = default;
+	variable &operator=(variable &&other) = default;
+
+	variable(const Type &data) : _data(data) {
+	}
+	variable(Type &&data) : _data(std::move(data)) {
+	}
+	variable &operator=(const Type &data) {
+		return assign(data);
+	}
+
+	template <
+		typename OtherType,
+		typename = std::enable_if_t<
+			std::is_constructible_v<Type, OtherType&&>
+			&& !std::is_same_v<std::decay_t<OtherType>, Type>>>
+	variable(OtherType &&data) : _data(std::forward<OtherType>(data)) {
+	}
+
+	template <
+		typename OtherType,
+		typename = std::enable_if_t<
+			std::is_assignable_v<Type, OtherType&&>
+			&& !std::is_same_v<std::decay_t<OtherType>, Type>>>
+	variable &operator=(OtherType &&data) {
+		return assign(std::forward<OtherType>(data));
+	}
+
+	template <
+		typename OtherType,
+		typename = std::enable_if_t<
+			std::is_assignable_v<Type, OtherType>>>
+	variable(rpl::producer<OtherType> &&stream) {
+		std::move(stream)
+			| start([this](auto &&data) {
+				*this = std::forward<decltype(data)>(data);
+			}, _lifetime);
+	}
+
+	template <
+		typename OtherType,
+		typename = std::enable_if_t<
+			std::is_assignable_v<Type, OtherType>>>
+	variable &operator=(rpl::producer<OtherType> &&stream) {
+		_lifetime.destroy();
+		std::move(stream)
+			| start([this](auto &&data) {
+				*this = std::forward<decltype(data)>(data);
+			}, _lifetime);
+	}
+
+	Type current() const {
+		return _data;
+	}
+	rpl::producer<Type> value() const {
+		return _changes.events_starting_with_copy(_data);
+	}
+
+private:
+	template <typename OtherType>
+	variable &assign(OtherType &&data) {
+		_lifetime.destroy();
+		_data = std::forward<OtherType>(data);
+		_changes.fire_copy(_data);
+		return *this;
+	}
+
+	Type _data;
+	rpl::event_stream<Type> _changes;
+	rpl::lifetime _lifetime;
+
+};
+
+} // namespace rpl

Telegram/gyp/tests/tests.gyp

@@ -99,6 +99,7 @@
     'sources': [
       '<(src_loc)/rpl/details/callable.h',
       '<(src_loc)/rpl/details/type_list.h',
+      '<(src_loc)/rpl/after_next.h',
       '<(src_loc)/rpl/before_next.h',
       '<(src_loc)/rpl/combine.h',
       '<(src_loc)/rpl/complete.h',
@@ -119,6 +120,7 @@
       '<(src_loc)/rpl/rpl.h',
       '<(src_loc)/rpl/single.h',
       '<(src_loc)/rpl/then.h',
+      '<(src_loc)/rpl/variable.h',
     ],
   }],
 }