Add core rpl::producer/consumer implementation.

Telegram/SourceFiles/base/assertion.h

@@ -43,10 +43,20 @@ inline constexpr void noop() {
 	std::abort();
 }
 
+#ifndef GSL_UNLIKELY
+#define DEFINED_GSL_UNLIKELY_
+#define GSL_UNLIKELY(expression) (expression)
+#endif // GSL_UNLIKELY
+
 inline constexpr void validate(bool condition, const char *message, const char *file, int line) {
 	(GSL_UNLIKELY(!(condition))) ? fail(message, file, line) : noop();
 }
 
+#ifdef DEFINED_GSL_UNLIKELY_
+#undef GSL_UNLIKELY
+#undef DEFINED_GSL_UNLIKELY_
+#endif // DEFINED_GSL_UNLIKELY_
+
 } // namespace assertion
 } // namespace base
 

Telegram/SourceFiles/base/tests_main.cpp

@@ -23,6 +23,17 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #include "reporters/catch_reporter_compact.hpp"
 #include <QFile>
 
+namespace base {
+namespace assertion {
+
+// For Assert() / Expects() / Ensures() / Unexpected() to work.
+void log(const char *message, const char *file, int line) {
+	std::cout << message << " (" << file << ":" << line << ")" << std::endl;
+}
+
+} // namespace assertion
+} // namespace base
+
 namespace Catch {
 
 	struct MinimalReporter : CompactReporter {
@@ -82,15 +93,20 @@ namespace Catch {
 } // end namespace Catch
 
 int main(int argc, const char *argv[]) {
-	const char *catch_argv[] = { argv[0], "-r", "minimal" };
+	auto touchFile = QString();
+	for (auto i = 0; i != argc; ++i) {
+		if (argv[i] == QString("--touch") && i + 1 != argc) {
+			touchFile = QFile::decodeName(argv[++i]);
+		}
+	}
+	const char *catch_argv[] = {
+		argv[0],
+		touchFile.isEmpty() ? "-b" : "-r",
+		touchFile.isEmpty() ? "-b" : "minimal" };
 	constexpr auto catch_argc = sizeof(catch_argv) / sizeof(catch_argv[0]);
 	auto result = Catch::Session().run(catch_argc, catch_argv);
-	if (result == 0) {
-		for (auto i = 0; i != argc; ++i) {
-			if (argv[i] == QString("--touch") && i + 1 != argc) {
-				QFile(QFile::decodeName(argv[++i])).open(QIODevice::WriteOnly);
-			}
-		}
+	if (result == 0 && !touchFile.isEmpty()) {
+		QFile(touchFile).open(QIODevice::WriteOnly);
 	}
 	return (result < 0xff ? result : 0xff);
 }

Telegram/SourceFiles/rpl/consumer.h

@@ -0,0 +1,242 @@
+/*
+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/lifetime.h"
+#include <mutex>
+
+namespace rpl {
+
+struct no_value {
+};
+
+struct no_error {
+	no_error() = delete;
+};
+
+template <typename Value, typename Error>
+class consumer {
+public:
+	template <
+		typename OnNext,
+		typename OnError,
+		typename OnDone,
+		typename = decltype(std::declval<OnNext>()(std::declval<Value>())),
+		typename = decltype(std::declval<OnError>()(std::declval<Error>())),
+		typename = decltype(std::declval<OnDone>()())>
+	consumer(
+		OnNext &&next,
+		OnError &&error,
+		OnDone &&done);
+
+	bool putNext(Value value) const;
+	void putError(Error error) const;
+	void putDone() const;
+
+	void setLifetime(lifetime &&lifetime) const;
+	void terminate() const;
+
+private:
+	class abstract_consumer_instance;
+
+	template <typename OnNext, typename OnError, typename OnDone>
+	class consumer_instance;
+
+	template <typename OnNext, typename OnError, typename OnDone>
+	std::shared_ptr<abstract_consumer_instance> ConstructInstance(
+		OnNext &&next,
+		OnError &&error,
+		OnDone &&done);
+
+	std::shared_ptr<abstract_consumer_instance> _instance;
+
+};
+
+
+template <typename Value, typename Error>
+class consumer<Value, Error>::abstract_consumer_instance {
+public:
+	virtual bool putNext(Value value) = 0;
+	virtual void putError(Error error) = 0;
+	virtual void putDone() = 0;
+
+	void setLifetime(lifetime &&lifetime);
+	void terminate();
+
+protected:
+	lifetime _lifetime;
+	bool _terminated = false;
+	std::mutex _mutex;
+
+};
+
+template <typename Value, typename Error>
+template <typename OnNext, typename OnError, typename OnDone>
+class consumer<Value, Error>::consumer_instance
+	: public consumer<Value, Error>::abstract_consumer_instance {
+public:
+	template <typename OnNextImpl, typename OnErrorImpl, typename OnDoneImpl>
+	consumer_instance(
+		OnNextImpl &&next,
+		OnErrorImpl &&error,
+		OnDoneImpl &&done)
+		: _next(std::forward<OnNextImpl>(next))
+		, _error(std::forward<OnErrorImpl>(error))
+		, _done(std::forward<OnDoneImpl>(done)) {
+	}
+
+	bool putNext(Value value) override;
+	void putError(Error error) override;
+	void putDone() override;
+
+private:
+	OnNext _next;
+	OnError _error;
+	OnDone _done;
+
+};
+
+template <typename Value, typename Error>
+template <typename OnNext, typename OnError, typename OnDone>
+std::shared_ptr<typename consumer<Value, Error>::abstract_consumer_instance>
+consumer<Value, Error>::ConstructInstance(
+		OnNext &&next,
+		OnError &&error,
+		OnDone &&done) {
+	return std::make_shared<consumer_instance<
+		std::decay_t<OnNext>,
+		std::decay_t<OnError>,
+		std::decay_t<OnDone>>>(
+			std::forward<OnNext>(next),
+			std::forward<OnError>(error),
+			std::forward<OnDone>(done));
+}
+
+template <typename Value, typename Error>
+template <
+	typename OnNext,
+	typename OnError,
+	typename OnDone,
+	typename,
+	typename,
+	typename>
+consumer<Value, Error>::consumer(
+	OnNext &&next,
+	OnError &&error,
+	OnDone &&done) : _instance(ConstructInstance(
+		std::forward<OnNext>(next),
+		std::forward<OnError>(error),
+		std::forward<OnDone>(done))) {
+}
+
+template <typename Value, typename Error>
+bool consumer<Value, Error>::putNext(Value value) const {
+	return _instance->putNext(std::move(value));
+}
+
+template <typename Value, typename Error>
+void consumer<Value, Error>::putError(Error error) const {
+	return _instance->putError(std::move(error));
+}
+
+template <typename Value, typename Error>
+void consumer<Value, Error>::putDone() const {
+	return _instance->putDone();
+}
+
+template <typename Value, typename Error>
+void consumer<Value, Error>::setLifetime(lifetime &&lifetime) const {
+	return _instance->setLifetime(std::move(lifetime));
+}
+
+template <typename Value, typename Error>
+void consumer<Value, Error>::terminate() const {
+	return _instance->terminate();
+}
+
+template <typename Value, typename Error>
+void consumer<Value, Error>::abstract_consumer_instance::setLifetime(
+		lifetime &&lifetime) {
+	std::unique_lock<std::mutex> lock(_mutex);
+	if (_terminated) {
+		lock.unlock();
+
+		lifetime.destroy();
+	} else {
+		_lifetime = std::move(lifetime);
+	}
+}
+
+template <typename Value, typename Error>
+void consumer<Value, Error>::abstract_consumer_instance::terminate() {
+	std::unique_lock<std::mutex> lock(_mutex);
+	if (!_terminated) {
+		_terminated = true;
+		auto handler = std::exchange(_lifetime, lifetime());
+		lock.unlock();
+
+		handler.destroy();
+	}
+}
+
+template <typename Value, typename Error>
+template <typename OnNext, typename OnError, typename OnDone>
+bool consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::putNext(
+		Value value) {
+	std::unique_lock<std::mutex> lock(this->_mutex);
+	if (this->_terminated) {
+		return false;
+	}
+	auto handler = this->_next;
+	lock.unlock();
+
+	handler(std::move(value));
+	return true;
+}
+
+template <typename Value, typename Error>
+template <typename OnNext, typename OnError, typename OnDone>
+void consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::putError(
+		Error error) {
+	std::unique_lock<std::mutex> lock(this->_mutex);
+	if (!this->_terminated) {
+		auto handler = std::move(this->_error);
+		lock.unlock();
+
+		handler(std::move(error));
+		this->terminate();
+	}
+}
+
+template <typename Value, typename Error>
+template <typename OnNext, typename OnError, typename OnDone>
+void consumer<Value, Error>::consumer_instance<OnNext, OnError, OnDone>::putDone() {
+	std::unique_lock<std::mutex> lock(this->_mutex);
+	if (!this->_terminated) {
+		auto handler = std::move(this->_done);
+		lock.unlock();
+
+		handler();
+		this->terminate();
+	}
+}
+
+} // namespace rpl

Telegram/SourceFiles/rpl/lifetime.h

@@ -0,0 +1,76 @@
+/*
+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/lambda.h"
+#include "base/algorithm.h"
+#include <functional>
+
+namespace rpl {
+
+class lifetime {
+public:
+	lifetime() = default;
+	lifetime(lifetime &&other);
+	lifetime &operator=(lifetime &&other);
+
+	template <typename Destroy, typename = decltype(std::declval<Destroy>()())>
+	lifetime(Destroy &&destroy) : _destroy(std::forward<Destroy>(destroy)) {
+	}
+
+	void add(lifetime other) {
+		_nested.push_back(std::move(other));
+	}
+
+	void destroy() {
+		auto nested = std::exchange(_nested, std::vector<lifetime>());
+		auto callback = std::exchange(_destroy, base::lambda_once<void()>());
+
+		if (!nested.empty()) {
+			nested.clear();
+		}
+		if (callback) {
+			callback();
+		}
+	}
+
+	~lifetime() {
+		destroy();
+	}
+
+private:
+	base::lambda_once<void()> _destroy;
+	std::vector<lifetime> _nested;
+
+};
+
+lifetime::lifetime(lifetime &&other)
+: _destroy(std::exchange(other._destroy, base::lambda_once<void()>()))
+, _nested(std::exchange(other._nested, std::vector<lifetime>())) {
+}
+
+lifetime &lifetime::operator=(lifetime &&other) {
+	_destroy = std::exchange(other._destroy, base::lambda_once<void()>());
+	_nested = std::exchange(other._nested, std::vector<lifetime>());
+	return *this;
+}
+
+} // namespace rpl

Telegram/SourceFiles/rpl/producer.h

@@ -0,0 +1,81 @@
+/*
+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/lambda.h"
+#include "rpl/consumer.h"
+#include "rpl/lifetime.h"
+
+namespace rpl {
+
+template <typename Value, typename Error>
+class producer {
+public:
+	template <typename Generator, typename = std::enable_if<std::is_convertible<
+		decltype(std::declval<Generator>()(std::declval<consumer<Value, Error>>())),
+		lifetime
+	>::value>>
+	producer(Generator &&generator);
+
+	template <
+		typename OnNext,
+		typename OnError,
+		typename OnDone,
+		typename = decltype(std::declval<OnNext>()(std::declval<Value>())),
+		typename = decltype(std::declval<OnError>()(std::declval<Error>())),
+		typename = decltype(std::declval<OnDone>()())>
+	lifetime start(
+		OnNext &&next,
+		OnError &&error,
+		OnDone &&done);
+
+private:
+	base::lambda<lifetime(consumer<Value, Error>)> _generator;
+
+};
+
+template <typename Value, typename Error>
+template <typename Generator, typename>
+producer<Value, Error>::producer(Generator &&generator)
+	: _generator(std::forward<Generator>(generator)) {
+}
+
+template <typename Value, typename Error>
+template <
+	typename OnNext,
+	typename OnError,
+	typename OnDone,
+	typename,
+	typename,
+	typename>
+lifetime producer<Value, Error>::start(
+		OnNext &&next,
+		OnError &&error,
+		OnDone &&done) {
+	auto result = consumer<Value, Error>(
+		std::forward<OnNext>(next),
+		std::forward<OnError>(error),
+		std::forward<OnDone>(done));
+	result.setLifetime(_generator(result));
+	return [result] { result.terminate(); };
+}
+
+} // namespace rpl

Telegram/SourceFiles/rpl/producer_tests.cpp

@@ -0,0 +1,159 @@
+/*
+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
+*/
+#include "catch.hpp"
+
+#include "rpl/producer.h"
+
+using namespace rpl;
+
+class OnDestructor {
+public:
+	OnDestructor(base::lambda_once<void()> callback) : _callback(std::move(callback)) {
+	}
+	~OnDestructor() {
+		if (_callback) {
+			_callback();
+		}
+	}
+
+private:
+	base::lambda_once<void()> _callback;
+
+};
+
+TEST_CASE("basic producer tests", "[rpl::producer]") {
+	SECTION("producer next, done and lifetime end test") {
+		auto lifetimeEnded = std::make_shared<bool>(false);
+		auto sum = std::make_shared<int>(0);
+		auto doneGenerated = std::make_shared<bool>(false);
+		auto destroyed = std::make_shared<bool>(false);
+		{
+			auto destroyCaller = std::make_shared<OnDestructor>([=] {
+				*destroyed = true;
+			});
+			{
+				producer<int, no_error>([=](auto consumer) {
+					destroyCaller;
+					consumer.putNext(1);
+					consumer.putNext(2);
+					consumer.putNext(3);
+					consumer.putDone();
+					return [=] {
+						destroyCaller;
+						*lifetimeEnded = true;
+					};
+				}).start([=](int value) {
+					destroyCaller;
+					*sum += value;
+				}, [=](no_error) {
+					destroyCaller;
+				}, [=]() {
+					destroyCaller;
+					*doneGenerated = true;
+				});
+			}
+		}
+		REQUIRE(*sum == 1 + 2 + 3);
+		REQUIRE(*doneGenerated);
+		REQUIRE(*lifetimeEnded);
+		REQUIRE(*destroyed);
+	}
+
+	SECTION("producer error test") {
+		auto errorGenerated = std::make_shared<bool>(false);
+		{
+			producer<no_value, bool>([=](auto consumer) {
+				consumer.putError(true);
+				return lifetime();
+			}).start([=](no_value) {
+			}, [=](bool error) {
+				*errorGenerated = error;
+			}, [=]() {
+			});
+		}
+		REQUIRE(*errorGenerated);
+	}
+
+	SECTION("nested lifetimes test") {
+		auto lifetimeEndCount = std::make_shared<int>(0);
+		{
+			auto lifetimes = lifetime();
+			{
+				auto testProducer = producer<no_value, no_error>([=](auto consumer) {
+					return [=] {
+						++*lifetimeEndCount;
+					};
+				});
+				lifetimes.add(testProducer.start([=](no_value) {
+				}, [=](no_error) {
+				}, [=] {
+				}));
+				lifetimes.add(testProducer.start([=](no_value) {
+				}, [=](no_error) {
+				}, [=] {
+				}));
+			}
+			REQUIRE(*lifetimeEndCount == 0);
+		}
+		REQUIRE(*lifetimeEndCount == 2);
+	}
+
+	SECTION("nested producers test") {
+		auto sum = std::make_shared<int>(0);
+		auto lifetimeEndCount = std::make_shared<int>(0);
+		auto saved = lifetime();
+		{
+			saved = producer<int, no_error>([=](auto consumer) {
+				auto inner = producer<int, no_error>([=](auto consumer) {
+					consumer.putNext(1);
+					consumer.putNext(2);
+					consumer.putNext(3);
+					return [=] {
+						++*lifetimeEndCount;
+					};
+				});
+				auto result = lifetime([=] {
+					++*lifetimeEndCount;
+				});
+				result.add(inner.start([=](int value) {
+					consumer.putNext(value);
+				}, [=](no_error) {
+				}, [=] {
+				}));
+				result.add(inner.start([=](int value) {
+					consumer.putNext(value);
+				}, [=](no_error) {
+				}, [=] {
+				}));
+				return result;
+			}).start([=](int value) {
+				*sum += value;
+			}, [=](no_error) {
+			}, [=] {
+			});
+		}
+		REQUIRE(*sum == 1 + 2 + 3 + 1 + 2 + 3);
+		REQUIRE(*lifetimeEndCount == 0);
+		saved.destroy();
+		REQUIRE(*lifetimeEndCount == 3);
+	}
+}
+

Telegram/gyp/telegram_sources.txt

@@ -393,6 +393,9 @@
 <(src_loc)/profile/profile_userpic_button.h
 <(src_loc)/profile/profile_widget.cpp
 <(src_loc)/profile/profile_widget.h
+<(src_loc)/rpl/consumer.h
+<(src_loc)/rpl/lifetime.h
+<(src_loc)/rpl/producer.h
 <(src_loc)/settings/settings_advanced_widget.cpp
 <(src_loc)/settings/settings_advanced_widget.h
 <(src_loc)/settings/settings_background_widget.cpp

Telegram/gyp/tests/tests.gyp

@@ -82,5 +82,16 @@
       '<(src_loc)/base/flags.h',
       '<(src_loc)/base/flags_tests.cpp',
     ],
+  }, {
+    'target_name': 'tests_producer',
+    'includes': [
+      'common_test.gypi',
+    ],
+    'sources': [
+      '<(src_loc)/rpl/consumer.h',
+      '<(src_loc)/rpl/lifetime.h',
+      '<(src_loc)/rpl/producer.h',
+      '<(src_loc)/rpl/producer_tests.cpp',
+    ],
   }],
 }

Telegram/gyp/tests/tests_list.txt

@@ -1,3 +1,4 @@
 tests_flat_map
 tests_flat_set
 tests_flags
+tests_producer