Open DerSchwilk opened 2 years ago
@DerSchwilk and I encountered that behaviour while including the reactive API of HiveMQ client in Eclipse Ditto.
We run our system-tests with Eclipse Mosquito, that's where we noticed the strange behaviour change when using FlowableWithSingle
.
When calling
subscribeSingleFuture()
on aFlowableWithSingle<MqttXPublish, MqttXSubAck>
i would expect that the original flowable behaves the same as if I would process it with thesubscribeSingleFuture()
call. If I run thesubscribeSingleFuture()
the flow should still process all MqttPublishes received by the subscription.We publish >20 message to the client. When running
subscribeSingleFuture()
on theFlowableWithSingle
and manually acking the publishes on the original flowable, the original flowable only emits 20 elements, capped by the max in-flight messages of mosquitto.To Reproduce
In our scenario we want to evaluate the SubAck prior to processing and consuming the publishes.
subscribePublishes(Mqtt5Subscribe, true)
FlowableWithSingle
in a variablesubscribeSingleFuture()
on itFlowableWithSingle
FlowableWithSingle
`final var flowableWithSingle = client.subscribePublishes(Mqtt5Subscribe.builder().topicFilter(TOPIC).qos(MqttQos.EXACTLY_ONCE).build(), true);
Without calling the
subscribeSingleFuture()
the messages are acked correctly and every message is received.Details