Switch to using reactive streams based APIs for Ditto managed MQTT connections #1411
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
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
…o accept a Throwable. This makes it easier to work with this method while handling CompletionStage results. The functionality of ExceptionToAcknowledgementConverter remains as is. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
This makes it easier to access. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Also added a package for abstracting MQTT Publish messages from their protocol version. This abstraction avoids split processing logic which makes it easier to understand what is going on. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Also added a package for abstracting MQTT Publish messages from their protocol version. This abstraction avoids split processing logic which makes it easier to understand what is going on. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
…ure/reactive_mqtt # Conflicts: # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/consuming/package-info.java # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/publish/package-info.java
…d of `Exception`. This change does not affect functionality of `ConnectionMonitor` at all, but it makes it easier to use without superfluous `instanceof` checks. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
The new actor operates on a generic MQTT Publish message and thus is suitable for MQTT version 3 and 5. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
This package provides all means to subscribe for topics at the MQTT broker. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Furthermore, * moved `InvalidMQttQosCodeException` to package `common` as it is used at another place, too, * added `ChildActorNanny` for starting and stopping child actors of `BaseClientActor`, * removed client ID from BasePublisherActor as it was unused. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Restructuring was necessary because there is not always a separate consumer and publisher client; both could use the same HiveMQ client. Therefore, subscribing client and publishing client were moved to client package and got hidden beyond the new `GenericMqttClient` interface. More changes: * Oriented package structure for generic messages toward HiveMQ API. * Got rid of `SourceSubscribeResult` because now each `SubscribeResult` has a connection `Source`. * Small improvements in `BaseClientActor`. * Moved `MqttClientConnected` from `AbstractMqttClientActor` in containing package. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
* Legacy MQTT implementation still has to be removed after testing. * Refactored `GenericMqttClientFactory` to be a utility class instead of an interface because it did not make sense to have an instance of this factory. This entailed some static mocking in `GenericMqttClientActorTest` and `ConnectionTesterTest`. * Provide `MqttSpecificConfig` to `HiveMqttClientProperties` because this config is required in `GenericMqttActor` anyway. This prevents duplicate creation of `MqttSpecificConfig`. * Implemented possibility of reconnecting of consumer client to `GenericMqttActor`. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
It is crucial that `ClientDisconnected` gets emitted even if the client was disconnected previously and is not available anymore. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
…it was its only inheritor. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
# Conflicts: # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BaseConsumerActor.java # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/AbstractMqttValidator.java # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/AbstractHiveMqttClientFactory.java # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/AbstractMqttClientActor.java # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/AbstractMqttSubscriptionHandler.java # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/DefaultHiveMqtt3ClientFactory.java # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/DefaultHiveMqtt5ClientFactory.java # connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/HiveMqttClientFactory.java # connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/AbstractMqttClientActorTest.java # connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/MockHiveMqtt3ClientFactory.java # connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/MockHiveMqtt5ClientFactory.java
…nnection. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
…queue fails. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
…suming and publishing, too, do not have 'Generic' in their name. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
- ChildActorNanny cannot be thread-safe; label it as such. - Add thread-safe wrapper AsyncChildActorNanny. - Add health check to MqttPublisherActor and MqttConsumerActor. - Move CompletableFutureUtils to ditto.base.service for reuse. Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
Doing so made AsyncChildActorNanny and ChildActorNannyActor superfluous because thread-safe child actor creation now can take place in ConnectionTesterActor itself. Made SubscriptionsFailedException public in order to use it directly to spare instanceof checks. Changed getter for failed subscription statuses to return a Stream instead of a List. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Unit testing connection testing now does not work anymore because having made ConnectionTester an actor eliminated possibilities for mocking. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
This allows to throttle incoming QoS 1 and QoS 2 publish messages. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
…ttConfig. Rationale: This value is set by the Ditto operator and cannot be set by users themselves. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
… TestConnectionActor. * Moved flag for disabling Last Will message to HiveMqttClientProperties. * Improved ConnectionTesterActor to be more sensitive to possible errors and to disconnect the client whenever possible in case of errors. * Renamed method in MqttSpecificConfig to comply to convention for names boolean getters values. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
Furthermore, removed unused MqttPublishTarget. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
thjaeckle
changed the title
The test fails on resource-constrained Github builder. Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
This pull request was closed.
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.
Rewrote Ditto's MQTT backend.
The client now consumes publish messages in a reactive manner. Together with throttling this effectively enables backpressure.
Furthermore, the code could be made simpler by unifying the processing algorithms instead of splitting them for different protocol versions; this was reached by providing a unified abstraction of HiveMQ's APIs for MQTT 3.x and 5.