Add Inbound Channel Adapter for MQTT based on hivemq#10989
Conversation
| GenericContainer<?> HIVEMQ_CONTAINER = new GenericContainer<>("hivemq/hivemq-ce:2024.3") | ||
| .withExposedPorts(CONTAINER_PORT) | ||
| .withCreateContainerCmdModifier(cmd -> cmd.withHostConfig(new HostConfig() | ||
| .withPortBindings(new PortBinding(Ports.Binding.bindPort(MAPPED_PORT), new ExposedPort(CONTAINER_PORT))) |
There was a problem hiding this comment.
due to mapped port will be changed after container stop then start,
I haven't found a way to set the new mapped port to the already built MqttClient,
so nows set a fixed mapped port here.
Related to: spring-projects#3102 - introduce a new `spring-integration-hivemq` module for MQTT adapters. - basically, the inbound message driven adapters are based on the `MqttClient`(either `Mqtt5AsyncClient` or `Mqtt3AsyncClient`) and a `topic`. - abstract common properties setup (topic/qos/manualAck/executor) for both mqtt v3 and v5 from hivemq side. - abstract common properties setup (messageConverter/payloadType) for both mqtt v3 and v5 from SI side. - for Mqtt5MessageDrivenChannelAdapter, there are some additional properties (noLocal/retainHandling/retainAsPublished) for subscription, also a `Mqtt5HeaderMapper` is used for mapping v5 specific headers like contentType/responseTopic/correlationData/UserProperties. - introduce `MqttClientConnectionCoordinator` for coordinating MQTT client connections and disconnections. avoid race-induced state exception. these v3 and v5 Coordinators are intent for internal use only. - provide `setMqttConnect(MqttConnect)` in v5 and `setMqtt3ConnectView(Mqtt3ConnectView)` in v3, if specific setup needed in connect. for example: cleanStart/keepAlive. - if connect or subscribe failed, a `MqttConnectionFailedEvent` will be published. after subscribe success, a `MqttSubscribedEvent` will be published. - if AutomaticReconnect is applied by the mqttClient, after reconnect, topic will be automatic subscribed as well. Signed-off-by: Jiandong Ma <jiandong.ma.cn@gmail.com>
|
here is the PR in my own repo for outbound v3/v5 adapters (mjd507#1). |
| } | ||
| } | ||
|
|
||
| project('spring-integration-hivemq') { |
There was a problem hiding this comment.
Thank you for doing this contribution!
I know that I've asked about this name in the issue, but now when I think about this again it looks like the name is wrong.
This module is the client for MQTT protocol and doesn't matter what is the broker (or peer) is there we are connecting to.
Let's see if we can rename it into spring-integration-mqtt-client!
I am considering to deprecated existing module based on Paho.
Then one day we will remove Paho dependency and rename this new module back to just spring-integration-mqtt.
It is a bit awkward to name it hivemq when it is going to work with any MQTT.
Isn't it?
Sorry for confusion; and I'm open for any other suggestions!
Note: this new feature is aimed at least for the next 7.2 due upcoming November release cycle.
Therefore, bear with us how slow we are responding to your effort.
There was a problem hiding this comment.
I am ok to go ahead with the name spring-integration-mqtt-client.
Just It might bring a little confusion for end users to adjust the dependency, we might take effort to explain why name as this and finally as that.
2 participants
Related to: #3102
spring-integration-hivemqmodule for MQTT adapters.MqttClient(eitherMqtt5AsyncClientorMqtt3AsyncClient) and atopic.Mqtt5HeaderMapperis used for mapping v5 specific headers like contentType/responseTopic/correlationData/UserProperties.MqttClientConnectionCoordinatorfor coordinating MQTT client connections and disconnections. avoid race-induced state exception. these v3 and v5 Coordinators are intent for internal use only.setMqttConnect(MqttConnect)in v5 andsetMqtt3ConnectView(Mqtt3ConnectView)in v3, if specific setup needed in connect. for example: cleanStart/keepAlive.MqttConnectionFailedEventwill be published. after subscribe success, aMqttSubscribedEventwill be published.