From 432072a3d41d762f4055cc75b88bc9f14a10bba3 Mon Sep 17 00:00:00 2001 From: John Preston Date: Thu, 9 Apr 2026 14:00:06 +0700 Subject: [PATCH] Stream several messages together. --- Telegram/SourceFiles/data/data_session.cpp | 15 +++ Telegram/SourceFiles/history/history.cpp | 7 +- Telegram/SourceFiles/history/history.h | 1 + Telegram/SourceFiles/history/history_item.cpp | 4 + Telegram/SourceFiles/history/history_item.h | 1 + .../history/history_streamed_drafts.cpp | 107 ++++++++++++++---- .../history/history_streamed_drafts.h | 12 +- 7 files changed, 116 insertions(+), 31 deletions(-) diff --git a/Telegram/SourceFiles/data/data_session.cpp b/Telegram/SourceFiles/data/data_session.cpp index a3704e2d37..8667b03fb0 100644 --- a/Telegram/SourceFiles/data/data_session.cpp +++ b/Telegram/SourceFiles/data/data_session.cpp @@ -30,6 +30,7 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL #include "history/history.h" #include "history/history_item.h" #include "history/history_item_components.h" +#include "history/history_streamed_drafts.h" #include "history/view/media/history_view_media.h" #include "history/view/history_view_element.h" #include "inline_bots/inline_bot_layout_item.h" @@ -3107,6 +3108,20 @@ HistoryItem *Session::addNewMessage( return nullptr; } + if (data.type() == mtpc_message) { + if (const auto h = historyLoaded(peerId)) { + if (const auto streamed = h->streamedDraftsIfExists()) { + if (const auto adopted = streamed->adoptIncoming( + data.c_message())) { + if (type == NewMessageType::Unread) { + CheckForSwitchInlineButton(adopted); + } + return adopted; + } + } + } + } + const auto result = history(peerId)->addNewMessage( id, data, diff --git a/Telegram/SourceFiles/history/history.cpp b/Telegram/SourceFiles/history/history.cpp index 497c1f91c4..930ce1c01f 100644 --- a/Telegram/SourceFiles/history/history.cpp +++ b/Telegram/SourceFiles/history/history.cpp @@ -1594,9 +1594,6 @@ void History::newItemAdded(not_null item) { if (const auto sublist = item->savedSublist()) { sublist->applyItemAdded(item); } - if (const auto streamed = _streamedDrafts.get()) { - streamed->applyItemAdded(item); - } if (const auto media = item->media()) { if (const auto gift = media->gift()) { if (const auto unique = gift->unique.get()) { @@ -3958,6 +3955,10 @@ HistoryStreamedDrafts &History::streamedDrafts() { return *_streamedDrafts; } +HistoryStreamedDrafts *History::streamedDraftsIfExists() const { + return _streamedDrafts.get(); +} + HistoryItem *History::joinedMessageInstance() const { return _joinedMessage; } diff --git a/Telegram/SourceFiles/history/history.h b/Telegram/SourceFiles/history/history.h index e8848ecae3..f85d8daa21 100644 --- a/Telegram/SourceFiles/history/history.h +++ b/Telegram/SourceFiles/history/history.h @@ -93,6 +93,7 @@ public: [[nodiscard]] Data::HistoryMessages *maybeMessages(); [[nodiscard]] HistoryStreamedDrafts &streamedDrafts(); + [[nodiscard]] HistoryStreamedDrafts *streamedDraftsIfExists() const; [[nodiscard]] HistoryItem *joinedMessageInstance() const; void checkLocalMessages(); diff --git a/Telegram/SourceFiles/history/history_item.cpp b/Telegram/SourceFiles/history/history_item.cpp index e698a95d65..2b56d44992 100644 --- a/Telegram/SourceFiles/history/history_item.cpp +++ b/Telegram/SourceFiles/history/history_item.cpp @@ -2711,6 +2711,10 @@ QString HistoryItem::notificationHeader() const { return QString(); } +void HistoryItem::markBeingSentForAdoption() { + _flags |= MessageFlag::BeingSent; +} + void HistoryItem::setRealId(MsgId newId) { Expects(_flags & MessageFlag::BeingSent); Expects(IsClientMsgId(id)); diff --git a/Telegram/SourceFiles/history/history_item.h b/Telegram/SourceFiles/history/history_item.h index f4c9b6d99d..9d30f6d271 100644 --- a/Telegram/SourceFiles/history/history_item.h +++ b/Telegram/SourceFiles/history/history_item.h @@ -440,6 +440,7 @@ public: bool isForumPost); void setPostAuthor(const QString &author); void setRealId(MsgId newId); + void markBeingSentForAdoption(); void incrementReplyToTopCounter(); void applyEffectWatchedOnUnreadKnown(); diff --git a/Telegram/SourceFiles/history/history_streamed_drafts.cpp b/Telegram/SourceFiles/history/history_streamed_drafts.cpp index e88d1b612e..ef67083543 100644 --- a/Telegram/SourceFiles/history/history_streamed_drafts.cpp +++ b/Telegram/SourceFiles/history/history_streamed_drafts.cpp @@ -9,6 +9,7 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL #include "api/api_text_entities.h" #include "data/data_forum_topic.h" +#include "data/data_peer_id.h" #include "data/data_session.h" #include "history/history.h" #include "history/history_item.h" @@ -17,6 +18,15 @@ namespace { constexpr auto kClearTimeout = 30 * crl::time(1000); +[[nodiscard]] int CommonPrefixLength(const QString &a, const QString &b) { + const auto count = std::min(a.size(), b.size()); + auto i = 0; + while (i < count && a[i] == b[i]) { + ++i; + } + return i; +} + } // namespace HistoryStreamedDrafts::HistoryStreamedDrafts(not_null history) @@ -25,7 +35,7 @@ HistoryStreamedDrafts::HistoryStreamedDrafts(not_null history) } HistoryStreamedDrafts::~HistoryStreamedDrafts() { - for (const auto &[rootId, draft] : base::take(_drafts)) { + for (const auto &[randomId, draft] : base::take(_drafts)) { draft.message->destroy(); } } @@ -41,19 +51,18 @@ void HistoryStreamedDrafts::apply( if (!rootId) { rootId = Data::ForumTopic::kGeneralId; } + const auto randomId = data.vrandom_id().v; if (!when) { - clear(rootId); + clearByRandomId(randomId); return; } const auto text = Api::ParseTextWithEntities( &_history->session(), data.vtext()); - const auto randomId = data.vrandom_id().v; - if (update(rootId, randomId, text)) { + if (update(randomId, text)) { return; } - clear(rootId); - _drafts.emplace(rootId, Draft{ + _drafts.emplace(randomId, Draft{ .message = _history->addNewLocalMessage({ .id = _history->owner().nextLocalMessageId(), .flags = (MessageFlag::Local @@ -66,7 +75,8 @@ void HistoryStreamedDrafts::apply( }, .date = when, }, text, MTP_messageMediaEmpty()), - .randomId = randomId, + .rootId = rootId, + .fromId = fromId, .updated = crl::now(), }); if (!_checkTimer.isActive()) { @@ -75,11 +85,10 @@ void HistoryStreamedDrafts::apply( } bool HistoryStreamedDrafts::update( - MsgId rootId, uint64 randomId, const TextWithEntities &text) { - const auto i = _drafts.find(rootId); - if (i == end(_drafts) || i->second.randomId != randomId) { + const auto i = _drafts.find(randomId); + if (i == end(_drafts)) { return false; } i->second.message->setText(text); @@ -87,8 +96,8 @@ bool HistoryStreamedDrafts::update( return true; } -void HistoryStreamedDrafts::clear(MsgId rootId) { - if (const auto draft = _drafts.take(rootId)) { +void HistoryStreamedDrafts::clearByRandomId(uint64 randomId) { + if (const auto draft = _drafts.take(randomId)) { draft->message->destroy(); } if (_drafts.empty()) { @@ -98,18 +107,13 @@ void HistoryStreamedDrafts::clear(MsgId rootId) { bool HistoryStreamedDrafts::hasFor(not_null item) const { const auto rootId = item->topicRootId(); - const auto i = _drafts.find(rootId); - return (i != end(_drafts)) - && (i->second.message->from() == item->from()); -} - -void HistoryStreamedDrafts::applyItemAdded(not_null item) { - const auto rootId = item->topicRootId(); - const auto i = _drafts.find(rootId); - if (i == end(_drafts) || i->second.message->from() != item->from()) { - return; + const auto from = item->from(); + for (const auto &[randomId, draft] : _drafts) { + if (draft.rootId == rootId && draft.message->from() == from) { + return true; + } } - clear(rootId); + return false; } void HistoryStreamedDrafts::applyItemRemoved(not_null item) { @@ -124,6 +128,63 @@ void HistoryStreamedDrafts::applyItemRemoved(not_null item) { } } +HistoryItem *HistoryStreamedDrafts::adoptIncoming( + const MTPDmessage &data) { + if (_drafts.empty()) { + return nullptr; + } + const auto fromId = data.vfrom_id() + ? peerFromMTP(*data.vfrom_id()) + : _history->peer->id; + auto rootId = MsgId(0); + if (const auto reply = data.vreply_to()) { + reply->match([&](const MTPDmessageReplyHeader &d) { + if (d.is_forum_topic()) { + rootId = d.vreply_to_top_id().value_or_empty(); + if (!rootId) { + rootId = d.vreply_to_msg_id().value_or_empty(); + } + } + }, [](const MTPDmessageReplyStoryHeader &) {}); + } + if (!rootId) { + rootId = Data::ForumTopic::kGeneralId; + } + const auto incomingText = qs(data.vmessage()); + auto best = end(_drafts); + auto bestPrefix = 0; + for (auto i = begin(_drafts); i != end(_drafts); ++i) { + const auto &draft = i->second; + if (draft.rootId != rootId) { + continue; + } + if (draft.message->from()->id != fromId) { + continue; + } + const auto prefix = CommonPrefixLength( + draft.message->originalText().text, + incomingText); + if (prefix > bestPrefix) { + bestPrefix = prefix; + best = i; + } + } + if (best == end(_drafts) || bestPrefix <= 0) { + return nullptr; + } + const auto item = best->second.message.get(); + _drafts.erase(best); + + item->markBeingSentForAdoption(); + item->setRealId(data.vid().v); + item->applySentMessage(data); + + if (_drafts.empty()) { + scheduleDestroy(); + } + return item; +} + void HistoryStreamedDrafts::check() { auto closest = crl::time(); const auto now = crl::now(); diff --git a/Telegram/SourceFiles/history/history_streamed_drafts.h b/Telegram/SourceFiles/history/history_streamed_drafts.h index d521ccc80c..c46f6d102b 100644 --- a/Telegram/SourceFiles/history/history_streamed_drafts.h +++ b/Telegram/SourceFiles/history/history_streamed_drafts.h @@ -11,6 +11,7 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL #include "base/weak_ptr.h" class History; +class MTPDmessage; class HistoryStreamedDrafts final : public base::has_weak_ptr { public: @@ -26,24 +27,25 @@ public: const MTPDsendMessageTextDraftAction &data); [[nodiscard]] bool hasFor(not_null item) const; - void applyItemAdded(not_null item); void applyItemRemoved(not_null item); + HistoryItem *adoptIncoming(const MTPDmessage &data); private: struct Draft { not_null message; - uint64 randomId = 0; + MsgId rootId = 0; + PeerId fromId = 0; crl::time updated = 0; }; - bool update(MsgId rootId, uint64 randomId, const TextWithEntities &text); - void clear(MsgId rootId); + bool update(uint64 randomId, const TextWithEntities &text); + void clearByRandomId(uint64 randomId); void check(); void scheduleDestroy(); const not_null _history; - base::flat_map _drafts; + base::flat_map _drafts; base::Timer _checkTimer;