Spring integration MQTT publish & subscribe to mul

问题:

I am trying to build an application which subscribes to multiple mqtt topics, get the information, process it and form xmls and upon processing trigger an event so that these can be sent to some cloud server and the successful response from there to be sent back to the mqtt channel.

<int-mqtt:message-driven-channel-adapter
    id="mqttAdapter" client-id="${clientId}" url="${brokerUrl}" topics="${topics}"
    channel="startCase" auto-startup="true" />

<int:channel id="startCase" />

<int:service-activator id="startCaseService"
        input-channel="startCase" ref="msgPollingService" method="pollMessages" />

    <bean id="mqttTaskExecutor"
        class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
        <property name="corePoolSize" value="5" />
        <property name="maxPoolSize" value="10" />
    </bean>

    <bean id="msgPollingService" class="com.xxxx.xxx.mqttclient.mqtt.MsgPollingService">
        <property name="taskExecutor" ref="mqttTaskExecutor" />
        <property name="vendorId" value="${vendorId}" />
    </bean>

My question is how do I publish this to multiple channels, i.e. if I have an option to publish X message to Y topic. At present I have the below:

<int:channel id="outbound" />

<int-mqtt:outbound-channel-adapter
    id="mqtt-publish" client-id="kj" client-factory="clientFactory"
    auto-startup="true" url="${brokerUrl}" default-qos="0"
    default-retained="true" default-topic="${responseTopic}" channel="outbound" />

    <bean id="eventListner" class="com.xxxx.xxxx.mqttclient.event.EventListener">
        <property name="sccUrl" value="${url}" />
        <property name="restTemplate" ref="restTemplate" />
        <property name="channel" ref="outbound" />
    </bean>

I can publish this like:

channel.send(MessageBuilder.withPayload("customResponse").build());

Can I do something like:

channel.send(Message<?>, topic)

回答1:

Your configuration looks good. However the MessageChannel is an abstraction for loosely-coupling and gets deal only with Message.

So, you request a-la channel.send(Message<?>, topic) isn't correct for Messaging concepts.

However we have a trick for you. From AbstractMqttMessageHandler:

String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC);
.....
this.publish(topic == null ? this.defaultTopic : topic, mqttMessage, message);

So, what you need from your code is this:

channel.send(MessageBuilder.withPayload("customResponse").setHeader(MqttHeaders.TOPIC, topic).build());

In other words you should send a Message with mqtt_topic header to achieve a dynamic publication from <int-mqtt:outbound-channel-adapter>.

From other side we don't recommend to use MessageChannels directly from the application. The <gateway> with service interface is for such a case for end-application. Where that topic can be one of service method argument marked as @Header(MqttHeaders.TOPIC)