MQTT streaming: Document PubAck handling

[huntc]

This attempts to improve the API around how to back-pressure the publication of Publish commands to avoid buffer overflow in QoS 1+ cases.

I'm looking for feedback as to whether this helps explain things, particularly given problems like this: https://discuss.lightbend.com/t/alpakka-mqtt-streaming-performance-problem/4913.

TODO:

• Provide Java sample

[ennru]

Yes, this helps. That scanning is not what I'd want to have every user to add to their code. That's definitely something I'd like to wrap into the convenience API, I've been experimenting with a while ago: #1565
Doing this from Java will be even harder to read...

[huntc]

In the case of publishing from a client, I think that we can provide the equivalent to the ask pattern where a Future[Done] is returned. This is in addition to the tell we already have.

I do think that specific cases like client publishing can be made more convenient.

WDYT?

[huntc]

In the case of publishing from a client, I think that we can provide the equivalent to the ask pattern where a Future[Done] is returned. This is in addition to the tell we already have.

I do think that specific cases like client publishing can be made more convenient.

Here's what the ask pattern may look like: https://github.com/akka/alpakka/pull/1908/files#diff-57b7cf48120d900f9398000d526e9854R70

Thoughts @ennru ?

[ennru]

Yes, that would be great and feel consistent with the tell/ask in Akka Actors.

[huntc]

@ennru I've just updated the PR with the ask API. I'm actually now wondering though if both tell and ask should be rolled into one and named offer i.e. treat the session the queue that it is. Thoughts?

[huntc]

I'm getting some MiMa issues given the new APIs... these changes should be permitted though, right? After all, origin is a 2.0 branch and there won't be any intention of binary compatibility, right?

[ennru]

I'm actually now wondering though if both tell and ask should be rolled into one and named offer i.e. treat the session the queue that it is.

Would the offer align well with Source.queue's offer? Would it fail for similar reasons eg. when buffers are full?

After all, origin is a 2.0 branch and there won't be any intention of binary compatibility, right?

The ambition is to stay as source- and binary-compatible as feasible. These changes are definitely OK to allow.

[huntc]

Would the offer align well with Source.queue's offer? Would it fail for similar reasons eg. when buffers are full?

Yeah, that’s what I was thinking...

[huntc]

@ennru This bit of code is essentially what we're doing today in production. The code completes the carry (which is a Promise[Done]) when the PubAck it is associated with is received.

val (commands: SourceQueueWithComplete[Command[Promise[Done]]], events: Future[ControlPacket]) =
  Source
    .queue(2, OverflowStrategy.fail)
    .via(mqttFlow)
    .collect {
      case Right(Event(p: Publish, _)) =>
        p
      case Right(Event(pa: PubAck, Some(carry))) =>
        carry.success(Done)
        pa
    }

My thinking now is have the docs explain the above requirement. Perhaps we can visit the higher level API PR you started and keep things the way they are here. Thoughts?

[ennru]

Sorry, I guess this was on me to get back to.