Updates the command enqueuing to Session Event Loop to return the status of the operation

andsel commented 2 years ago

Release notes

[rn:skip]

What does this PR do?

Fix a design problem in the enqueuing of commands to Session Event loops. In the first version the of the process publish methods, the receivedPublish* operations returned a CompletableFuture which wrapped the logic of the message processing. In case of validation error, was returned an already completed future. For example the receivedPublishQos1 returned a future that was obtained as composition of the action to publish to subscribers, concatenated with the logic to send the PUBACK to the publisher. So it happened that the PUBACK was sent when publish was forwarded to the matching subscribers. The problem is that the operation to enqueue could be successful or fail, and in case of success there is a CompleteableFuture that rappresents the part of the forwarding to subscribers. The PUBACK message should be sent when the message is taken into account from the broker and not when the publish has completely forwarded to all subscribers.

This PR changes the receivedPublishQos0 receivedPublishQos1 receivedPublishQos2 to return a RouteResult object that contains the status of the enqueuing operation plus the future of the command task.

It touches a lot of parts of the code due to signature changes, but the core changes are all in PostOffice.

Why is it important/What is the impact to the user?

The final user is not impated by this, the developer has a more straightforward and clean process logic

Checklist

[x] My code follows the style guidelines of this project
[x] I have commented my code, particularly in hard-to-understand areas
~~[ ] I have made corresponding changes to the documentation~~
~~[ ] I have made corresponding change to the default configuration files (and/or docker env variables)~~
[x] I have added tests that prove my fix is effective or that my feature works

Related issues

Related #631

hylkevds commented 2 years ago

:+1: That makes things a lot easier to follow!

andsel commented 2 years ago

Hi @hylkevds may I ask you for a review of this PR? I know it's big, but the core part of the changes is in PostOffice class. Let me know :-)

hylkevds commented 2 years ago

Hi @hylkevds may I ask you for a review of this PR? I know it's big, but the core part of the changes is in PostOffice class. Let me know :-)

Of course, no problem.

hylkevds commented 2 years ago

(Review in progress...) Found one leak, when a publish fails because a Session command queue is full (at least for QoS 2). I think it is either because

MQTTConnection.processPublish does a retain, but when the queue is full the release in PostOffice.receivedPublishQos2 never runs.
PostOffice.publish2Subscribers:499 does a retain for each relevant queue, but the command lambda never executes in PostOffice.publish2Subscribers:503

Or maybe it's both reasons.

Maybe we can change Server.internalPublish for embedded applications to return the RoutingResults too, that way the embedded application can check if the publish was OK, and re-try if it wants to.

andsel commented 2 years ago

Hi @hylkevds thanks a ton to have spotted all those missed release. Have fixed and aligned to main, now thi s PR is ready for a second run of review :-)

andsel commented 2 years ago

@hylkevds wait before review, because the test testClientDoesntRemainSubscribedAfterASubscriptionAndServerRestart started to hangup

moquette-io / moquette