mqtt-streaming: Various fixes #1595

longshorej · 2019-03-19T21:30:35Z

A number of commits here that improve the resilience of the mqtt-streaming connector. #1577 remains important as well, but will require an Akka release.

mqtt-streaming: Require a connection id to be specified for client flows

This allows the MQTT session to better track which events are relevant, given that there can be races when old connections are torn down and new ones are established.

mqtt-streaming: Fix a bug where actor state was closed over

This could cause duplicate publications to never be sent to new connections

mqtt-streaming: Republish messages on reconnect only by default

Previously, messages for QoS1/2 were only republished on an interval after not receiving an ack.

It is more conventional to instead republish everything only on connect, and indeed to be compliant for MQTT 5, that is the only time this is allowed.

To accommodate this, the timeouts default to 0, but the previous behavior can still be restored by changing the default producer timeout settings.

longshorej · 2019-03-20T22:58:55Z

This is ready for a review @ennru @2m when you have a moment.

/cc @huntc

longshorej · 2019-03-20T23:06:36Z

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ClientState.scala

+ final case class ConsumerFree(topicName: String) extends Event(ByteString.empty)
+
+ final case class PublishReceivedLocally(publish: Publish, publishData: Producer.PublishData)
+ extends Event(ByteString.empty)


These and related aren't associated with a connection id, so their connection id is an empty ByteString. This is similar to how the server does things.

longshorej · 2019-03-20T23:09:00Z

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ClientState.scala

+ context.self ! connect
+ disconnect(context, data.remote, data)
+ case (_, event) if event.connectionId.nonEmpty && event.connectionId != data.connectionId =>
+ Behaviors.same


This case is responsible for letting empty (thus session bound) or matching connection ids fall through to other cases.

longshorej · 2019-03-20T23:10:37Z

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ClientState.scala

