MQTT streaming: Fixup the continuation of a session #1414
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Prior to this commit, if a server side session was publishing to a client and the client disconnected, and then subsequently reconnected, its publications would be lost. This is because I was closing over some actor state within a future. Whoops. I’ve now written a comprehensive test for this scenario, along with some additional subscription tests along the way.
huntc
force-pushed
the
resuming-session
branch
from
f37c191
to
e0cb8a5
Compare
huntc
commented
Jan 13, 2019
| reply.future.foreach(command => context.self ! ReceivedProducerPublishingCommand(command)) | ||
| reply.future.foreach { | ||
| _.runForeach(command => context.self ! ReceivedProducerPublishingCommand(command)) | ||
| } |
There was a problem hiding this comment.
This file contains the main change
huntc
commented
Jan 13, 2019
| case Failure(_: WatchedActorTerminatedException) => | ||
| case _ => | ||
| clientConnector ! ClientConnector.ConnectionLost | ||
| } |
There was a problem hiding this comment.
Slight optimisation to reduce log messages. We are explicitly watching an actor so don't send it a message when terminated. :-)
huntc
commented
Jan 13, 2019
| @@ -1233,6 +1349,146 @@ class MqttSessionSpec | |||
| server.offer(Command(pubAck)) | |||
| client.expectMsg(pubAckBytes) | |||
| } | |||
|
|
|||
| "produce a duplicate publish on the server given two client connections" in { | |||
There was a problem hiding this comment.
This is the test that confirms the PR fix.
huntc
commented
Jan 13, 2019
| @@ -167,6 +167,51 @@ class MqttSessionSpec | |||
| result.futureValue shouldBe Right(Event(subAck)) | |||
| } | |||
|
|
|||
| "Subscribe and stash any subsequent subscriptions" in { | |||
There was a problem hiding this comment.
Additional test - nothing directly to do with the problem being solved by the PR
huntc
commented
Jan 13, 2019
| @@ -900,6 +944,78 @@ class MqttSessionSpec | |||
| } | |||
| } | |||
|
|
|||
| "receive two subscriptions for the same topic" in { | |||
There was a problem hiding this comment.
Additional test - nothing directly to do with the problem being solved by the PR
huntc
commented
Jan 13, 2019
| @@ -1165,7 +1281,7 @@ class MqttSessionSpec | |||
| "re-connect given connect, subscribe, connect again, publish" in | |||
| reconnectTest(explicitDisconnect = false) | |||
|
|
|||
| "receive a duplicate publish" in { | |||
| "consume a duplicate publish on the server" in { | |||
There was a problem hiding this comment.
Clarification of this test
So that the logs are clearer, but also to ensure no untoward behaviour, we now cancel timers when transitioning to another behaviour.
To facilitate debugging, we now log MQTT events as they pass through the session. These events are available only when the akka.loglevel is set to debug. We also take care to avoid the output of sensitive information.
huntc
changed the title
|
Ready for review. |
longshorej
approved these changes
Jan 14, 2019
ennru
changed the title
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Prior to this commit, if a server-side session was publishing to a client and the client disconnected, and then subsequently reconnected, its publications would be lost. This is because I was closing over some actor state within a future. Whoops.
I’ve now written a comprehensive test for this scenario, along with some additional, unrelated, subscription tests along the way.
I also cancel timers explicitly to be tidy (and to avoid potential misfiring issues, although I've not seen any), and the logging has been improved so that debug log levels outputs MQTT events sans sensitive info (passwords, payloads); the latter being useful toward logging.