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

fix(server): message handling order when using beforeHandleMessage (#880) #901

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
+32 −1
Open

fix(server): message handling order when using beforeHandleMessage (#880) #901

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e79f2c9
fix: pass `WebSocket` instead of Connection to `applyAwarenessUpdate`
RubaXa Dec 5, 2024
f3bc2b0
Merge branch 'ueberdosis:main' into main
RubaXa Feb 11, 2025
42aa1ed
fix(server): message handling order when using `beforeHandleMessage` …
RubaXa Feb 11, 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
33 changes: 32 additions & 1 deletion packages/server/src/Connection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,10 @@ export class Connection {

logger: Debugger

private beforeHandleMessageQueue: Uint8Array[] = []
Copy link

@raimohanska raimohanska Feb 16, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

You might be able to do with just 1 additional state variable if you shifted the message only after successfully processing it.


private beforeHandleMessageProcessing = false

/**
* Constructor.
*/
Expand Down Expand Up @@ -216,21 +220,40 @@ export class Connection {
* @public
*/
public handleMessage(data: Uint8Array): void {
if (this.beforeHandleMessageQueue.length > 0 || this.beforeHandleMessageProcessing) {
this.beforeHandleMessageQueue.push(data)
return
}

this.processBeforeHandleMessage(data)
}

/** Process an incoming message */
private processBeforeHandleMessage(data: Uint8Array): void {
Copy link

Choose a reason for hiding this comment

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

The name of the method is out of sync with the comment. I like the comment more, as what this methos does is process an incoming message after applying the beforeHandleMessage hooks.

Copy link

Choose a reason for hiding this comment

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

I would prefer using async-await syntax for this method. This would allow you to use a while (this. beforeHandleMessageQueue.length > 0 loop.

this.beforeHandleMessageProcessing = true

const message = new IncomingMessage(data)
const documentName = message.readVarString()

if (documentName !== this.document.name) return
if (documentName !== this.document.name) {
this.nextProcessBeforeHandleMessage()
return
}

message.writeVarString(documentName)

this.callbacks.beforeHandleMessage(this, data)
.then(() => {
this.nextProcessBeforeHandleMessage()
Copy link

Choose a reason for hiding this comment

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

The ordering of things here seems a bit off to me: you trigger the processing of a next message before you've processed this one (below).

Copy link

Choose a reason for hiding this comment

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

The invocation of this.nextProcessBeforeHandleMessage will initiate a new async sequence and to me it seems possible that things go out-of-order again.


new MessageReceiver(
message,
this.logger,
).apply(this.document, this)
})
.catch((e: any) => {
this.nextProcessBeforeHandleMessage()

console.log('closing connection because of exception', e)
this.close({
code: 'code' in e ? e.code : Forbidden.code,
Expand All @@ -239,6 +262,14 @@ export class Connection {
})
}

/** Process next an incoming message */
private nextProcessBeforeHandleMessage(): void {
this.beforeHandleMessageProcessing = false

if (this.beforeHandleMessageQueue.length > 0) {
this.processBeforeHandleMessage(this.beforeHandleMessageQueue.shift()!)
}
}
}

export default Connection