procxx
/
kepka
mirror of https://github.com/procxx/kepka.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Support errors in rpl::event_stream.

This commit is contained in:
John Preston 2019-02-16 16:47:52 +04:00
parent cf275b152a
commit 771a51224e
2 changed files with 156 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

225
Telegram/SourceFiles/rpl/event_stream.h
View File

@ -12,6 +12,7 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
#include <rpl/then.h> #include <rpl/then.h>
#include <rpl/range.h> #include <rpl/range.h>
#include <algorithm> #include <algorithm>
#include <optional>
#include "base/assertion.h" #include "base/assertion.h"
#include "base/index_based_iterator.h" #include "base/index_based_iterator.h"
@ -19,7 +20,7 @@ namespace rpl {
// Currently not thread-safe :( // Currently not thread-safe :(
template <typename Value = empty_value> template <typename Value = empty_value, typename Error = no_error>
class event_stream { class event_stream {
public: public:
event_stream(); event_stream();
@ -34,17 +35,29 @@ public:
void fire_copy(const Value &value) const { void fire_copy(const Value &value) const {
return fire_forward(value); return fire_forward(value);
} }
#if defined _MSC_VER && _MSC_VER >= 1914 && false
producer<Value> events() const { template <typename OtherError>
#else // _MSC_VER >= 1914 void fire_error_forward(OtherError &&error) const;
void fire_error(Error &&error) const {
return fire_error_forward(std::move(error));
}
void fire_error_copy(const Error &error) const {
return fire_error_forward(error);
}
void fire_done() const;
#if defined _MSC_VER && _MSC_VER >= 1914 && _MSC_VER < 1916
producer<Value, Error> events() const {
#else // _MSC_VER >= 1914 && _MSC_VER < 1916
auto events() const { auto events() const {
#endif // _MSC_VER >= 1914 #endif // _MSC_VER >= 1914 && _MSC_VER < 1916
return make_producer<Value>([weak = make_weak()]( return make_producer<Value, Error>([weak = make_weak()](
const auto &consumer) { const auto &consumer) {
if (auto strong = weak.lock()) { if (const auto strong = weak.lock()) {
auto result = [weak, consumer] { auto result = [weak, consumer] {
if (auto strong = weak.lock()) { if (const auto strong = weak.lock()) {
auto it = std::find( const auto it = std::find(
strong->consumers.begin(), strong->consumers.begin(),
strong->consumers.end(), strong->consumers.end(),
consumer); consumer);
@ -73,7 +86,7 @@ public:
private: private:
struct Data { struct Data {
std::vector<consumer<Value, no_error>> consumers; std::vector<consumer<Value, Error>> consumers;
int depth = 0; int depth = 0;
}; };
std::weak_ptr<Data> make_weak() const; std::weak_ptr<Data> make_weak() const;
@ -82,72 +95,113 @@ private:
}; };
template <typename Value> template <typename Value, typename Error>
inline event_stream<Value>::event_stream() { inline event_stream<Value, Error>::event_stream() {
} }
template <typename Value> template <typename Value, typename Error>
inline event_stream<Value>::event_stream(event_stream &&other) inline event_stream<Value, Error>::event_stream(event_stream &&other)
: _data(details::take(other._data)) { : _data(details::take(other._data)) {
} }
template <typename Value> template <typename Value, typename Error>
inline event_stream<Value> &event_stream<Value>::operator=( inline event_stream<Value, Error> &event_stream<Value, Error>::operator=(
event_stream &&other) { event_stream &&other) {
_data = details::take(other._data); if (this != &other) {
fire_done();
_data = details::take(other._data);
}
return *this; return *this;
} }
template <typename Value> template <typename Value, typename Error>
template <typename OtherValue> template <typename OtherValue>
inline void event_stream<Value>::fire_forward( inline void event_stream<Value, Error>::fire_forward(
OtherValue &&value) const { OtherValue &&value) const {
auto copy = _data; if (!_data) {
if (!copy) { return;
}
const auto copy = _data;
auto &consumers = copy->consumers;
if (consumers.empty()) {
return; return;
} }
++copy->depth; ++copy->depth;
auto &consumers = copy->consumers; const auto begin = base::index_based_begin(consumers);
auto begin = base::index_based_begin(consumers); const auto end = base::index_based_end(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);
});
// Perhaps move value for the last consumer. // Copy value for every consumer except the last.
if (prev->put_next_forward(std::forward<OtherValue>(value))) { const auto prev = end - 1;
if (removeFrom != prev) { auto staleFrom = std::remove_if(begin, prev, [&](const auto &consumer) {
*removeFrom++ = std::move(*prev); return !consumer.put_next_copy(value);
} else { });
++removeFrom;
// Perhaps move value for the last consumer.
if (prev->put_next_forward(std::forward<OtherValue>(value))) {
if (staleFrom != prev) {
*staleFrom++ = std::move(*prev);
} else {
++staleFrom;
}
}
if (staleFrom != end) {
// Move new consumers.
const auto newEnd = base::index_based_end(consumers);
if (newEnd != end) {
Assert(newEnd > end);
for (auto i = end; i != newEnd;) {
*staleFrom++ = *i++;
} }
} }
if (removeFrom != end) { // Erase stale consumers.
// Move new consumers. if (copy->depth == 1) {
auto newEnd = base::index_based_end(consumers); consumers.erase(staleFrom.base(), consumers.end());
if (newEnd != end) {
Assert(newEnd > end);
while (end != newEnd) {
*removeFrom++ = *end++;
}
}
// Erase stale consumers.
if (copy->depth == 1) {
consumers.erase(removeFrom.base(), consumers.end());
}
} }
} }
--copy->depth; --copy->depth;
} }
template <typename Value> template <typename Value, typename Error>
inline auto event_stream<Value>::make_weak() const template <typename OtherError>
inline void event_stream<Value, Error>::fire_error_forward(
OtherError &&error) const {
if (!_data) {
return;
}
const auto data = std::move(_data);
const auto &consumers = data->consumers;
if (consumers.empty()) {
return;
}
const auto begin = base::index_based_begin(consumers);
const auto end = base::index_based_end(consumers);
// Copy error for every consumer except the last.
const auto prev = end - 1;
std::for_each(begin, prev, [&](const auto &consumer) {
consumer.put_error_copy(error);
});
// Perhaps move error for the last consumer.
prev->put_error_forward(std::forward<OtherError>(error));
// Just drop any new consumers.
}
template <typename Value, typename Error>
void event_stream<Value, Error>::fire_done() const {
if (const auto data = details::take(_data)) {
for (const auto &consumer : data->consumers) {
consumer.put_done();
}
}
}
template <typename Value, typename Error>
inline auto event_stream<Value, Error>::make_weak() const
-> std::weak_ptr<Data> { -> std::weak_ptr<Data> {
if (!_data) { if (!_data) {
_data = std::make_shared<Data>(); _data = std::make_shared<Data>();
@ -155,23 +209,30 @@ inline auto event_stream<Value>::make_weak() const
return _data; return _data;
} }
template <typename Value, typename Error>
template <typename Value> inline event_stream<Value, Error>::~event_stream() {
inline event_stream<Value>::~event_stream() { fire_done();
if (auto data = details::take(_data)) {
for (auto &consumer : data->consumers) {
consumer.put_done();
}
}
} }
template <typename Value> template <typename Value, typename Error>
inline auto start_to_stream( inline auto start_to_stream(
event_stream<Value> &stream, event_stream<Value, Error> &stream,
lifetime &alive_while) { lifetime &alive_while) {
return start_with_next([&stream](auto &&value) { if constexpr (std::is_same_v<Error, no_error>) {
stream.fire_forward(std::forward<decltype(value)>(value)); return start_with_next_done([&](auto &&value) {
}, alive_while); stream.fire_forward(std::forward<decltype(value)>(value));
}, [&] {
stream.fire_done();
}, alive_while);
} else {
return start_with_next_error_done([&](auto &&value) {
stream.fire_forward(std::forward<decltype(value)>(value));
}, [&](auto &&error) {
stream.fire_error_forward(std::forward<decltype(error)>(error));
}, [&] {
stream.fire_done();
}, alive_while);
}
} }
namespace details { namespace details {
@ -184,19 +245,37 @@ public:
template <typename Value, typename Error, typename Generator> template <typename Value, typename Error, typename Generator>
auto operator()(producer<Value, Error, Generator> &&initial) { auto operator()(producer<Value, Error, Generator> &&initial) {
auto stream = _lifetime.make_state<event_stream<Value>>(); auto stream = _lifetime.make_state<event_stream<Value, Error>>();
auto collected = std::vector<Value>(); auto values = std::vector<Value>();
{ if constexpr (std::is_same_v<Error, rpl::no_error>) {
auto collecting = stream->events().start( auto collecting = stream->events().start(
[&collected](Value &&value) { [&](Value &&value) { values.push_back(std::move(value)); },
collected.push_back(std::move(value));
},
[](const Error &error) {}, [](const Error &error) {},
[] {}); [] {});
std::move(initial) | start_to_stream(*stream, _lifetime); std::move(initial) | start_to_stream(*stream, _lifetime);
collecting.destroy();
return vector(std::move(values)) | then(stream->events());
} else {
auto maybeError = std::optional<Error>();
auto collecting = stream->events().start(
[&](Value && value) { values.push_back(std::move(value)); },
[](Error &&error) { maybeError = std::move(error); },
[] {});
std::move(initial) | start_to_stream(*stream, _lifetime);
collecting.destroy();
if (maybeError.has_value()) {
return rpl::producer<Value, Error>([
error = std::move(*maybeError)
](const auto &consumer) mutable {
consumer.put_error(std::move(error));
});
}
return rpl::producer<Value, Error>(vector<Value, Error>(
std::move(values)
) | then(stream->events()));
} }
return vector(std::move(collected))
| then(stream->events());
} }
private: private:

7
Telegram/SourceFiles/rpl/range.h
View File

@ -31,9 +31,9 @@ inline auto single() {
}); });
} }
template <typename Value> template <typename Value, typename Error = no_error>
inline auto vector(std::vector<Value> &&values) { inline auto vector(std::vector<Value> &&values) {
return make_producer<Value>([ return make_producer<Value, Error>([
values = std::move(values) values = std::move(values)
](const auto &consumer) mutable { ](const auto &consumer) mutable {
for (auto &value : values) { for (auto &value : values) {
@ -44,8 +44,9 @@ inline auto vector(std::vector<Value> &&values) {
}); });
} }
template <typename Error = no_error>
inline auto vector(std::vector<bool> &&values) { inline auto vector(std::vector<bool> &&values) {
return make_producer<bool>([ return make_producer<bool, Error>([
values = std::move(values) values = std::move(values)
](const auto &consumer) { ](const auto &consumer) {
for (auto value : values) { for (auto value : values) {