spring-integration-mqtt inbound & outbound client

[YGwan]

Hello,
I would like to ask you a question regarding inbound outbound Clients.
As a result of checking the example in the official document related to spring-integration-mqtt, it is confirmed that the inbound outbound client ID is set differently and the message channel is set differently to manage it.
Also, I was able to confirm that the Client Manager can be implemented and handled for the MQTT Client shared at the end.

I wonder in what situation 2 methods are used, separating and not separating
Also, if you could briefly explain what each has its pros and cons, I think we can refer to it for future learning. Thank you.

[artembilan]

Sorry, we don’t develop here final products, so cannot imaging real use-cases. The choice of such an API is just for flexibility where you might have only one or another channel adapter . And even if you have both, that does not mean that they have to be against the same client. Other brokers might not let more than one client per IP address. That’s why we g go ave introduced that ClientManager.
There is no difference in behavior between internal client and managed externally .

[artembilan]

The error you show (it is not full though), has nothing to do with MQTT Client ID.
try to imaging solution with just plain MQTT Driver, then apply over it Spring Integration flow.
Please, consider to remove the second subscriber on the channel you use in that mqttOutboundHandler.
Then share the whole stack trace for the error.
Right now I don't really see what is your concern and what we are trying to solve.

[YGwan]

My concern is whether it is necessary to separate the message channel exclusively for publishing and the message channel exclusively for subscribers.

For example, we have confirmed that the above error occurs when using a combination of a publish-related message channel and a subscription-related message channel.
However if the two channels are separated, publish multiple topics on the publishing-only message channel and processing multiple subscriptions on the subscription-only message channel was no problem.

Therefore, for this reason, I would like to ask if we should separate the publish-related message channel (outbound) and the subscribe-related message channel (inbound).

Or, are you saying that the reason for this error is not because of the message channel?

[artembilan]

The point is that we don’t know what is the error. You just didn’t share that with us: only leftover after failover which, I guess , not suppose to happen here.
So, sounds like the question has shifted from the MQTT client to channel configuration . Can you share with us that problematic configuration, though?
Might be even the case to start a new discussion with a proper context and info.

[YGwan]

@artembilan
I'm sorry. You didn't have enough prior explanation.

I basically added the environment and dependency below.

springboot : version "3.3.4"
io.spring.dependency-management : version "1.1.3"
kotlin("jvm") : version "1.9.20"
kotlin("plugin.spring") : version "1.9.20"

implementation("org.springframework.integration:spring-integration-mqtt:6.3.4")

and my MQTTConfig class is

@Bean
fun clientManager(): ClientManager<IMqttAsyncClient, MqttConnectOptions> {
    val connectionOptions = mqttConnectOptions()

    val clientManager = Mqttv3ClientManager(connectionOptions, "mqttServerClient")
    clientManager.setPersistence(MqttDefaultFilePersistence())

    return clientManager
}

private fun mqttConnectOptions(): MqttConnectOptions {
    val options = MqttConnectOptions().apply {
        serverURIs = arrayOf(mqttBrokerUrl)
        socketFactory = sslSocketFactory()  # to connect AWS IOT
        isAutomaticReconnect = true
    }
    return options
}

@Bean
@ServiceActivator(inputChannel = "messageChannel")
fun mqttOutboundHandler(): MessageHandler {
    val handler = MqttPahoMessageHandler(clientManager())
    handler.setAsync(true)
    return handler
}

@Bean
@ServiceActivator(inputChannel = "messageChannel")
fun mqttInboundHandler(): MessageHandler = MessageHandler { message ->
  val topic = message.headers[MqttHeaders.RECEIVED_TOPIC] as String
  val payload = message.payload.toString()

  mqttMessageHandlerDispatcher.dispatch(topic, payload)
}

@Bean
fun mqttInboundAdapter(): MqttPahoMessageDrivenChannelAdapter {
  val adapter = MqttPahoMessageDrivenChannelAdapter(
      clientManager(),
      "test/topic",
  )

  adapter.setConverter(DefaultPahoMessageConverter())
  adapter.outputChannel = messageChannel()
  return adapter
}

@Bean
fun messageChannel(): MessageChannel = DirectChannel()

@MessagingGateway(defaultRequestChannel = "messageChannel")
interface MQTTGateway {

