Implement SharedMediaWithLastViewer using rpl.

Telegram/SourceFiles/history/history_shared_media.cpp

@@ -28,7 +28,7 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 
 namespace {
 
-using Type = SharedMediaViewer::Type;
+using Type = SharedMediaSlice::Type;
 
 inline MediaOverviewType SharedMediaTypeToOverview(Type type) {
 	switch (type) {
@@ -61,14 +61,116 @@ void SharedMediaShowOverview(
 	}
 }
 
-SharedMediaSlice::SharedMediaSlice(Key key) : SharedMediaSlice(key, base::none) {
+class SharedMediaSliceBuilder {
+public:
+	using Type = Storage::SharedMediaType;
+	using Key = Storage::SharedMediaKey;
+
+	SharedMediaSliceBuilder(Key key, int limitBefore, int limitAfter);
+
+	using Result = Storage::SharedMediaResult;
+	using SliceUpdate = Storage::SharedMediaSliceUpdate;
+	using RemoveOne = Storage::SharedMediaRemoveOne;
+	using RemoveAll = Storage::SharedMediaRemoveAll;
+	bool applyUpdate(const Result &result);
+	bool applyUpdate(const SliceUpdate &update);
+	bool applyUpdate(const RemoveOne &update);
+	bool applyUpdate(const RemoveAll &update);
+
+	void checkInsufficientMedia();
+	using AroundData = std::pair<MsgId, ApiWrap::SliceType>;
+	rpl::producer<AroundData> insufficientMediaAround() const {
+		return _insufficientMediaAround.events();
+	}
+
+	SharedMediaSlice snapshot() const;
+
+private:
+	enum class RequestDirection {
+		Before,
+		After,
+	};
+	void requestMessages(RequestDirection direction);
+	void sliceToLimits();
+
+	void mergeSliceData(
+		base::optional<int> count,
+		const base::flat_set<MsgId> &messageIds,
+		base::optional<int> skippedBefore = base::none,
+		base::optional<int> skippedAfter = base::none);
+
+	Key _key;
+	base::flat_set<MsgId> _ids;
+	MsgRange _range;
+	base::optional<int> _fullCount;
+	base::optional<int> _skippedBefore;
+	base::optional<int> _skippedAfter;
+	int _limitBefore = 0;
+	int _limitAfter = 0;
+
+	rpl::event_stream<AroundData> _insufficientMediaAround;
+
+};
+
+class SharedMediaMergedSliceBuilder {
+public:
+	using Type = SharedMediaMergedSlice::Type;
+	using Key = SharedMediaMergedSlice::Key;
+
+	SharedMediaMergedSliceBuilder(Key key);
+
+	void applyPartUpdate(SharedMediaSlice &&update);
+	void applyMigratedUpdate(SharedMediaSlice &&update);
+
+	SharedMediaMergedSlice snapshot() const;
+
+private:
+	Key _key;
+	SharedMediaSlice _part;
+	base::optional<SharedMediaSlice> _migrated;
+
+};
+
+class SharedMediaWithLastSliceBuilder {
+public:
+	using Type = SharedMediaWithLastSlice::Type;
+	using Key = SharedMediaWithLastSlice::Key;
+
+	SharedMediaWithLastSliceBuilder(Key key);
+
+	void applyViewerUpdate(SharedMediaMergedSlice &&update);
+	void applyEndingUpdate(SharedMediaMergedSlice &&update);
+
+	SharedMediaWithLastSlice snapshot() const;
+
+private:
+	Key _key;
+	SharedMediaWithLastSlice _data;
+
+};
+
+SharedMediaSlice::SharedMediaSlice(Key key) : SharedMediaSlice(
+	key,
+	{},
+	{},
+	base::none,
+	base::none,
+	base::none) {
 }
 
 SharedMediaSlice::SharedMediaSlice(
 	Key key,
-	base::optional<int> fullCount)
+	const base::flat_set<MsgId> &ids,
+	MsgRange range,
+	base::optional<int> fullCount,
+	base::optional<int> skippedBefore,
+	base::optional<int> skippedAfter)
 	: _key(key)
-	, _fullCount(fullCount) {
+	, _ids(ids)
+	, _range(range)
+	, _fullCount(fullCount)
+	, _skippedBefore(skippedBefore)
+	, _skippedAfter(skippedAfter) {
 }
 
 base::optional<int> SharedMediaSlice::indexOf(MsgId msgId) const {
@@ -119,129 +221,35 @@ QString SharedMediaSlice::debug() const {
 	return before + middle + after;
 }
 
-SharedMediaViewer::SharedMediaViewer(
+SharedMediaSliceBuilder::SharedMediaSliceBuilder(
 	Key key,
 	int limitBefore,
 	int limitAfter)
 	: _key(key)
 	, _limitBefore(limitBefore)
-	, _limitAfter(limitAfter)
+	, _limitAfter(limitAfter) {
-	, _data(_key) {
-	Expects(IsServerMsgId(key.messageId) || (key.messageId == 0));
-	Expects((key.messageId != 0) || (limitBefore == 0 && limitAfter == 0));
 }
 
-void SharedMediaViewer::start() {
+bool SharedMediaSliceBuilder::applyUpdate(const Result &result) {
-	auto applyUpdateCallback = [this](auto &update) {
-		this->applyUpdate(update);
-	};
-	subscribe(Auth().storage().sharedMediaSliceUpdated(), applyUpdateCallback);
-	subscribe(Auth().storage().sharedMediaOneRemoved(), applyUpdateCallback);
-	subscribe(Auth().storage().sharedMediaAllRemoved(), applyUpdateCallback);
-
-	loadInitial();
-}
-
-void SharedMediaViewer::loadInitial() {
-	auto weak = base::make_weak_unique(this);
-	Auth().storage().query(Storage::SharedMediaQuery(
-		_key,
-		_limitBefore,
-		_limitAfter), [weak](Storage::SharedMediaResult &&result) {
-		if (weak) {
-			weak->applyStoredResult(std::move(result));
-		}
-	});
-}
-
-void SharedMediaViewer::applyStoredResult(Storage::SharedMediaResult &&result) {
 	mergeSliceData(
 		result.count,
 		result.messageIds,
 		result.skippedBefore,
 		result.skippedAfter);
+	return true;
 }
 
-void SharedMediaViewer::mergeSliceData(
+bool SharedMediaSliceBuilder::applyUpdate(const SliceUpdate &update) {
-		base::optional<int> count,
-		const base::flat_set<MsgId> &messageIds,
-		base::optional<int> skippedBefore,
-		base::optional<int> skippedAfter) {
-	if (messageIds.empty()) {
-		if (count && _data._fullCount != count) {
-			_data._fullCount = count;
-			if (*_data._fullCount <= _data.size()) {
-				_data._fullCount = _data.size();
-				_data._skippedBefore = _data._skippedAfter = 0;
-			}
-			updated.notify(_data);
-		}
-		sliceToLimits();
-		return;
-	}
-	if (count) {
-		_data._fullCount = count;
-	}
-	auto wasMinId = _data._ids.empty() ? -1 : _data._ids.front();
-	auto wasMaxId = _data._ids.empty() ? -1 : _data._ids.back();
-	_data._ids.merge(messageIds.begin(), messageIds.end());
-
-	auto adjustSkippedBefore = [&](MsgId oldId, int oldSkippedBefore) {
-		auto it = _data._ids.find(oldId);
-		Assert(it != _data._ids.end());
-		_data._skippedBefore = oldSkippedBefore - (it - _data._ids.begin());
-		accumulate_max(*_data._skippedBefore, 0);
-	};
-	if (skippedBefore) {
-		adjustSkippedBefore(messageIds.front(), *skippedBefore);
-	} else if (wasMinId >= 0 && _data._skippedBefore) {
-		adjustSkippedBefore(wasMinId, *_data._skippedBefore);
-	} else {
-		_data._skippedBefore = base::none;
-	}
-
-	auto adjustSkippedAfter = [&](MsgId oldId, int oldSkippedAfter) {
-		auto it = _data._ids.find(oldId);
-		Assert(it != _data._ids.end());
-		_data._skippedAfter = oldSkippedAfter - (_data._ids.end() - it - 1);
-		accumulate_max(*_data._skippedAfter, 0);
-	};
-	if (skippedAfter) {
-		adjustSkippedAfter(messageIds.back(), *skippedAfter);
-	} else if (wasMaxId >= 0 && _data._skippedAfter) {
-		adjustSkippedAfter(wasMaxId, *_data._skippedAfter);
-	} else {
-		_data._skippedAfter = base::none;
-	}
-
-	if (_data._fullCount) {
-		if (_data._skippedBefore && !_data._skippedAfter) {
-			_data._skippedAfter = *_data._fullCount
-				- *_data._skippedBefore
-				- int(_data._ids.size());
-		} else if (_data._skippedAfter && !_data._skippedBefore) {
-			_data._skippedBefore = *_data._fullCount
-				- *_data._skippedAfter
-				- int(_data._ids.size());
-		}
-	}
-
-	sliceToLimits();
-
-	updated.notify(_data);
-}
-
-void SharedMediaViewer::applyUpdate(const SliceUpdate &update) {
 	if (update.peerId != _key.peerId || update.type != _key.type) {
-		return;
+		return false;
 	}
 	auto intersects = [](MsgRange range1, MsgRange range2) {
 		return (range1.from <= range2.till) && (range2.from <= range1.till);
 	};
 	if (!intersects(update.range, {
-		_data._ids.empty() ? _key.messageId : _data._ids.front(),
+		_ids.empty() ? _key.messageId : _ids.front(),
-		_data._ids.empty() ? _key.messageId : _data._ids.back() })) {
+		_ids.empty() ? _key.messageId : _ids.back() })) {
-		return;
+		return false;
 	}
 	auto skippedBefore = (update.range.from == 0)
 		? 0
@@ -254,92 +262,226 @@ void SharedMediaViewer::applyUpdate(const SliceUpdate &update) {
 		update.messages ? *update.messages : base::flat_set<MsgId> {},
 		skippedBefore,
 		skippedAfter);
+	return true;
 }
 
-void SharedMediaViewer::applyUpdate(const OneRemoved &update) {
+bool SharedMediaSliceBuilder::applyUpdate(const RemoveOne &update) {
 	if (update.peerId != _key.peerId || !update.types.test(_key.type)) {
-		return;
+		return false;
 	}
 	auto changed = false;
-	if (_data._fullCount && *_data._fullCount > 0) {
+	if (_fullCount && *_fullCount > 0) {
-		--*_data._fullCount;
+		--*_fullCount;
 		changed = true;
 	}
-	if (_data._ids.contains(update.messageId)) {
+	if (_ids.contains(update.messageId)) {
-		_data._ids.remove(update.messageId);
+		_ids.remove(update.messageId);
 		changed = true;
-	} else if (!_data._ids.empty()) {
+	} else if (!_ids.empty()) {
-		if (_data._ids.front() > update.messageId
+		if (_ids.front() > update.messageId
-			&& _data._skippedBefore
+			&& _skippedBefore
-			&& *_data._skippedBefore > 0) {
+			&& *_skippedBefore > 0) {
-			--*_data._skippedBefore;
+			--*_skippedBefore;
 			changed = true;
-		} else if (_data._ids.back() < update.messageId
+		} else if (_ids.back() < update.messageId
-			&& _data._skippedAfter
+			&& _skippedAfter
-			&& *_data._skippedAfter > 0) {
+			&& *_skippedAfter > 0) {
-			--*_data._skippedAfter;
+			--*_skippedAfter;
 			changed = true;
 		}
 	}
-	if (changed) {
+	return changed;
-		updated.notify(_data);
-	}
 }
 
-void SharedMediaViewer::applyUpdate(const AllRemoved &update) {
+bool SharedMediaSliceBuilder::applyUpdate(const RemoveAll &update) {
 	if (update.peerId != _key.peerId) {
+		return false;
+	}
+	_ids = {};
+	_range = { 0, ServerMaxMsgId };
+	_fullCount = 0;
+	_skippedBefore = 0;
+	_skippedAfter = 0;
+	return true;
+}
+
+void SharedMediaSliceBuilder::checkInsufficientMedia() {
+	sliceToLimits();
+}
+
+void SharedMediaSliceBuilder::mergeSliceData(
+	base::optional<int> count,
+	const base::flat_set<MsgId> &messageIds,
+	base::optional<int> skippedBefore,
+	base::optional<int> skippedAfter) {
+	if (messageIds.empty()) {
+		if (count && _fullCount != count) {
+			_fullCount = count;
+			if (*_fullCount <= _ids.size()) {
+				_fullCount = _ids.size();
+				_skippedBefore = _skippedAfter = 0;
+			}
+		}
+		sliceToLimits();
 		return;
 	}
-	_data = SharedMediaSlice(_key, 0);
+	if (count) {
-	updated.notify(_data);
+		_fullCount = count;
+	}
+	auto wasMinId = _ids.empty() ? -1 : _ids.front();
+	auto wasMaxId = _ids.empty() ? -1 : _ids.back();
+	_ids.merge(messageIds.begin(), messageIds.end());
+
+	auto adjustSkippedBefore = [&](MsgId oldId, int oldSkippedBefore) {
+		auto it = _ids.find(oldId);
+		Assert(it != _ids.end());
+		_skippedBefore = oldSkippedBefore - (it - _ids.begin());
+		accumulate_max(*_skippedBefore, 0);
+	};
+	if (skippedBefore) {
+		adjustSkippedBefore(messageIds.front(), *skippedBefore);
+	} else if (wasMinId >= 0 && _skippedBefore) {
+		adjustSkippedBefore(wasMinId, *_skippedBefore);
+	} else {
+		_skippedBefore = base::none;
+	}
+
+	auto adjustSkippedAfter = [&](MsgId oldId, int oldSkippedAfter) {
+		auto it = _ids.find(oldId);
+		Assert(it != _ids.end());
+		_skippedAfter = oldSkippedAfter - (_ids.end() - it - 1);
+		accumulate_max(*_skippedAfter, 0);
+	};
+	if (skippedAfter) {
+		adjustSkippedAfter(messageIds.back(), *skippedAfter);
+	} else if (wasMaxId >= 0 && _skippedAfter) {
+		adjustSkippedAfter(wasMaxId, *_skippedAfter);
+	} else {
+		_skippedAfter = base::none;
+	}
+
+	if (_fullCount) {
+		if (_skippedBefore && !_skippedAfter) {
+			_skippedAfter = *_fullCount
+				- *_skippedBefore
+				- int(_ids.size());
+		} else if (_skippedAfter && !_skippedBefore) {
+			_skippedBefore = *_fullCount
+				- *_skippedAfter
+				- int(_ids.size());
+		}
+	}
+
+	sliceToLimits();
 }
 
-void SharedMediaViewer::sliceToLimits() {
+void SharedMediaSliceBuilder::sliceToLimits() {
-	auto aroundIt = base::lower_bound(_data._ids, _key.messageId);
+	auto aroundIt = base::lower_bound(_ids, _key.messageId);
-	auto removeFromBegin = (aroundIt - _data._ids.begin() - _limitBefore);
+	auto removeFromBegin = (aroundIt - _ids.begin() - _limitBefore);
-	auto removeFromEnd = (_data._ids.end() - aroundIt - _limitAfter - 1);
+	auto removeFromEnd = (_ids.end() - aroundIt - _limitAfter - 1);
 	if (removeFromBegin > 0) {
-		_data._ids.erase(_data._ids.begin(), _data._ids.begin() + removeFromBegin);
+		_ids.erase(_ids.begin(), _ids.begin() + removeFromBegin);
-		if (_data._skippedBefore) {
+		if (_skippedBefore) {
-			*_data._skippedBefore += removeFromBegin;
+			*_skippedBefore += removeFromBegin;
 		}
-	} else if (removeFromBegin < 0 && (!_data._skippedBefore || *_data._skippedBefore > 0)) {
+	} else if (removeFromBegin < 0 && (!_skippedBefore || *_skippedBefore > 0)) {
 		requestMessages(RequestDirection::Before);
 	}
 	if (removeFromEnd > 0) {
-		_data._ids.erase(_data._ids.end() - removeFromEnd, _data._ids.end());
+		_ids.erase(_ids.end() - removeFromEnd, _ids.end());
-		if (_data._skippedAfter) {
+		if (_skippedAfter) {
-			*_data._skippedAfter += removeFromEnd;
+			*_skippedAfter += removeFromEnd;
 		}
-	} else if (removeFromEnd < 0 && (!_data._skippedAfter || *_data._skippedAfter > 0)) {
+	} else if (removeFromEnd < 0 && (!_skippedAfter || *_skippedAfter > 0)) {
 		requestMessages(RequestDirection::After);
 	}
 }
 
-void SharedMediaViewer::requestMessages(RequestDirection direction) {
+void SharedMediaSliceBuilder::requestMessages(RequestDirection direction) {
 	using SliceType = ApiWrap::SliceType;
-	auto requestAroundData = [&]() -> std::pair<MsgId, SliceType> {
+	auto requestAroundData = [&]() -> AroundData {
-		if (_data._ids.empty()) {
+		if (_ids.empty()) {
 			return { _key.messageId, SliceType::Around };
 		} else if (direction == RequestDirection::Before) {
-			return { _data._ids.front(), SliceType::Before };
+			return { _ids.front(), SliceType::Before };
 		}
-		return { _data._ids.back(), SliceType::After };
+		return { _ids.back(), SliceType::After };
-	}();
+	};
-	Auth().api().requestSharedMedia(
+	_insufficientMediaAround.fire(requestAroundData());
-		App::peer(_key.peerId),
-		_key.type,
-		requestAroundData.first,
-		requestAroundData.second);
 }
 
-SharedMediaSliceMerged::SharedMediaSliceMerged(Key key) : SharedMediaSliceMerged(
+SharedMediaSlice SharedMediaSliceBuilder::snapshot() const {
+	return SharedMediaSlice(
+		_key,
+		_ids,
+		_range,
+		_fullCount,
+		_skippedBefore,
+		_skippedAfter);
+}
+
+rpl::producer<SharedMediaSlice> SharedMediaViewer(
+		SharedMediaSlice::Key key,
+		int limitBefore,
+		int limitAfter) {
+	Expects(IsServerMsgId(key.messageId) || (key.messageId == 0));
+	Expects((key.messageId != 0) || (limitBefore == 0 && limitAfter == 0));
+
+	return [=](auto consumer) {
+		auto lifetime = rpl::lifetime();
+		auto builder = lifetime.make_state<SharedMediaSliceBuilder>(
+			key,
+			limitBefore,
+			limitAfter);
+		auto applyUpdate = [=](auto &&update) {
+			if (builder->applyUpdate(std::move(update))) {
+				consumer.put_next(builder->snapshot());
+			}
+		};
+		auto requestMediaAround = [
+			peer = App::peer(key.peerId),
+			type = key.type
+		](SharedMediaSliceBuilder::AroundData data) {
+			Auth().api().requestSharedMedia(
+				peer,
+				type,
+				data.first,
+				data.second);
+		};
+		builder->insufficientMediaAround()
+			| rpl::on_next(requestMediaAround)
+			| rpl::start(lifetime);
+
+		Auth().storage().sharedMediaSliceUpdated()
+			| rpl::on_next(applyUpdate)
+			| rpl::start(lifetime);
+		Auth().storage().sharedMediaOneRemoved()
+			| rpl::on_next(applyUpdate)
+			| rpl::start(lifetime);
+		Auth().storage().sharedMediaAllRemoved()
+			| rpl::on_next(applyUpdate)
+			| rpl::start(lifetime);
+
+		Auth().storage().query(Storage::SharedMediaQuery(
+			key,
+			limitBefore,
+			limitAfter
+		))
+			| rpl::on_next(applyUpdate)
+			| rpl::on_done([=] { builder->checkInsufficientMedia(); })
+			| rpl::start(lifetime);
+
+		return lifetime;
+	};
+}
+
+SharedMediaMergedSlice::SharedMediaMergedSlice(Key key) : SharedMediaMergedSlice(
 	key,
 	SharedMediaSlice(PartKey(key)),
 	MigratedSlice(key)) {
 }
 
-SharedMediaSliceMerged::SharedMediaSliceMerged(
+SharedMediaMergedSlice::SharedMediaMergedSlice(
 	Key key,
 	SharedMediaSlice part,
 	base::optional<SharedMediaSlice> migrated)
@@ -348,13 +490,13 @@ SharedMediaSliceMerged::SharedMediaSliceMerged(
 	, _migrated(std::move(migrated)) {
 }
 
-base::optional<int> SharedMediaSliceMerged::fullCount() const {
+base::optional<int> SharedMediaMergedSlice::fullCount() const {
 	return Add(
 		_part.fullCount(),
 		_migrated ? _migrated->fullCount() : 0);
 }
 
-base::optional<int> SharedMediaSliceMerged::skippedBefore() const {
+base::optional<int> SharedMediaMergedSlice::skippedBefore() const {
 	return Add(
 		isolatedInMigrated() ? 0 : _part.skippedBefore(),
 		_migrated
@@ -365,14 +507,14 @@ base::optional<int> SharedMediaSliceMerged::skippedBefore() const {
 	);
 }
 
-base::optional<int> SharedMediaSliceMerged::skippedAfter() const {
+base::optional<int> SharedMediaMergedSlice::skippedAfter() const {
 	return Add(
 		isolatedInMigrated() ? _part.fullCount() : _part.skippedAfter(),
 		isolatedInPart() ? 0 : _migrated->skippedAfter()
 	);
 }
 
-base::optional<int> SharedMediaSliceMerged::indexOf(FullMsgId fullId) const {
+base::optional<int> SharedMediaMergedSlice::indexOf(FullMsgId fullId) const {
 	return isFromPart(fullId)
 		? (_part.indexOf(fullId.msg) | func::add(migratedSize()))
 		: isolatedInPart()
@@ -382,12 +524,12 @@ base::optional<int> SharedMediaSliceMerged::indexOf(FullMsgId fullId) const {
 				: base::none;
 }
 
-int SharedMediaSliceMerged::size() const {
+int SharedMediaMergedSlice::size() const {
 	return (isolatedInPart() ? 0 : migratedSize())
 		+ (isolatedInMigrated() ? 0 : _part.size());
 }
 
-FullMsgId SharedMediaSliceMerged::operator[](int index) const {
+FullMsgId SharedMediaMergedSlice::operator[](int index) const {
 	Expects(index >= 0 && index < size());
 
 	if (auto size = migratedSize()) {
@@ -399,7 +541,7 @@ FullMsgId SharedMediaSliceMerged::operator[](int index) const {
 	return ComputeId(_part, index);
 }
 
-base::optional<int> SharedMediaSliceMerged::distance(const Key &a, const Key &b) const {
+base::optional<int> SharedMediaMergedSlice::distance(const Key &a, const Key &b) const {
 	if (a.type != _key.type
 		|| b.type != _key.type
 		|| a.peerId != _key.peerId
@@ -416,65 +558,79 @@ base::optional<int> SharedMediaSliceMerged::distance(const Key &a, const Key &b)
 	return base::none;
 }
 
-QString SharedMediaSliceMerged::debug() const {
+QString SharedMediaMergedSlice::debug() const {
 	return (_migrated ? (_migrated->debug() + '|') : QString()) + _part.debug();
 }
 
-SharedMediaViewerMerged::SharedMediaViewerMerged(
+SharedMediaMergedSliceBuilder::SharedMediaMergedSliceBuilder(Key key)
-	Key key,
-	int limitBefore,
-	int limitAfter)
 	: _key(key)
-	, _limitBefore(limitBefore)
+	, _part(SharedMediaMergedSlice::PartKey(_key))
-	, _limitAfter(limitAfter)
+	, _migrated(SharedMediaMergedSlice::MigratedSlice(_key)) {
-	, _part(SharedMediaSliceMerged::PartKey(_key), _limitBefore, _limitAfter)
+}
-	, _migrated(MigratedViewer(_key, _limitBefore, _limitAfter))
+
-	, _data(_key) {
+void SharedMediaMergedSliceBuilder::applyPartUpdate(SharedMediaSlice &&update) {
+	_part = std::move(update);
+}
+
+void SharedMediaMergedSliceBuilder::applyMigratedUpdate(SharedMediaSlice &&update) {
+	_migrated = std::move(update);
+}
+
+SharedMediaMergedSlice SharedMediaMergedSliceBuilder::snapshot() const {
+	return SharedMediaMergedSlice(
+		_key,
+		_part,
+		_migrated
+	);
+}
+
+rpl::producer<SharedMediaMergedSlice> SharedMediaMergedViewer(
+		SharedMediaMergedSlice::Key key,
+		int limitBefore,
+		int limitAfter) {
 	Expects(IsServerMsgId(key.universalId)
 		|| (key.universalId == 0)
 		|| (IsServerMsgId(-key.universalId) && key.migratedPeerId != 0));
 	Expects((key.universalId != 0) || (limitBefore == 0 && limitAfter == 0));
-}
 
-std::unique_ptr<SharedMediaViewer> SharedMediaViewerMerged::MigratedViewer(
+	return [=](auto consumer) {
-		const Key &key,
+		auto lifetime = rpl::lifetime();
-		int limitBefore,
+		auto builder = lifetime.make_state<SharedMediaMergedSliceBuilder>(key);
-		int limitAfter) {
+
-	return key.migratedPeerId
+		SharedMediaViewer(
-		? std::make_unique<SharedMediaViewer>(
+			SharedMediaMergedSlice::PartKey(key),
-			SharedMediaSliceMerged::MigratedKey(key),
 			limitBefore,
-			limitAfter)
+			limitAfter
-		: nullptr;
+		) | rpl::on_next([=](SharedMediaSlice &&update) {
+			builder->applyPartUpdate(std::move(update));
+			consumer.put_next(builder->snapshot());
+		}) | rpl::start(lifetime);
+
+		if (key.migratedPeerId) {
+			SharedMediaViewer(
+				SharedMediaMergedSlice::MigratedKey(key),
+				limitBefore,
+				limitAfter
+			) | rpl::on_next([=](SharedMediaSlice &&update) {
+				builder->applyMigratedUpdate(std::move(update));
+				consumer.put_next(builder->snapshot());
+			}) | rpl::start(lifetime);
+		}
+
+		return lifetime;
+	};
 }
 
-void SharedMediaViewerMerged::start() {
+SharedMediaWithLastSlice::SharedMediaWithLastSlice(Key key) : SharedMediaWithLastSlice(
-	subscribe(_part.updated, [this](const SharedMediaSlice &update) {
-		_data = SharedMediaSliceMerged(_key, update, std::move(_data._migrated));
-		updated.notify(_data);
-	});
-	if (_migrated) {
-		subscribe(_migrated->updated, [this](const SharedMediaSlice &update) {
-			_data = SharedMediaSliceMerged(_key, std::move(_data._part), update);
-			updated.notify(_data);
-		});
-	}
-	_part.start();
-	if (_migrated) {
-		_migrated->start();
-	}
-}
-
-SharedMediaSliceWithLast::SharedMediaSliceWithLast(Key key) : SharedMediaSliceWithLast(
 	key,
-	SharedMediaSliceMerged(ViewerKey(key)),
+	SharedMediaMergedSlice(ViewerKey(key)),
 	EndingSlice(key)) {
 }
 
-SharedMediaSliceWithLast::SharedMediaSliceWithLast(
+SharedMediaWithLastSlice::SharedMediaWithLastSlice(
 	Key key,
-	SharedMediaSliceMerged slice,
+	SharedMediaMergedSlice slice,
-	base::optional<SharedMediaSliceMerged> ending)
+	base::optional<SharedMediaMergedSlice> ending)
 	: _key(key)
 	, _slice(std::move(slice))
 	, _ending(std::move(ending))
@@ -484,17 +640,17 @@ SharedMediaSliceWithLast::SharedMediaSliceWithLast(
 		: false) {
 }
 
-base::optional<int> SharedMediaSliceWithLast::fullCount() const {
+base::optional<int> SharedMediaWithLastSlice::fullCount() const {
 	return Add(
 		_slice.fullCount(),
 		_isolatedLastPhoto | [](bool isolated) { return isolated ? 1 : 0; });
 }
 
-base::optional<int> SharedMediaSliceWithLast::skippedBefore() const {
+base::optional<int> SharedMediaWithLastSlice::skippedBefore() const {
 	return _slice.skippedBefore();
 }
 
-base::optional<int> SharedMediaSliceWithLast::skippedAfter() const {
+base::optional<int> SharedMediaWithLastSlice::skippedAfter() const {
 	return isolatedInSlice()
 		? Add(
 			_slice.skippedAfter(),
@@ -502,7 +658,7 @@ base::optional<int> SharedMediaSliceWithLast::skippedAfter() const {
 		: (lastPhotoSkip() | [](int) { return 0; });
 }
 
-base::optional<int> SharedMediaSliceWithLast::indexOf(Value value) const {
+base::optional<int> SharedMediaWithLastSlice::indexOf(Value value) const {
 	return base::get_if<FullMsgId>(&value)
 		? _slice.indexOf(*base::get_if<FullMsgId>(&value))
 		: (isolatedInSlice()
@@ -511,12 +667,12 @@ base::optional<int> SharedMediaSliceWithLast::indexOf(Value value) const {
 			: Add(_slice.size() - 1, lastPhotoSkip());
 }
 
-int SharedMediaSliceWithLast::size() const {
+int SharedMediaWithLastSlice::size() const {
 	return _slice.size()
 		+ ((!isolatedInSlice() && lastPhotoSkip() == 1) ? 1 : 0);
 }
 
-SharedMediaSliceWithLast::Value SharedMediaSliceWithLast::operator[](int index) const {
+SharedMediaWithLastSlice::Value SharedMediaWithLastSlice::operator[](int index) const {
 	Expects(index >= 0 && index < size());
 
 	return (index < _slice.size())
@@ -524,7 +680,7 @@ SharedMediaSliceWithLast::Value SharedMediaSliceWithLast::operator[](int index)
 		: Value(App::photo(_lastPhotoId));
 }
 
-base::optional<int> SharedMediaSliceWithLast::distance(const Key &a, const Key &b) const {
+base::optional<int> SharedMediaWithLastSlice::distance(const Key &a, const Key &b) const {
 	if (a.type != _key.type
 		|| b.type != _key.type
 		|| a.peerId != _key.peerId
@@ -541,22 +697,22 @@ base::optional<int> SharedMediaSliceWithLast::distance(const Key &a, const Key &
 	return base::none;
 }
 
-QString SharedMediaSliceWithLast::debug() const {
+QString SharedMediaWithLastSlice::debug() const {
 	return _slice.debug() + (_isolatedLastPhoto
 		? (*_isolatedLastPhoto ? "@" : "")
 		: "?");
 }
 
-PhotoId SharedMediaSliceWithLast::LastPeerPhotoId(PeerId peerId) {
+PhotoId SharedMediaWithLastSlice::LastPeerPhotoId(PeerId peerId) {
 	if (auto peer = App::peerLoaded(peerId)) {
 		return peer->photoId;
 	}
 	return UnknownPeerPhotoId;
 }
 
-base::optional<bool> SharedMediaSliceWithLast::IsLastIsolated(
+base::optional<bool> SharedMediaWithLastSlice::IsLastIsolated(
-		const SharedMediaSliceMerged &slice,
+		const SharedMediaMergedSlice &slice,
-		const base::optional<SharedMediaSliceMerged> &ending,
+		const base::optional<SharedMediaMergedSlice> &ending,
 		PhotoId lastPeerPhotoId) {
 	if (lastPeerPhotoId == UnknownPeerPhotoId) {
 		return base::none;
@@ -575,8 +731,8 @@ base::optional<bool> SharedMediaSliceWithLast::IsLastIsolated(
 		| [&](PhotoId photoId) { return lastPeerPhotoId != photoId; };
 }
 
-base::optional<FullMsgId> SharedMediaSliceWithLast::LastFullMsgId(
+base::optional<FullMsgId> SharedMediaWithLastSlice::LastFullMsgId(
-		const SharedMediaSliceMerged &slice) {
+		const SharedMediaMergedSlice &slice) {
 	if (slice.fullCount() == 0) {
 		return FullMsgId();
 	} else if (slice.size() == 0 || slice.skippedAfter() != 0) {
@@ -585,43 +741,59 @@ base::optional<FullMsgId> SharedMediaSliceWithLast::LastFullMsgId(
 	return slice[slice.size() - 1];
 }
 
-SharedMediaViewerWithLast::SharedMediaViewerWithLast(
+rpl::producer<SharedMediaWithLastSlice> SharedMediaWithLastViewer(
-	Key key,
+		SharedMediaWithLastSlice::Key key,
-	int limitBefore,
+		int limitBefore,
-	int limitAfter)
+		int limitAfter) {
+	return [=](auto consumer) {
+		auto lifetime = rpl::lifetime();
+		auto builder = lifetime.make_state<SharedMediaWithLastSliceBuilder>(key);
+
+		SharedMediaMergedViewer(
+			SharedMediaWithLastSlice::ViewerKey(key),
+			limitBefore,
+			limitAfter
+		) | rpl::on_next([=](SharedMediaMergedSlice &&update) {
+			builder->applyViewerUpdate(std::move(update));
+			consumer.put_next(builder->snapshot());
+		}) | rpl::start(lifetime);
+
+		if (base::get_if<SharedMediaWithLastSlice::MessageId>(&key.universalId)) {
+			SharedMediaMergedViewer(
+				SharedMediaWithLastSlice::EndingKey(key),
+				1,
+				1
+			) | rpl::on_next([=](SharedMediaMergedSlice &&update) {
+				builder->applyEndingUpdate(std::move(update));
+				consumer.put_next(builder->snapshot());
+			}) | rpl::start(lifetime);
+		}
+
+		return lifetime;
+	};
+}
+
+SharedMediaWithLastSliceBuilder::SharedMediaWithLastSliceBuilder(Key key)
 	: _key(key)
-	, _limitBefore(limitBefore)
-	, _limitAfter(limitAfter)
-	, _viewer(SharedMediaSliceWithLast::ViewerKey(_key), _limitBefore, _limitAfter)
-	, _ending(EndingViewer(_key, _limitBefore, _limitAfter))
 	, _data(_key) {
 }
 
-std::unique_ptr<SharedMediaViewerMerged> SharedMediaViewerWithLast::EndingViewer(
+void SharedMediaWithLastSliceBuilder::applyViewerUpdate(
-		const Key &key,
+		SharedMediaMergedSlice &&update) {
-		int limitBefore,
+	_data = SharedMediaWithLastSlice(
-		int limitAfter) {
+		_key,
-	return base::get_if<SharedMediaSliceWithLast::MessageId>(&key.universalId)
+		std::move(update),
-		? std::make_unique<SharedMediaViewerMerged>(
+		std::move(_data._ending));
-			SharedMediaSliceWithLast::EndingKey(key),
-			1,
-			1)
-		: nullptr;
 }
 
-void SharedMediaViewerWithLast::start() {
+void SharedMediaWithLastSliceBuilder::applyEndingUpdate(
-	subscribe(_viewer.updated, [this](const SharedMediaSliceMerged &update) {
+		SharedMediaMergedSlice &&update) {
-		_data = SharedMediaSliceWithLast(_key, update, std::move(_data._ending));
+	_data = SharedMediaWithLastSlice(
-		updated.notify(_data);
+		_key,
-	});
+		std::move(_data._slice),
-	if (_ending) {
+		std::move(update));
-		subscribe(_ending->updated, [this](const SharedMediaSliceMerged &update) {
+}
-			_data = SharedMediaSliceWithLast(_key, std::move(_data._slice), update);
+
-			updated.notify(_data);
+SharedMediaWithLastSlice SharedMediaWithLastSliceBuilder::snapshot() const {
-		});
+	return _data;
-	}
-	_viewer.start();
-	if (_ending) {
-		_ending->start();
-	}
 }

Telegram/SourceFiles/history/history_shared_media.h

@@ -29,13 +29,19 @@ void SharedMediaShowOverview(
 	Storage::SharedMediaType type,
 	not_null<History*> history);
 
-class SharedMediaViewer;
 class SharedMediaSlice {
 public:
+	using Type = Storage::SharedMediaType;
 	using Key = Storage::SharedMediaKey;
 
 	SharedMediaSlice(Key key);
-	SharedMediaSlice(Key key, base::optional<int> fullCount);
+	SharedMediaSlice(
+		Key key,
+		const base::flat_set<MsgId> &ids,
+		MsgRange range,
+		base::optional<int> fullCount,
+		base::optional<int> skippedBefore,
+		base::optional<int> skippedAfter);
 
 	const Key &key() const { return _key; }
 
@@ -57,56 +63,16 @@ private:
 	base::optional<int> _skippedBefore;
 	base::optional<int> _skippedAfter;
 
-	friend class SharedMediaViewer;
+	class SharedMediaSliceBuilder;
 
 };
 
-class SharedMediaViewer :
+rpl::producer<SharedMediaSlice> SharedMediaViewer(
-	private base::Subscriber,
+	SharedMediaSlice::Key key,
-	public base::enable_weak_from_this {
+	int limitBefore,
-public:
+	int limitAfter);
-	using Type = Storage::SharedMediaType;
-	using Key = Storage::SharedMediaKey;
 
-	SharedMediaViewer(Key key, int limitBefore, int limitAfter);
+class SharedMediaMergedSlice {
-
-	void start();
-
-	base::Observable<SharedMediaSlice> updated;
-
-private:
-	using InitialResult = Storage::SharedMediaResult;
-	using SliceUpdate = Storage::SharedMediaSliceUpdate;
-	using OneRemoved = Storage::SharedMediaRemoveOne;
-	using AllRemoved = Storage::SharedMediaRemoveAll;
-
-	void loadInitial();
-	enum class RequestDirection {
-		Before,
-		After,
-	};
-	void requestMessages(RequestDirection direction);
-	void applyStoredResult(InitialResult &&result);
-	void applyUpdate(const SliceUpdate &update);
-	void applyUpdate(const OneRemoved &update);
-	void applyUpdate(const AllRemoved &update);
-	void sliceToLimits();
-
-	void mergeSliceData(
-		base::optional<int> count,
-		const base::flat_set<MsgId> &messageIds,
-		base::optional<int> skippedBefore = base::none,
-		base::optional<int> skippedAfter = base::none);
-
-	Key _key;
-	int _limitBefore = 0;
-	int _limitAfter = 0;
-	SharedMediaSlice _data;
-
-};
-
-class SharedMediaViewerMerged;
-class SharedMediaSliceMerged {
 public:
 	using Type = Storage::SharedMediaType;
 	using UniversalMsgId = MsgId;
@@ -136,8 +102,8 @@ public:
 
 	};
 
-	SharedMediaSliceMerged(Key key);
+	SharedMediaMergedSlice(Key key);
-	SharedMediaSliceMerged(
+	SharedMediaMergedSlice(
 		Key key,
 		SharedMediaSlice part,
 		base::optional<SharedMediaSlice> migrated);
@@ -154,7 +120,6 @@ public:
 
 	QString debug() const;
 
-private:
 	static SharedMediaSlice::Key PartKey(const Key &key) {
 		return {
 			key.peerId,
@@ -169,6 +134,8 @@ private:
 			(key.universalId <= 0) ? (-key.universalId) : (ServerMaxMsgId - 1)
 		};
 	}
+
+private:
 	static base::optional<SharedMediaSlice> MigratedSlice(const Key &key) {
 		return key.migratedPeerId
 			? base::make_optional(SharedMediaSlice(MigratedKey(key)))
@@ -222,47 +189,22 @@ private:
 	SharedMediaSlice _part;
 	base::optional<SharedMediaSlice> _migrated;
 
-	friend class SharedMediaViewerMerged;
+	friend class SharedMediaMergedSliceBuilder;
 
 };
 
-class SharedMediaViewerMerged : private base::Subscriber {
+rpl::producer<SharedMediaMergedSlice> SharedMediaMergedViewer(
-public:
+	SharedMediaMergedSlice::Key key,
-	using Type = SharedMediaSliceMerged::Type;
+	int limitBefore,
-	using Key = SharedMediaSliceMerged::Key;
+	int limitAfter);
 
-	SharedMediaViewerMerged(
+class SharedMediaWithLastSlice {
-		Key key,
-		int limitBefore,
-		int limitAfter);
-
-	void start();
-
-	base::Observable<SharedMediaSliceMerged> updated;
-
-private:
-	static std::unique_ptr<SharedMediaViewer> MigratedViewer(
-		const Key &key,
-		int limitBefore,
-		int limitAfter);
-
-	Key _key;
-	int _limitBefore = 0;
-	int _limitAfter = 0;
-	SharedMediaViewer _part;
-	std::unique_ptr<SharedMediaViewer> _migrated;
-	SharedMediaSliceMerged _data;
-
-};
-
-class SharedMediaViewerWithLast;
-class SharedMediaSliceWithLast {
 public:
 	using Type = Storage::SharedMediaType;
 
 	// base::none in those mean CurrentPeerPhoto.
 	using Value = base::variant<FullMsgId, not_null<PhotoData*>>;
-	using MessageId = SharedMediaSliceMerged::UniversalMsgId;
+	using MessageId = SharedMediaMergedSlice::UniversalMsgId;
 	using UniversalMsgId = base::variant<
 		MessageId,
 		not_null<PhotoData*>>;
@@ -295,11 +237,11 @@ public:
 
 	};
 
-	SharedMediaSliceWithLast(Key key);
+	SharedMediaWithLastSlice(Key key);
-	SharedMediaSliceWithLast(
+	SharedMediaWithLastSlice(
 		Key key,
-		SharedMediaSliceMerged slice,
+		SharedMediaMergedSlice slice,
-		base::optional<SharedMediaSliceMerged> ending);
+		base::optional<SharedMediaMergedSlice> ending);
 
 	base::optional<int> fullCount() const;
 	base::optional<int> skippedBefore() const;
@@ -311,18 +253,17 @@ public:
 
 	QString debug() const;
 
-private:
+	static SharedMediaMergedSlice::Key ViewerKey(const Key &key) {
-	static SharedMediaSliceMerged::Key ViewerKey(const Key &key) {
 		return {
 			key.peerId,
 			key.migratedPeerId,
 			key.type,
 			base::get_if<MessageId>(&key.universalId)
-				? (*base::get_if<MessageId>(&key.universalId))
+			? (*base::get_if<MessageId>(&key.universalId))
-				: ServerMaxMsgId - 1
+			: ServerMaxMsgId - 1
 		};
 	}
-	static SharedMediaSliceMerged::Key EndingKey(const Key &key) {
+	static SharedMediaMergedSlice::Key EndingKey(const Key &key) {
 		return {
 			key.peerId,
 			key.migratedPeerId,
@@ -330,19 +271,21 @@ private:
 			ServerMaxMsgId - 1
 		};
 	}
-	static base::optional<SharedMediaSliceMerged> EndingSlice(const Key &key) {
+
+private:
+	static base::optional<SharedMediaMergedSlice> EndingSlice(const Key &key) {
 		return base::get_if<MessageId>(&key.universalId)
-			? base::make_optional(SharedMediaSliceMerged(EndingKey(key)))
+			? base::make_optional(SharedMediaMergedSlice(EndingKey(key)))
 			: base::none;
 	}
 
 	static PhotoId LastPeerPhotoId(PeerId peerId);
 	static base::optional<bool> IsLastIsolated(
-		const SharedMediaSliceMerged &slice,
+		const SharedMediaMergedSlice &slice,
-		const base::optional<SharedMediaSliceMerged> &ending,
+		const base::optional<SharedMediaMergedSlice> &ending,
 		PhotoId lastPeerPhotoId);
 	static base::optional<FullMsgId> LastFullMsgId(
-		const SharedMediaSliceMerged &slice);
+		const SharedMediaMergedSlice &slice);
 	static base::optional<int> Add(
 			const base::optional<int> &a,
 			const base::optional<int> &b) {
@@ -371,40 +314,16 @@ private:
 	}
 
 	Key _key;
-	SharedMediaSliceMerged _slice;
+	SharedMediaMergedSlice _slice;
-	base::optional<SharedMediaSliceMerged> _ending;
+	base::optional<SharedMediaMergedSlice> _ending;
 	PhotoId _lastPhotoId = 0;
 	base::optional<bool> _isolatedLastPhoto;
 
-	friend class SharedMediaViewerWithLast;
+	friend class SharedMediaWithLastSliceBuilder;
 
 };
 
-class SharedMediaViewerWithLast : private base::Subscriber {
+rpl::producer<SharedMediaWithLastSlice> SharedMediaWithLastViewer(
-public:
+	SharedMediaWithLastSlice::Key key,
-	using Type = SharedMediaSliceWithLast::Type;
+	int limitBefore,
-	using Key = SharedMediaSliceWithLast::Key;
+	int limitAfter);
-
-	SharedMediaViewerWithLast(
-		Key key,
-		int limitBefore,
-		int limitAfter);
-
-	void start();
-
-	base::Observable<SharedMediaSliceWithLast> updated;
-
-private:
-	static std::unique_ptr<SharedMediaViewerMerged> EndingViewer(
-		const Key &key,
-		int limitBefore,
-		int limitAfter);
-
-	Key _key;
-	int _limitBefore = 0;
-	int _limitAfter = 0;
-	SharedMediaViewerMerged _viewer;
-	std::unique_ptr<SharedMediaViewerMerged> _ending;
-	SharedMediaSliceWithLast _data;
-
-};

Telegram/SourceFiles/history/history_user_photos.cpp

@@ -25,6 +25,41 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #include "storage/storage_facade.h"
 #include "storage/storage_user_photos.h"
 
+class UserPhotosSliceBuilder {
+public:
+	using Key = UserPhotosSlice::Key;
+
+	UserPhotosSliceBuilder(Key key, int limitBefore, int limitAfter);
+
+	bool applyUpdate(const Storage::UserPhotosResult &update);
+	bool applyUpdate(const Storage::UserPhotosSliceUpdate &update);
+	void checkInsufficientPhotos();
+	rpl::producer<PhotoId> insufficientPhotosAround() const {
+		return _insufficientPhotosAround.events();
+	}
+
+	UserPhotosSlice snapshot() const;
+
+private:
+	void mergeSliceData(
+		base::optional<int> count,
+		const std::deque<PhotoId> &photoIds,
+		base::optional<int> skippedBefore,
+		int skippedAfter);
+	void sliceToLimits();
+
+	Key _key;
+	std::deque<PhotoId> _ids;
+	base::optional<int> _fullCount;
+	base::optional<int> _skippedBefore;
+	int _skippedAfter = 0;
+	int _limitBefore = 0;
+	int _limitAfter = 0;
+
+	rpl::event_stream<PhotoId> _insufficientPhotosAround;
+
+};
+
 UserPhotosSlice::UserPhotosSlice(Key key) : UserPhotosSlice(
 	key,
 	{},
@@ -90,41 +125,6 @@ QString UserPhotosSlice::debug() const {
 	return before + middle + after;
 }
 
-class UserPhotosSliceBuilder {
-public:
-	using Key = UserPhotosSlice::Key;
-
-	UserPhotosSliceBuilder(Key key, int limitBefore, int limitAfter);
-
-	bool applyUpdate(const Storage::UserPhotosResult &update);
-	bool applyUpdate(const Storage::UserPhotosSliceUpdate &update);
-	void checkInsufficientPhotos();
-	rpl::producer<PhotoId> insufficientPhotosAround() const {
-		return _insufficientPhotosAround.events();
-	}
-
-	UserPhotosSlice snapshot() const;
-
-private:
-	void mergeSliceData(
-		base::optional<int> count,
-		const std::deque<PhotoId> &photoIds,
-		base::optional<int> skippedBefore,
-		int skippedAfter);
-	void sliceToLimits();
-
-	Key _key;
-	std::deque<PhotoId> _ids;
-	base::optional<int> _fullCount;
-	base::optional<int> _skippedBefore;
-	int _skippedAfter = 0;
-	int _limitBefore = 0;
-	int _limitAfter = 0;
-
-	rpl::event_stream<PhotoId> _insufficientPhotosAround;
-
-};
-
 UserPhotosSliceBuilder::UserPhotosSliceBuilder(
 	Key key,
 	int limitBefore,

Telegram/SourceFiles/mediaview.cpp

@@ -68,18 +68,15 @@ constexpr auto kIdsPreloadAfter = 28;
 } // namespace
 
 struct MediaView::SharedMedia {
-	SharedMedia(SharedMediaViewerWithLast::Key key)
+	SharedMedia(SharedMediaWithLastSlice::Key key) : key(key) {
-		: key(key)
-		, slice(key, kIdsLimit, kIdsLimit) {
 	}
 
-	SharedMediaViewerWithLast::Key key;
+	SharedMediaWithLastSlice::Key key;
-	SharedMediaViewerWithLast slice;
+	rpl::lifetime lifetime;
 };
 
 struct MediaView::UserPhotos {
-	UserPhotos(UserPhotosSlice::Key key)
+	UserPhotos(UserPhotosSlice::Key key) : key(key) {
-		: key(key) {
 	}
 
 	UserPhotosSlice::Key key;
@@ -1044,7 +1041,7 @@ bool MediaView::validSharedMedia() const {
 			return false;
 		}
 		auto countDistanceInData = [](const auto &a, const auto &b) {
-			return [&](const SharedMediaSliceWithLast &data) {
+			return [&](const SharedMediaWithLastSlice &data) {
 				return data.distance(a, b);
 			};
 		};
@@ -1063,21 +1060,24 @@ bool MediaView::validSharedMedia() const {
 void MediaView::validateSharedMedia() {
 	if (auto key = sharedMediaKey()) {
 		_sharedMedia = std::make_unique<SharedMedia>(*key);
-		subscribe(_sharedMedia->slice.updated, [this](const SharedMediaSliceWithLast &data) {
+		SharedMediaWithLastViewer(
-			handleSharedMediaUpdate(data);
+			*key,
-		});
+			kIdsLimit,
-		_sharedMedia->slice.start();
+			kIdsLimit
+		) | rpl::on_next([this](SharedMediaWithLastSlice &&update) {
+			handleSharedMediaUpdate(std::move(update));
+		}) | rpl::start(_sharedMedia->lifetime);
 	} else {
 		_sharedMedia = nullptr;
 		_sharedMediaData = base::none;
 	}
 }
 
-void MediaView::handleSharedMediaUpdate(const SharedMediaSliceWithLast &update) {
+void MediaView::handleSharedMediaUpdate(SharedMediaWithLastSlice &&update) {
 	if ((!_photo && !_doc) || !_sharedMedia) {
 		_sharedMediaData = base::none;
 	} else {
-		_sharedMediaData = update;
+		_sharedMediaData = std::move(update);
 	}
 	findCurrent();
 	updateControls();

Telegram/SourceFiles/mediaview.h

@@ -169,13 +169,13 @@ private:
 	void updateMixerVideoVolume() const;
 
 	struct SharedMedia;
-	using SharedMediaType = SharedMediaViewerWithLast::Type;
+	using SharedMediaType = SharedMediaWithLastSlice::Type;
-	using SharedMediaKey = SharedMediaViewerWithLast::Key;
+	using SharedMediaKey = SharedMediaWithLastSlice::Key;
 	base::optional<SharedMediaType> sharedMediaType() const;
 	base::optional<SharedMediaKey> sharedMediaKey() const;
 	bool validSharedMedia() const;
 	void validateSharedMedia();
-	void handleSharedMediaUpdate(const SharedMediaSliceWithLast &update);
+	void handleSharedMediaUpdate(SharedMediaWithLastSlice &&update);
 
 	struct UserPhotos;
 	using UserPhotosKey = UserPhotosSlice::Key;
@@ -251,7 +251,7 @@ private:
 	PhotoData *_photo = nullptr;
 	DocumentData *_doc = nullptr;
 	std::unique_ptr<SharedMedia> _sharedMedia;
-	base::optional<SharedMediaSliceWithLast> _sharedMediaData;
+	base::optional<SharedMediaWithLastSlice> _sharedMediaData;
 	std::unique_ptr<UserPhotos> _userPhotos;
 	base::optional<UserPhotosSlice> _userPhotosData;
 

Telegram/SourceFiles/storage/storage_facade.cpp

@@ -32,13 +32,11 @@ public:
 	void add(SharedMediaAddSlice &&query);
 	void remove(SharedMediaRemoveOne &&query);
 	void remove(SharedMediaRemoveAll &&query);
-	void query(
+	rpl::producer<SharedMediaResult> query(SharedMediaQuery &&query) const;
-		SharedMediaQuery &&query,
-		base::lambda_once<void(SharedMediaResult&&)> &&callback);
 
-	base::Observable<SharedMediaSliceUpdate> &sharedMediaSliceUpdated();
+	rpl::producer<SharedMediaSliceUpdate> sharedMediaSliceUpdated() const;
-	base::Observable<SharedMediaRemoveOne> &sharedMediaOneRemoved();
+	rpl::producer<SharedMediaRemoveOne> sharedMediaOneRemoved() const;
-	base::Observable<SharedMediaRemoveAll> &sharedMediaAllRemoved();
+	rpl::producer<SharedMediaRemoveAll> sharedMediaAllRemoved() const;
 
 	void add(UserPhotosAddNew &&query);
 	void add(UserPhotosAddSlice &&query);
@@ -74,22 +72,20 @@ void Facade::Impl::remove(SharedMediaRemoveAll &&query) {
 	_sharedMedia.remove(std::move(query));
 }
 
-void Facade::Impl::query(
+rpl::producer<SharedMediaResult> Facade::Impl::query(SharedMediaQuery &&query) const {
-		SharedMediaQuery &&query,
+	return _sharedMedia.query(std::move(query));
-		base::lambda_once<void(SharedMediaResult&&)> &&callback) {
-	_sharedMedia.query(query, std::move(callback));
 }
 
-base::Observable<SharedMediaSliceUpdate> &Facade::Impl::sharedMediaSliceUpdated() {
+rpl::producer<SharedMediaSliceUpdate> Facade::Impl::sharedMediaSliceUpdated() const {
-	return _sharedMedia.sliceUpdated;
+	return _sharedMedia.sliceUpdated();
 }
 
-base::Observable<SharedMediaRemoveOne> &Facade::Impl::sharedMediaOneRemoved() {
+rpl::producer<SharedMediaRemoveOne> Facade::Impl::sharedMediaOneRemoved() const {
-	return _sharedMedia.oneRemoved;
+	return _sharedMedia.oneRemoved();
 }
 
-base::Observable<SharedMediaRemoveAll> &Facade::Impl::sharedMediaAllRemoved() {
+rpl::producer<SharedMediaRemoveAll> Facade::Impl::sharedMediaAllRemoved() const {
-	return _sharedMedia.allRemoved;
+	return _sharedMedia.allRemoved();
 }
 
 void Facade::Impl::add(UserPhotosAddNew &&query) {
@@ -139,21 +135,19 @@ void Facade::remove(SharedMediaRemoveAll &&query) {
 	_impl->remove(std::move(query));
 }
 
-void Facade::query(
+rpl::producer<SharedMediaResult> Facade::query(SharedMediaQuery &&query) const {
-		SharedMediaQuery &&query,
+	return _impl->query(std::move(query));
-		base::lambda_once<void(SharedMediaResult&&)> &&callback) {
-	_impl->query(std::move(query), std::move(callback));
 }
 
-base::Observable<SharedMediaSliceUpdate> &Facade::sharedMediaSliceUpdated() {
+rpl::producer<SharedMediaSliceUpdate> Facade::sharedMediaSliceUpdated() const {
 	return _impl->sharedMediaSliceUpdated();
 }
 
-base::Observable<SharedMediaRemoveOne> &Facade::sharedMediaOneRemoved() {
+rpl::producer<SharedMediaRemoveOne> Facade::sharedMediaOneRemoved() const {
 	return _impl->sharedMediaOneRemoved();
 }
 
-base::Observable<SharedMediaRemoveAll> &Facade::sharedMediaAllRemoved() {
+rpl::producer<SharedMediaRemoveAll> Facade::sharedMediaAllRemoved() const {
 	return _impl->sharedMediaAllRemoved();
 }
 

Telegram/SourceFiles/storage/storage_facade.h

@@ -51,13 +51,11 @@ public:
 	void add(SharedMediaAddSlice &&query);
 	void remove(SharedMediaRemoveOne &&query);
 	void remove(SharedMediaRemoveAll &&query);
-	void query(
-		SharedMediaQuery &&query,
-		base::lambda_once<void(SharedMediaResult&&)> &&callback);
 
-	base::Observable<SharedMediaSliceUpdate> &sharedMediaSliceUpdated();
+	rpl::producer<SharedMediaResult> query(SharedMediaQuery &&query) const;
-	base::Observable<SharedMediaRemoveOne> &sharedMediaOneRemoved();
+	rpl::producer<SharedMediaSliceUpdate> sharedMediaSliceUpdated() const;
-	base::Observable<SharedMediaRemoveAll> &sharedMediaAllRemoved();
+	rpl::producer<SharedMediaRemoveOne> sharedMediaOneRemoved() const;
+	rpl::producer<SharedMediaRemoveAll> sharedMediaAllRemoved() const;
 
 	void add(UserPhotosAddNew &&query);
 	void add(UserPhotosAddSlice &&query);

Telegram/SourceFiles/storage/storage_shared_media.cpp

@@ -128,7 +128,7 @@ void SharedMedia::List::addRange(
 		}
 	}
 	update.count = _count;
-	sliceUpdated.notify(update, true);
+	_sliceUpdated.fire(std::move(update));
 }
 
 void SharedMedia::List::addNew(MsgId messageId) {
@@ -171,33 +171,28 @@ void SharedMedia::List::removeAll() {
 	_count = 0;
 }
 
-void SharedMedia::List::query(
+rpl::producer<SharedMediaResult> SharedMedia::List::query(
-		const SharedMediaQuery &query,
+		SharedMediaQuery &&query) const {
-		base::lambda_once<void(SharedMediaResult&&)> &&callback) {
+	return [this, query = std::move(query)](auto consumer) {
-	auto result = SharedMediaResult {};
+		auto slice = base::lower_bound(
-	result.count = _count;
+			_slices,
-
+			query.key.messageId,
-	auto slice = base::lower_bound(
+			[](const Slice &slice, MsgId id) { return slice.range.till < id; });
-		_slices,
+		if (slice != _slices.end() && slice->range.from <= query.key.messageId) {
-		query.key.messageId,
+			consumer.put_next(queryFromSlice(query, *slice));
-		[](const Slice &slice, MsgId id) { return slice.range.till < id; });
+		} else if (_count) {
-	if (slice != _slices.end() && slice->range.from <= query.key.messageId) {
+			auto result = SharedMediaResult {};
-		result = queryFromSlice(query, *slice);
+			result.count = _count;
-	} else {
+			consumer.put_next(std::move(result));
-		result.count = _count;
+		}
-	}
+		consumer.put_done();
-	base::TaskQueue::Main().Put(
+		return rpl::lifetime();
-		[
+	};
-			callback = std::move(callback),
-			result = std::move(result)
-		]() mutable {
-		callback(std::move(result));
-	});
 }
 
 SharedMediaResult SharedMedia::List::queryFromSlice(
 		const SharedMediaQuery &query,
-		const Slice &slice) {
+		const Slice &slice) const {
 	auto result = SharedMediaResult {};
 	auto position = base::lower_bound(slice.messages, query.key.messageId);
 	auto haveBefore = int(position - slice.messages.begin());
@@ -237,14 +232,17 @@ std::map<PeerId, SharedMedia::Lists>::iterator
 	for (auto index = 0; index != kSharedMediaTypeCount; ++index) {
 		auto &list = result->second[index];
 		auto type = static_cast<SharedMediaType>(index);
-		subscribe(list.sliceUpdated, [this, type, peer](const SliceUpdate &update) {
+
-			sliceUpdated.notify(SharedMediaSliceUpdate(
+		list.sliceUpdated()
-				peer,
+			| rpl::on_next([this, peer, type](SliceUpdate &&update) {
-				type,
+				_sliceUpdated.fire(SharedMediaSliceUpdate(
-				update.messages,
+					peer,
-				update.range,
+					type,
-				update.count), true);
+					update.messages,
-		});
+					update.range,
+					update.count));
+			})
+			| rpl::start(_lifetime);
 	}
 	return result;
 }
@@ -284,7 +282,7 @@ void SharedMedia::remove(SharedMediaRemoveOne &&query) {
 			auto type = static_cast<SharedMediaType>(index);
 			if (query.types.test(type)) {
 				peerIt->second[index].removeOne(query.messageId);
-				oneRemoved.notify(query, true);
+				_oneRemoved.fire(std::move(query));
 			}
 		}
 	}
@@ -296,26 +294,21 @@ void SharedMedia::remove(SharedMediaRemoveAll &&query) {
 		for (auto index = 0; index != kSharedMediaTypeCount; ++index) {
 			peerIt->second[index].removeAll();
 		}
-		allRemoved.notify(query, true);
+		_allRemoved.fire(std::move(query));
 	}
 }
 
-void SharedMedia::query(
+rpl::producer<SharedMediaResult> SharedMedia::query(SharedMediaQuery &&query) const {
-		const SharedMediaQuery &query,
-		base::lambda_once<void(SharedMediaResult&&)> &&callback) {
 	Expects(IsValidSharedMediaType(query.key.type));
 	auto peerIt = _lists.find(query.key.peerId);
 	if (peerIt != _lists.end()) {
 		auto index = static_cast<int>(query.key.type);
-		peerIt->second[index].query(query, std::move(callback));
+		return peerIt->second[index].query(std::move(query));
-	} else {
-		base::TaskQueue::Main().Put(
-			[
-				callback = std::move(callback)
-			]() mutable {
-			callback(SharedMediaResult());
-		});
 	}
+	return [](auto consumer) {
+		consumer.put_done();
+		return rpl::lifetime();
+	};
 }
 
 } // namespace Storage

Telegram/SourceFiles/storage/storage_shared_media.h

@@ -21,6 +21,7 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
 #pragma once
 
 #include "storage/storage_facade.h"
+#include "rpl/event_stream.h"
 
 namespace Storage {
 
@@ -191,7 +192,7 @@ struct SharedMediaSliceUpdate {
 	base::optional<int> count;
 };
 
-class SharedMedia : private base::Subscriber {
+class SharedMedia {
 public:
 	using Type = SharedMediaType;
 
@@ -200,13 +201,18 @@ public:
 	void add(SharedMediaAddSlice &&query);
 	void remove(SharedMediaRemoveOne &&query);
 	void remove(SharedMediaRemoveAll &&query);
-	void query(
-		const SharedMediaQuery &query,
-		base::lambda_once<void(SharedMediaResult&&)> &&callback);
 
-	base::Observable<SharedMediaSliceUpdate> sliceUpdated;
+	rpl::producer<SharedMediaResult> query(SharedMediaQuery &&query) const;
-	base::Observable<SharedMediaRemoveOne> oneRemoved;
+
-	base::Observable<SharedMediaRemoveAll> allRemoved;
+	rpl::producer<SharedMediaSliceUpdate> sliceUpdated() const {
+		return _sliceUpdated.events();
+	}
+	rpl::producer<SharedMediaRemoveOne> oneRemoved() const {
+		return _oneRemoved.events();
+	}
+	rpl::producer<SharedMediaRemoveAll> allRemoved() const {
+		return _allRemoved.events();
+	}
 
 private:
 	class List {
@@ -219,16 +225,16 @@ private:
 			base::optional<int> count);
 		void removeOne(MsgId messageId);
 		void removeAll();
-		void query(
+		rpl::producer<SharedMediaResult> query(SharedMediaQuery &&query) const;
-			const SharedMediaQuery &query,
-			base::lambda_once<void(SharedMediaResult&&)> &&callback);
 
 		struct SliceUpdate {
 			const base::flat_set<MsgId> *messages = nullptr;
 			MsgRange range;
 			base::optional<int> count;
 		};
-		base::Observable<SliceUpdate> sliceUpdated;
+		rpl::producer<SliceUpdate> sliceUpdated() const {
+			return _sliceUpdated.events();
+		}
 
 	private:
 		struct Slice {
@@ -267,11 +273,13 @@ private:
 
 		SharedMediaResult queryFromSlice(
 			const SharedMediaQuery &query,
-			const Slice &slice);
+			const Slice &slice) const;
 
 		base::optional<int> _count;
 		base::flat_set<Slice> _slices;
 
+		rpl::event_stream<SliceUpdate> _sliceUpdated;
+
 	};
 	using SliceUpdate = List::SliceUpdate;
 	using Lists = std::array<List, kSharedMediaTypeCount>;
@@ -280,6 +288,11 @@ private:
 
 	std::map<PeerId, Lists> _lists;
 
+	rpl::lifetime _lifetime;
+	rpl::event_stream<SharedMediaSliceUpdate> _sliceUpdated;
+	rpl::event_stream<SharedMediaRemoveOne> _oneRemoved;
+	rpl::event_stream<SharedMediaRemoveAll> _allRemoved;
+
 };
 
 } // namespace Storage