Dima/matrix-js-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Messages sent out of order after one message fails (#3131)

Browse Source 
* Instead of skipping, bail out by clearing queue
* Allow additional status transition for events from QUEUED to NOT_SENT
This commit is contained in:
Janne Mareike Koschinski
2023-02-08 13:23:30 +01:00
committed by GitHub
parent 16672b3d0c
commit 5cf0bb46a4
3 changed files with 28 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/models/room.ts
+1 -1
View File
@@ -3380,7 +3380,7 @@ export class Room extends ReadReceipt<RoomEmittedEvents, RoomEventHandlerMap> {
const ALLOWED_TRANSITIONS: Record<EventStatus, EventStatus[]> = {
[EventStatus.ENCRYPTING]: [EventStatus.SENDING, EventStatus.NOT_SENT, EventStatus.CANCELLED],
[EventStatus.SENDING]: [EventStatus.ENCRYPTING, EventStatus.QUEUED, EventStatus.NOT_SENT, EventStatus.SENT],
[EventStatus.QUEUED]: [EventStatus.SENDING, EventStatus.CANCELLED],
[EventStatus.QUEUED]: [EventStatus.SENDING, EventStatus.NOT_SENT, EventStatus.CANCELLED],
[EventStatus.SENT]: [],
[EventStatus.NOT_SENT]: [EventStatus.SENDING, EventStatus.QUEUED, EventStatus.CANCELLED],
[EventStatus.CANCELLED]: [],
src/scheduler.ts
+20 -10
View File
@@ -245,12 +245,7 @@ export class MatrixScheduler<T = ISendEventResponse> {
// get head of queue
const obj = this.peekNextEvent(queueName);
if (!obj) {
// queue is empty. Mark as inactive and stop recursing.
const index = this.activeQueues.indexOf(queueName);
if (index >= 0) {
this.activeQueues.splice(index, 1);
}
debuglog("Stopping queue '%s' as it is now empty", queueName);
this.disableQueue(queueName);
return;
}
debuglog("Queue '%s' has %s pending events", queueName, this.queues[queueName].length);
@@ -289,10 +284,7 @@ export class MatrixScheduler<T = ISendEventResponse> {
// give up (you quitter!)
debuglog("Queue '%s' giving up on event %s", queueName, obj.event.getId());
// remove this from the queue
this.removeNextEvent(queueName);
obj.defer.reject(err);
// process next event
this.processQueue(queueName);
this.clearQueue(queueName, err);
} else {
setTimeout(this.processQueue, waitTimeMs, queueName);
}
@@ -300,6 +292,24 @@ export class MatrixScheduler<T = ISendEventResponse> {
);
};
private disableQueue(queueName: string): void {
// queue is empty. Mark as inactive and stop recursing.
const index = this.activeQueues.indexOf(queueName);
if (index >= 0) {
this.activeQueues.splice(index, 1);
}
debuglog("Stopping queue '%s' as it is now empty", queueName);
}
private clearQueue(queueName: string, err: unknown): void {
debuglog("clearing queue '%s'", queueName);
let obj: IQueueEntry<T> | undefined;
while ((obj = this.removeNextEvent(queueName))) {
obj.defer.reject(err);
}
this.disableQueue(queueName);
}
private peekNextEvent(queueName: string): IQueueEntry<T> | undefined {
const queue = this.queues[queueName];
if (!Array.isArray(queue)) {