- case (_, ReceivedProducerPublishingCommand(command)) =>
- command.runWith(Sink.foreach {
- case Producer.ForwardPublish(publish, packetId) => data.remote.offer(ForwardPublish(publish, packetId))
- case Producer.ForwardPubRel(_, packetId) => data.remote.offer(ForwardPubRel(packetId))


Closing over old state -- if the client ever reconnected, data.remote would be from an old connection and thus never actually republished.

longshorej · 2019-03-20T23:12:36Z

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ClientState.scala

@@ -226,8 +257,13 @@ import scala.util.{Failure, Success}
 )
 )
 } else {
+ data.activeProducers.values.foreach { producer =>
+ producer ! Producer.ReceiveConnect
+ }


Main change on the client to let the producers know about new ReceiveConnect so that they can republish.

longshorej · 2019-03-20T23:13:11Z

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ServerState.scala

@@ -390,6 +390,9 @@ import scala.util.{Failure, Success}
 data.stash.foreach(context.self.tell)
 timer.cancel(ReceiveConnAck)

+ data.activeProducers.values
+ .foreach(_ ! Producer.ReceiveConnect)


Similarly on the server.

huntc

Review of the connection id logic: Looking good. One comment, although more of a question than anything that should stop this from being merged.

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/scaladsl/MqttSession.scala

huntc

Re. closing over - it took me a few times, but now I get it. Really nice find.

huntc

Re-republishing on reconnect. Really nicely done.

huntc · 2019-03-21T02:41:20Z

@ennru @2m All LGTM. Thanks @longshorej !

One thing to note is that the client connection flow's now require a connection id, in a similar way to the server connection flows. This was an oversight in our original APIs.

ennru · 2019-03-21T08:56:47Z

Can you add a MiMa filter for everything in the impl package, please?
Than we'll see how much the public API changed.

huntc · 2019-03-21T09:13:20Z

Can you add a MiMa filter for everything in the impl package, please? Than we'll see how much the public API changed.

I've not used MiMa before, and I've been looking a way for each project to be able to conveniently exclude their impl resident types. Do you know how that can be achieved with ease?

ennru · 2019-03-21T11:31:44Z

Fair enough. I'll add it later today.

huntc · 2019-03-21T11:35:52Z

Thanks @ennru

ennru · 2019-03-21T12:53:07Z

Now we're down to the API breaking changes we want to see

method clientSessionFlow(akka.stream.alpakka.mqtt.streaming.javadsl.MqttClientSession): akka.stream.javadsl.BidiFlow in object akka.stream.alpakka.mqtt.streaming.javadsl.Mqtt does not have a correspondent in current version
method clientSessionFlow(akka.stream.alpakka.mqtt.streaming.scaladsl.MqttClientSession): akka.stream.scaladsl.BidiFlow in object akka.stream.alpakka.mqtt.streaming.scaladsl.Mqtt does not have a correspondent in current version

as those got a new parameter introduced.

longshorej · 2019-03-21T16:13:22Z

Yeah, that's expected as each flow needs a connection id now (similar to the server). This module is considered an "API may change" one still, right? Given that we're still going through hardening exercises that seems reasonable to me.

huntc · 2019-03-21T20:33:30Z

The lack of a connection id was an oversight on my part initially.

It is all because of this notion of a session with MQTT. A session must distinguish connections. We were doing this on the server side and exposing it in the api, but we weren’t exposing it for the client api. The client session needs to understand which connection is sending a message as there can be a race between an active one shutting down and a new one being established.

Incidentally, your higher level APIs can hide this complexity given that you’re managing the connection entirely. A good candidate for a client connection id is the client port that the TCP socket has (not ther server port!).

HTH.

ennru

LGTM.

Technically this module is not marked "API may change" so you should add a clientSessionFlow method to Mqtt without the connectionId and generating one. That method should be deprecated and point to the new version.

We will be very open to do major releases once 1.0 is out when some API needs improvement that can't keep backwards-compatibilty.

This allows the MQTT session to better track which events are relevant, given that there can be races when old connections are torn down and new ones are established.

This could cause duplicate publications to never be sent to new connections

Previously, messages for QoS1/2 were only republished on an interval after not receiving an ack. It is more conventional to instead republish everything only on connect, and indeed to be compliant for MQTT 5, that is the only time this is allowed. To accomodate this, the timeouts default to 0, but the previous behavior can still be restored by changing the default producer timeout settings.

longshorej · 2019-03-25T17:36:32Z

I added the requested deprecated methods, but MiMa is still complaining about something. Any advice @ennru?

ennru · 2019-03-25T17:44:17Z

Looks like you removed the MiMa exclusions file, the version I provided is required for it to accept the other changes.

longshorej · 2019-03-25T18:11:59Z

Ah thanks, that must have happened in the rebase. I've readded it.

…ient flows

ennru · 2019-03-26T07:20:52Z

Thank you @longshorej and @huntc for your continued work in this! Great to see other contributors helping out, as well.

longshorej force-pushed the publish branch from 0ea893a to b7b8745 Compare March 20, 2019 22:55

longshorej changed the title ~~mqtt-streaming: Only publish when reconnecting, not after a timeout~~ mqtt-streaming: Various fixes Mar 20, 2019

longshorej marked this pull request as ready for review March 20, 2019 22:58

longshorej force-pushed the publish branch from b7b8745 to b1cbeb8 Compare March 20, 2019 23:07

longshorej commented Mar 20, 2019

View reviewed changes

huntc approved these changes Mar 21, 2019

View reviewed changes

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/scaladsl/MqttSession.scala Outdated Show resolved Hide resolved

huntc approved these changes Mar 21, 2019

View reviewed changes

longshorej force-pushed the publish branch from b1cbeb8 to f0902ad Compare March 21, 2019 04:39

ennru added the p:mqtt-streaming label Mar 21, 2019

longshorej force-pushed the publish branch from 909bf91 to 2e24b6e Compare March 21, 2019 16:17

ennru approved these changes Mar 22, 2019

View reviewed changes

longshorej force-pushed the publish branch from 2e24b6e to 091152a Compare March 25, 2019 16:24

longshorej added 3 commits March 25, 2019 11:37

mqtt-streaming: Require a connection id to be specified for client flows

ae488a0

This allows the MQTT session to better track which events are relevant, given that there can be races when old connections are torn down and new ones are established.

mqtt-streaming: Fix a bug where actor state was closed over

0323c25

This could cause duplicate publications to never be sent to new connections

longshorej force-pushed the publish branch from 091152a to 0e260c4 Compare March 25, 2019 16:40

Add MiMa filters for internal API changes

be6af71

fixup! mqtt-streaming: Require a connection id to be specified for cl…

9fc6f1c

…ient flows

ennru merged commit 0ee0ae5 into akka:master Mar 26, 2019

ennru added this to the 1.0-RC2 milestone Mar 26, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

mqtt-streaming: Various fixes #1595

mqtt-streaming: Various fixes #1595

longshorej commented Mar 19, 2019 •

edited

Loading

longshorej commented Mar 20, 2019

longshorej Mar 20, 2019

longshorej Mar 20, 2019

longshorej Mar 20, 2019

longshorej Mar 20, 2019

longshorej Mar 20, 2019

huntc left a comment

huntc left a comment

huntc left a comment

huntc commented Mar 21, 2019 •

edited

Loading

ennru commented Mar 21, 2019

huntc commented Mar 21, 2019

ennru commented Mar 21, 2019

huntc commented Mar 21, 2019

ennru commented Mar 21, 2019

longshorej commented Mar 21, 2019

huntc commented Mar 21, 2019

ennru left a comment

longshorej commented Mar 25, 2019

ennru commented Mar 25, 2019

longshorej commented Mar 25, 2019

ennru commented Mar 26, 2019

mqtt-streaming: Various fixes #1595

mqtt-streaming: Various fixes #1595

Conversation

longshorej commented Mar 19, 2019 • edited Loading

mqtt-streaming: Require a connection id to be specified for client flows

mqtt-streaming: Fix a bug where actor state was closed over

mqtt-streaming: Republish messages on reconnect only by default

longshorej commented Mar 20, 2019

longshorej Mar 20, 2019

Choose a reason for hiding this comment

longshorej Mar 20, 2019

Choose a reason for hiding this comment

longshorej Mar 20, 2019

Choose a reason for hiding this comment

longshorej Mar 20, 2019

Choose a reason for hiding this comment

longshorej Mar 20, 2019

Choose a reason for hiding this comment

huntc left a comment

Choose a reason for hiding this comment

huntc left a comment

Choose a reason for hiding this comment

huntc left a comment

Choose a reason for hiding this comment

huntc commented Mar 21, 2019 • edited Loading

ennru commented Mar 21, 2019

huntc commented Mar 21, 2019

ennru commented Mar 21, 2019

huntc commented Mar 21, 2019

ennru commented Mar 21, 2019

longshorej commented Mar 21, 2019

huntc commented Mar 21, 2019

ennru left a comment

Choose a reason for hiding this comment

longshorej commented Mar 25, 2019

ennru commented Mar 25, 2019

longshorej commented Mar 25, 2019

ennru commented Mar 26, 2019

longshorej commented Mar 19, 2019 •

edited

Loading

huntc commented Mar 21, 2019 •

edited

Loading