MQTT streaming: Express the carry type

[huntc]

This ensures that the client code usage is nicer, and tighter, when working with the carry object. The Java API is more affected that the Scala API. The Scala API is hardly affected at all, and where it is, it requires substituting Command[_] and Event[_] with Command[Nothing] and Event[Nothing] respectively (unless you have a carry, where then you provide its type instead of Nothing)(see the Scala API changes via our tests).

An interesting point to note is that Java's class method does not return the type of a generic e.g.:

DecodeErrorOrEvent.class

...will return a Class<DecodeErrorOrEvent>. However, we want Class<DecodeErrorOrEvent<A>> where A is the type of the element to carry. In Scala, classOf will return this i.e. classOf[DecodeErrorOrEvent[A]] returns what we want. So, I've created a Java API classOf function for DecodeErrorOrEvent. Thus, we can correctly express situations like:

Pair<
        SourceQueueWithComplete<Command<Object>>,
        Source<DecodeErrorOrEvent<Object>, NotUsed>>
    run =
        Source.<Command<Object>>queue(2, OverflowStrategy.dropHead())
            .via(mqttFlow)
            .toMat(BroadcastHub.of(DecodeErrorOrEvent.classOf()), Keep.both())
            .run(materializer);

If there's a better way to provide the classOf outcome then I'm all ears. However, I've dwelled on this for a couple of days. I think the solution here works well.

[ennru]

We recently got this contribution to the docs: Pass through

[huntc]

Thanks @ennru. I’ve long wanted to see Akka Streams itself provide for this scenario. Our situation though is that the MQTT connector already caters for carrying these elements.

Any advice on how to use BroadcastHub with generic types from Java?

[huntc]

I should add that the PassThroughFlow example wouldn’t work for us because we have two flows: a command flow and an event flow. For example a publish command needs to provide a carry that is emitted with its correlating publish ack event. As well as there being two flows, there’s not guaranteed order in which event flow elements are received eg you may get a publish ack for a later message and then get one for an earlier one, or it may be that a ping req is received before you get a publish ack etc.

[huntc]

@ennru I've now fixed the Java API and updated the description of the PR accordingly.

[ennru]

LGTM.

[dwijnand]

JFYI BroadcastHub.of doesn't even use the class value. It's only there to help define the type parameter. So you can actually pass null and let type inference define type parameter. 😄