Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Post Event PendingEventItems #861

Open
wants to merge 2 commits into
base: dev
Choose a base branch
from
Open

Post Event PendingEventItems #861

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4a66126
Always return pending event items when posting events rather than tra…
nvrWhere Jan 25, 2025
722bc3d
Add a future for about to merge
nvrWhere Jan 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 26 additions & 9 deletions Quotient/eventitem.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,11 +98,13 @@ class QUOTIENT_API TimelineItem : public EventItemBase {

class QUOTIENT_API PendingEventItem : public EventItemBase {
public:
using future_type = QFuture<std::reference_wrapper<const RoomEvent>>;
using about_to_merge_future_type = QFuture<std::pair<std::reference_wrapper<const RoomEvent>, int>>;
using merged_future_type = QFuture<std::reference_wrapper<const RoomEvent>>;

explicit PendingEventItem(RoomEventPtr&& e) : EventItemBase(std::move(e))
{
_promise.setProgressRange(0, 5);
_aboutToMergePromise.setProgressRange(0, 5);
_mergePromise.setProgressRange(0, 5);
}

EventStatus::Code deliveryStatus() const { return _status; }
Expand All @@ -116,10 +118,15 @@ class QUOTIENT_API PendingEventItem : public EventItemBase {
setStatus(EventStatus::ReachedServer);
(*this)->addId(eventId);
}
void setAboutToMerge(const RoomEvent& intoEvent, int pendingEventIndex)
{
_aboutToMergePromise.addResult(std::make_pair(std::ref(intoEvent), pendingEventIndex));
_aboutToMergePromise.finish();
}
void setMerged(const RoomEvent& intoEvent)
{
_promise.addResult(intoEvent);
_promise.finish();
_mergePromise.addResult(intoEvent);
_mergePromise.finish();
}
void setSendingFailed(QString errorText)
{
Expand All @@ -128,28 +135,38 @@ class QUOTIENT_API PendingEventItem : public EventItemBase {
}
void resetStatus() { setStatus(EventStatus::Submitted); }

//! \brief Get a future for the moment when the item gets merged in the timeline
//! \brief Get a future for the moment just before the item gets merged in the timeline
//!
//! The future will get finished just before this pending item is merged into its remote
//! counterpart that comes with /sync. The pending item will always be in ReachedServer state.
//! The future result has type implicitly convertible to std::pair with a `const RoomEvent&` and an `int`.
about_to_merge_future_type whenAboutToMerged() const { return _aboutToMergePromise.future(); }
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you have code that would use it? Or at least can you give an idea of code that would use it? Futures are generally heavier than signal-slot connections.


//! \brief Get a future for the moment when the item gets merged in the timeline
//!
//! The future will get finished just after this pending item is merged into its remote
//! counterpart that comes with /sync. The pending item will always be in ReachedServer state.
//! The future result has type implicitly convertible to `const RoomEvent&`.
future_type whenMerged() const { return _promise.future(); }
merged_future_type whenMerged() const { return _mergePromise.future(); }

private:
// Unlike TimelineItems, it's reasonable to assume PendingEventItems are not many; so we can
// add extra fields without breaking the memory bill
EventStatus::Code _status = EventStatus::Submitted;
QDateTime _lastUpdated = QDateTime::currentDateTimeUtc();
QString _annotation;
QPromise<std::reference_wrapper<const RoomEvent>> _promise;
QPromise<std::pair<std::reference_wrapper<const RoomEvent>, int>> _aboutToMergePromise;
QPromise<std::reference_wrapper<const RoomEvent>> _mergePromise;

void setStatus(EventStatus::Code status)
{
_status = status;
_lastUpdated = QDateTime::currentDateTimeUtc();
_annotation.clear();
_promise.start();
_promise.setProgressValue(_status);
_aboutToMergePromise.start();
_mergePromise.start();
_aboutToMergePromise.setProgressValue(_status);
_mergePromise.setProgressValue(_status);
}
};

Expand Down
24 changes: 13 additions & 11 deletions Quotient/room.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,7 +297,7 @@ class Q_DECL_HIDDEN Room::Private {

const PendingEventItem& sendEvent(RoomEventPtr&& event);

QString doPostFile(event_ptr_tt<RoomMessageEvent> fileEvent, const QUrl& localUrl);
const PendingEventItem& doPostFile(event_ptr_tt<RoomMessageEvent> fileEvent, const QUrl& localUrl);

PendingEvents::iterator addAsPending(RoomEventPtr&& event);

Expand Down Expand Up @@ -1976,7 +1976,7 @@ Room::PendingEvents::iterator Room::Private::addAsPending(RoomEventPtr&& event)
event->setSender(connection->userId());
emit q->pendingEventAboutToAdd(std::to_address(event));
auto it = unsyncedEvents.emplace(unsyncedEvents.end(), std::move(event));
emit q->pendingEventAdded(it->event());
emit q->pendingEventAdded(*it);
return it;
}

Expand Down Expand Up @@ -2093,7 +2093,7 @@ void Room::Private::onEventSendingFailure(PendingEvents::iterator eventItemIter,
emit q->pendingEventChanged(int(eventItemIter - unsyncedEvents.begin()));
}

PendingEventItem::future_type Room::whenMessageMerged(QString txnId) const
PendingEventItem::merged_future_type Room::whenMessageMerged(QString txnId) const
{
if (auto it = findPendingEvent(txnId); it != d->unsyncedEvents.cend())
return it->whenMerged();
Expand Down Expand Up @@ -2162,14 +2162,15 @@ void Room::discardMessage(const QString& txnId)
emit pendingEventDiscarded();
}

QString Room::postReaction(const QString& eventId, const QString& key)
const PendingEventItem& Room::postReaction(const QString& eventId, const QString& key)
{
return post<ReactionEvent>(eventId, key)->transactionId();
return post<ReactionEvent>(eventId, key);
}

QString Room::Private::doPostFile(event_ptr_tt<RoomMessageEvent> fileEvent, const QUrl& localUrl)
const PendingEventItem& Room::Private::doPostFile(event_ptr_tt<RoomMessageEvent> fileEvent, const QUrl& localUrl)
{
const auto txnId = addAsPending(std::move(fileEvent))->event()->transactionId();
const auto& pendingItem = *addAsPending(std::move(fileEvent));
const auto txnId = pendingItem->transactionId();
// Remote URL will only be known after upload; fill in the local path
// to enable the preview while the event is pending.
q->uploadFile(txnId, localUrl);
Expand Down Expand Up @@ -2212,10 +2213,10 @@ QString Room::Private::doPostFile(event_ptr_tt<RoomMessageEvent> fileEvent, cons
emit q->pendingEventDiscarded();
});

return txnId;
return pendingItem;
}

QString Room::postFile(const QString& plainText,
const PendingEventItem& Room::postFile(const QString& plainText,
std::unique_ptr<EventContent::FileContentBase> fileContent,
std::optional<EventRelation> relatesTo)
{
Expand All @@ -2237,9 +2238,9 @@ const PendingEventItem& Room::post(RoomEventPtr event)
return d->sendEvent(std::move(event));
}

QString Room::postJson(const QString& matrixType, const QJsonObject& eventContent)
const PendingEventItem& Room::postJson(const QString& matrixType, const QJsonObject& eventContent)
{
return d->sendEvent(loadEvent<RoomEvent>(matrixType, eventContent))->transactionId();
return d->sendEvent(loadEvent<RoomEvent>(matrixType, eventContent));
}

SetRoomStateWithKeyJob* Room::setState(const StateEvent& evt)
Expand Down Expand Up @@ -2802,6 +2803,7 @@ Room::Timeline::size_type Room::Private::mergePendingEvent(PendingEvents::iterat
auto* remoteEcho = remoteEchoIt->get();
const auto pendingEvtIdx = int(localEchoIt - unsyncedEvents.begin());
onEventReachedServer(localEchoIt, remoteEcho->id());
localEchoIt->setAboutToMerge(*remoteEcho, pendingEvtIdx);
emit q->pendingEventAboutToMerge(remoteEcho, pendingEvtIdx);
qCDebug(MESSAGES) << "Merging pending event from transaction" << remoteEcho->transactionId()
<< "into" << remoteEcho->id();
Expand Down
14 changes: 7 additions & 7 deletions Quotient/room.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -725,7 +725,7 @@ class QUOTIENT_API Room : public QObject {
//!
//! This means MessageEventType Text, Emote or Notice.
template<MessageEventType type = MessageEventType::Text>
QString postText(const QString& plainText,
const PendingEventItem& postText(const QString& plainText,
const std::optional<QString>& html = std::nullopt,
const std::optional<EventRelation>& relatesTo = std::nullopt)
{
Expand All @@ -739,21 +739,21 @@ class QUOTIENT_API Room : public QObject {
if (html) {
content = std::make_unique<EventContent::TextContent>(*html, u"text/html"_s);
}
return post<RoomMessageEvent>(plainText, type, std::move(content), relatesTo)->transactionId();
return post<RoomMessageEvent>(plainText, type, std::move(content), relatesTo);
}

//! Send a file with the given content
QString postFile(const QString& plainText,
const PendingEventItem& postFile(const QString& plainText,
std::unique_ptr<EventContent::FileContentBase> fileContent,
std::optional<EventRelation> relatesTo = std::nullopt);

//! Send the given Json as a message
QString postJson(const QString& matrixType, const QJsonObject& eventContent);
const PendingEventItem& postJson(const QString& matrixType, const QJsonObject& eventContent);

//! Send a reaction on a given event with a given key
QString postReaction(const QString& eventId, const QString& key);
const PendingEventItem& postReaction(const QString& eventId, const QString& key);

PendingEventItem::future_type whenMessageMerged(QString txnId) const;
PendingEventItem::merged_future_type whenMessageMerged(QString txnId) const;

//! Send a request to update the room state with the given event
SetRoomStateWithKeyJob* setState(const StateEvent& evt);
Expand Down Expand Up @@ -850,7 +850,7 @@ public Q_SLOTS:
/// The event is about to be appended to the list of pending events
void pendingEventAboutToAdd(Quotient::RoomEvent* event);
/// An event has been appended to the list of pending events
void pendingEventAdded(const Quotient::RoomEvent* event);
void pendingEventAdded(const PendingEventItem& pendingItem);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure it's a good idea; for one, you cannot use a queued connection with this signal because a reference cannot be copied; and besides, this reference is at risk of dangling if any slot connected to pendingEventAdded causes adding another pending event to the same room (I stumbled over it when writing Quotest, some time ago). At the very least, these caveats should be in the signal documentation.

/// The remote echo has arrived with the sync and will be merged
/// with its local counterpart
/** NB: Requires a sync loop to be emitted */
Expand Down
13 changes: 7 additions & 6 deletions quotest/quotest.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -379,12 +379,13 @@ TEST_IMPL(loadMembers)

TEST_IMPL(sendMessage)
{
auto txnId = targetRoom->postText("Hello, "_L1 % origin % " is here"_L1);
const auto& pendingItem = targetRoom->postText("Hello, "_L1 % origin % " is here"_L1);
const auto txnId = pendingItem->transactionId();
if (!validatePendingEvent<RoomMessageEvent>(txnId)) {
clog << "Invalid pending event right after submitting" << endl;
FAIL_TEST();
}
targetRoom->whenMessageMerged(txnId).then(this, [this, thisTest, txnId](const RoomEvent& evt) {
pendingItem.whenMerged().then([this, thisTest, txnId](const RoomEvent& evt) {
const auto pendingIt = targetRoom->findPendingEvent(txnId);
if (pendingIt == targetRoom->pendingEvents().end()) {
clog << "Pending event not found at the moment of local echo merging\n";
Expand Down Expand Up @@ -414,7 +415,7 @@ TEST_IMPL(sendReaction)
}

const auto key = u"+"_s;
const auto txnId = targetRoom->postReaction(targetEvtId, key);
const auto txnId = targetRoom->postReaction(targetEvtId, key)->transactionId();
FAIL_TEST_IF(!validatePendingEvent<ReactionEvent>(txnId),
"Invalid pending event right after submitting");

Expand Down Expand Up @@ -452,7 +453,7 @@ TEST_IMPL(sendFile)
const auto tfName = tfi.fileName();
clog << "Sending file " << tfName.toStdString() << endl;
const auto txnId = targetRoom->postFile(
"Test file"_L1, std::make_unique<EventContent::FileContent>(tfi));
"Test file"_L1, std::make_unique<EventContent::FileContent>(tfi))->transactionId();
if (!validatePendingEvent<RoomMessageEvent>(txnId)) {
clog << "Invalid pending event right after submitting" << endl;
tf->deleteLater();
Expand Down Expand Up @@ -905,7 +906,7 @@ TEST_IMPL(visitResources)

TEST_IMPL(thread)
{
auto rootTxnId = targetRoom->postText("Threadroot"_L1);
auto rootTxnId = targetRoom->postText("Threadroot"_L1)->transactionId();
connect(targetRoom, &Room::pendingEventAboutToMerge, this, [this, thisTest, rootTxnId](Quotient::RoomEvent* rootEvt) {
if (rootEvt->transactionId() == rootTxnId) {
const auto relation = EventRelation::replyInThread(rootEvt->id(), true, rootEvt->id());
Expand Down Expand Up @@ -975,7 +976,7 @@ void TestManager::conclude()
htmlReport += "<br><strong>Did not finish:</strong>"_L1 + QString::fromUtf8(dnfList);
}

auto txnId = room->postText(plainReport, htmlReport);
auto txnId = room->postText(plainReport, htmlReport)->transactionId();
// Now just wait until all the pending events reach the server
connectUntil(room, &Room::messageSent, this, [this, txnId, room, plainReport] {
const auto& pendingEvents = room->pendingEvents();
Expand Down