    fun sendToMqtt(
        @Header(MqttHeaders.TOPIC) topic: String,
        message: String,
    )
}

The above situation is that the InboundAdapter and the OutboundHandler share a directChannel() called the same messageChannel. If you share a message channel as above, In this state, when the subscribed message reaches the server, Unicast Dispatcher INFO as above occurs.

but, If separate message channel inbound & outbound
dont occur this INFO.

thanks!!

[artembilan]

That is very strange configuration.
So, your mqttInboundAdapter produces consumed messages to the messageChannel.
That channel is a DirectChannel with a round-robin strategy for its subscribers.
Now you have two subscribers for this channel: mqttOutboundHandler and one more mqttInboundHandler - those @ServiceActivators.

The problem with DirectChannel that it hands of sent message to only one subscriber and that one is determined by the looping index. As I said - round-robin.
So, your intention for those mqttInboundHandler and mqttOutboundHandler might be not how it works.
You need to revise if you indeed need to use the same channel for the mqttOutboundHandler.
Plus you show MQTTGateway which produces messages to the same channel.
You also show mqttMessageHandlerDispatcher for which we don't know what it is doing.

[YGwan]

Thank you for your kind explanation.

So, your mqttInboundAdapter produces consumed messages to the messageChannel.
That channel is a DirectChannel with a round-robin strategy for its subscribers.
Now you have two subscribers for this channel: mqttOutboundHandler and one more mqttInboundHandler - those @ServiceActivators.

In other words, the problem in the current code is that the subscribers to the message channel are currently mqttOutboundHandler and mqttInboundHandler, and the channel is using round robin to select one subscriber in a loop-indexed fashion to deliver messages only to that subscriber. that's correct?

So the solution is to separate the message channels that the two handlers are subscribed to, and that will solve the problem.

Plus you show MQTTGateway which produces messages to the same channel.

@MessagingGateway(defaultRequestChannel = "messageChannel")
interface MQTTGateway {

    fun sendToMqtt(
        @Header(MqttHeaders.TOPIC) topic: String,
        message: String,
    )
}

Isn't this handled by the code above?

You also show mqttMessageHandlerDispatcher for which we don't know what it is doing.
Don't consider the mqttMessageHandlerDispatcher class as an additional class simply related to topic-specific processing!!!
I should have removed it, but didn't, sorry.

[artembilan]

See more details about DirectChannel in docs: https://docs.spring.io/spring-integration/reference/channel/implementations.html#channel-implementations-directchannel.

No, that definitely not clear from the code you show how your MQTTGateway is used, but suspect it is something what really produces to the mentioned before mqttOutboundHandler. So, yeah, the channel here has to be totally different. Otherwise you mess up a bit with whatever is received from the topic .
Not sure what have driven you to have only one channel for such not connected logic…

[YGwan]

No, that definitely not clear from the code you show how your MQTTGateway

The MQTT Gateway is used by the service logic to publish messages to the MQTT Broker.
like,

@Service
public class PublishService {

 void publish(String topic, String message) {
   mqttGateway.sendToMqtt(topic, message)
  }
}

but suspect it is something what really produces to the mentioned before mqttOutboundHandler.

i dont understand this sentence. Like the example code in the spring integration mqtt official documentation(https://docs.spring.io/spring-integration/reference/mqtt.html#mqtt-outbound), where I created a mqtt message channel, created a Gateway interface, took the defaultRequestChannel to be the same as the message channel I created, and connected to @serviceactivator(inputChannel = “messageChannel”) in the mqttOutboundHandle, what's wrong with my mqttOutboundHandler settings? I thought the problem is caused by the inputChannel being shared by InboundHandler and OutboundHandler, but...

[artembilan]

Yes, i meant how that gateway is used in your logic.
You change either this channel, or the one for inbound part.
Still not clear why you have made a choice just for one channel for everything…
See Spring Integration docs for the purpose of channels : https://docs.spring.io/spring-integration/reference/overview.html#overview-components-channel

[YGwan]

Still not clear why you have made a choice just for one channel for everything…

that is only test. This was simply to test whether we needed to separate the message channels. That should be sufficient explanation.

I think my questions have been answered so far with your answers and that link, thank you.