diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java index d76b9e1b5b..994168610e 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java @@ -44,6 +44,7 @@ public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapte builder.addConstructorArgValue(element.getAttribute("topics")); builder.addPropertyReference("outputChannel", channelName); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "qos"); return builder.getBeanDefinition(); } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java index d1b0697f02..f8f42d54d9 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java @@ -59,6 +59,8 @@ public class MqttOutboundChannelAdapterParser extends AbstractOutboundChannelAda } IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-qos"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-retained"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async-events"); return builder.getBeanDefinition(); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java index 48286dd900..0a6d8e1653 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java @@ -19,6 +19,7 @@ import java.util.Properties; import javax.net.SocketFactory; +import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttClientPersistence; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; @@ -112,6 +113,12 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory { return new MqttClient(uri == null ? "tcp://NO_URL_PROVIDED" : uri, clientId, this.persistence); } + @Override + public MqttAsyncClient getAsyncClientInstance(String uri, String clientId) throws MqttException { + // Client validates URI even if overridden by options + return new MqttAsyncClient(uri == null ? "tcp://NO_URL_PROVIDED" : uri, clientId, this.persistence); + } + @Override public MqttConnectOptions getConnectionOptions() { MqttConnectOptions options = new MqttConnectOptions(); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java index 8d86ae5c0c..3308fec038 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java @@ -15,6 +15,7 @@ */ package org.springframework.integration.mqtt.core; +import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; @@ -36,6 +37,17 @@ public interface MqttPahoClientFactory { */ MqttClient getClientInstance(String url, String clientId) throws MqttException; + /** + * Retrieve an async client instance. + * + * @param url The URL. + * @param clientId The client id. + * @return The client instance. + * @throws MqttException Any. + * @since 4.1 + */ + MqttAsyncClient getAsyncClientInstance(String url, String clientId) throws MqttException; + /** * Retrieve the connection options. * diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttIntegrationEvent.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttIntegrationEvent.java new file mode 100644 index 0000000000..e3aad3f34c --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttIntegrationEvent.java @@ -0,0 +1,37 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt.event; + +import org.springframework.integration.event.IntegrationEvent; + +/** + * Base class for Mqtt Events. + * @author Gary Russell + * + * @since 4.1 + */ +@SuppressWarnings("serial") +public abstract class MqttIntegrationEvent extends IntegrationEvent { + + public MqttIntegrationEvent(Object source) { + super(source); + } + + public MqttIntegrationEvent(Object source, Throwable cause) { + super(source, cause); + } + +} diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageDeliveredEvent.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageDeliveredEvent.java new file mode 100644 index 0000000000..9d200e5c4a --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageDeliveredEvent.java @@ -0,0 +1,44 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt.event; + + + +/** + * An event emitted (when using aysnc) when the client indicates the message + * was delivered. + * + * @author Gary Russell + * @since 4.1 + * + */ +@SuppressWarnings("serial") +public class MqttMessageDeliveredEvent extends MqttMessageDeliveryEvent { + + public MqttMessageDeliveredEvent(Object source, int messageId, String clientId, + int clientInstance) { + super(source, messageId, clientId, clientInstance); + } + + @Override + public String toString() { + return "MqttMessageSentEvent [clientId=" + getClientId() + + ", clientInstance=" + getClientInstance() + + ", messageId=" + getMessageId() + + "]"; + } + +} diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageDeliveryEvent.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageDeliveryEvent.java new file mode 100644 index 0000000000..4e572da197 --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageDeliveryEvent.java @@ -0,0 +1,55 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt.event; + + +/** + * Base class for events related to message delivery. Properties {@link #messageId}, + * {@link #clientId} and {@link #clientInstance} can be used to correlate events. + * + * @author Gary Russell + * @since 4.1 + * + */ +@SuppressWarnings("serial") +public abstract class MqttMessageDeliveryEvent extends MqttIntegrationEvent { + + private final int messageId; + + private final String clientId; + + private final int clientInstance; + + public MqttMessageDeliveryEvent(Object source, int messageId, String clientId, int clientInstance) { + super(source); + this.messageId = messageId; + this.clientId = clientId; + this.clientInstance = clientInstance; + } + + public int getMessageId() { + return messageId; + } + + public String getClientId() { + return clientId; + } + + public int getClientInstance() { + return clientInstance; + } + +} diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageSentEvent.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageSentEvent.java new file mode 100644 index 0000000000..4c73fd788d --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageSentEvent.java @@ -0,0 +1,60 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt.event; + +import org.springframework.messaging.Message; + +/** + * An event emitted (when using aysnc) when the client indicates that a message + * has been sent. + * + * @author Gary Russell + * @since 4.1 + * + */ +@SuppressWarnings("serial") +public class MqttMessageSentEvent extends MqttMessageDeliveryEvent { + + private final Message message; + + private final String topic; + + public MqttMessageSentEvent(Object source, Message message, String topic, int messageId, + String clientId, int clientInstance) { + super(source, messageId, clientId, clientInstance); + this.message = message; + this.topic = topic; + } + + public Message getMessage() { + return message; + } + + public String getTopic() { + return topic; + } + + @Override + public String toString() { + return "MqttMessageSentEvent [message=" + message + + ", topic=" + topic + + ", clientId=" + getClientId() + + ", clientInstance=" + getClientInstance() + + ", messageId=" + getMessageId() + + "]"; + } + +} diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/package-info.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/package-info.java new file mode 100644 index 0000000000..f7007f8b60 --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/package-info.java @@ -0,0 +1,4 @@ +/** + * ApplicationEvents generated by the mqtt module. + */ +package org.springframework.integration.mqtt.event; diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java index deb0799457..3b1ffdaded 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java @@ -35,6 +35,8 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro private final String[] topic; + private volatile int[] qos; + private volatile MqttMessageConverter converter; public AbstractMqttMessageDrivenChannelAdapter(String url, String clientId, String... topic) { @@ -45,6 +47,16 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro this.url = url; this.clientId = clientId; this.topic = topic; + // set the topic qos to 1 by default + this.qos = buildQosArray(1); + } + + private int[] buildQosArray(int value) { + int[] qos = new int[this.topic.length]; + for (int i = 0; i < qos.length; i++) { + qos[i] = value; + } + return qos; } public void setConverter(MqttMessageConverter converter) { @@ -52,6 +64,27 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro this.converter = converter; } + /** + * Set the QoS for each topic; a single value will apply to all topics otherwise + * the correct number of qos values must be provided. + * @param qos The qos value(s). + * @since 4.1 + */ + public void setQos(int... qos) { + Assert.notNull(qos, "'qos' cannot be null"); + if (qos.length == 1) { + this.qos = buildQosArray(qos[0]); + } + else { + Assert.isTrue(qos.length == this.topic.length); + this.qos = qos; + } + } + + protected int[] getQos() { + return qos; + } + protected String getUrl() { return url; } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java index 5549fe2d5e..9fef19f194 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java @@ -19,8 +19,8 @@ import java.util.Arrays; import java.util.concurrent.ScheduledFuture; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; +import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; -import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; @@ -40,14 +40,18 @@ import org.springframework.util.Assert; public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDrivenChannelAdapter implements MqttCallback { + private static final int DEFAULT_COMPLETION_TIMEOUT = 30000; + private final MqttPahoClientFactory clientFactory; - private volatile MqttClient client; + private volatile MqttAsyncClient client; private volatile ScheduledFuture reconnectFuture; private volatile boolean connected; + private volatile int completionTimeout = DEFAULT_COMPLETION_TIMEOUT; + /** * Use this constructor for a single url (although it may be overridden @@ -88,6 +92,16 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv this(url, clientId, new DefaultMqttPahoClientFactory(), topic); } + /** + * Set the completion timeout for async operations. Not settable using the namespace. + * Default 30000 milliseconds. + * @param completionTimeout The timeout. + * @since 4.1 + */ + public void setCompletionTimeout(int completionTimeout) { + this.completionTimeout = completionTimeout; + } + @Override protected void doStart() { super.doStart(); @@ -105,13 +119,15 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv this.cancelReconnect(); super.doStop(); try { - this.client.unsubscribe(this.getTopic()); + this.client.unsubscribe(this.getTopic()) + .waitForCompletion(this.completionTimeout); } catch (MqttException e) { logger.error("Exception while unsubscribing", e); } try { - this.client.disconnect(); + this.client.disconnect() + .waitForCompletion(this.completionTimeout); } catch (MqttException e) { logger.error("Exception while disconnecting", e); @@ -127,18 +143,21 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } private void connectAndSubscribe() throws MqttException { - this.client.setCallback(this); MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions(); Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null, "If no 'url' provided, connectionOptions.getServerURIs() must not be null"); - this.client = this.clientFactory.getClientInstance(this.getUrl(), this.getClientId()); - this.client.connect(connectionOptions); - + this.client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId()); + this.client.setCallback(this); + this.client.connect(connectionOptions) + .waitForCompletion(this.completionTimeout); try { - this.client.subscribe(this.getTopic()); + this.client.subscribe(this.getTopic(), this.getQos()) + .waitForCompletion(this.completionTimeout); } catch (MqttException e) { - this.client.disconnect(); + logger.error("Error subscribing to " + Arrays.asList(this.getTopic()), e); + this.client.disconnect() + .waitForCompletion(this.completionTimeout); throw e; } if (this.client.isConnected()) { diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java index d9f0832210..e8a508d827 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java @@ -16,15 +16,13 @@ package org.springframework.integration.mqtt.outbound; -import org.eclipse.paho.client.mqttv3.MqttMessage; - import org.springframework.context.SmartLifecycle; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.mqtt.support.MqttHeaders; -import org.springframework.integration.mqtt.support.MqttMessageConverter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.converter.MessageConverter; import org.springframework.util.Assert; /** @@ -46,7 +44,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler private volatile boolean defaultRetained = false; - private volatile MqttMessageConverter converter; + private volatile MessageConverter converter; private boolean running; @@ -54,6 +52,8 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler private volatile boolean autoStartup; + private volatile int clientInstance; + public AbstractMqttMessageHandler(String url, String clientId) { Assert.hasText(clientId, "'clientId' cannot be null or empty"); this.url = url; @@ -72,24 +72,41 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler this.defaultRetained = defaultRetain; } - public void setConverter(MqttMessageConverter converter) { + public void setConverter(MessageConverter converter) { Assert.notNull(converter, "'converter' cannot be null"); this.converter = converter; } + protected MessageConverter getConverter() { + return converter; + } + protected String getUrl() { return url; } - protected String getClientId() { + public String getClientId() { return clientId; } + /** + * Incremented each time the client is connected. + * @return The instance; + * @since 4.1 + */ + public int getClientInstance() { + return clientInstance; + } + @Override public String getComponentType() { return "mqtt:outbound-channel-adapter"; } + protected void incrementClientInstance() { + this.clientInstance++; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -145,16 +162,16 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler protected void handleMessageInternal(Message message) throws Exception { this.connectIfNeeded(); String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC); - MqttMessage mqttMessage = (MqttMessage) this.converter.fromMessage(message, MqttMessage.class); + Object mqttMessage = this.converter.fromMessage(message, Object.class); if (topic == null && this.defaultTopic == null) { throw new MessageHandlingException(message, "No '" + MqttHeaders.TOPIC + "' header and no default topic defined"); } - this.publish(topic == null ? this.defaultTopic : topic, mqttMessage); + this.publish(topic == null ? this.defaultTopic : topic, mqttMessage, message); } protected abstract void connectIfNeeded(); - protected abstract void publish(String topic, Object mqttMessage) throws Exception; + protected abstract void publish(String topic, Object mqttMessage, Message message) throws Exception; } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java index 6808b7ab76..9dfa8c55e7 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java @@ -16,14 +16,20 @@ package org.springframework.integration.mqtt.outbound; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; +import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; -import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; +import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent; +import org.springframework.integration.mqtt.event.MqttMessageSentEvent; +import org.springframework.integration.mqtt.support.MqttMessageConverter; +import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; @@ -35,11 +41,21 @@ import org.springframework.util.Assert; * */ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler - implements MqttCallback { + implements MqttCallback, ApplicationEventPublisherAware { + + private static final int DEFAULT_COMPLETION_TIMEOUT = 30000; + + private volatile int completionTimeout = DEFAULT_COMPLETION_TIMEOUT; private final MqttPahoClientFactory clientFactory; - private volatile MqttClient client; + private volatile MqttAsyncClient client; + + private volatile boolean async; + + private volatile boolean asyncEvents; + + private volatile ApplicationEventPublisher applicationEventPublisher; /** * Use this constructor for a single url (although it may be overridden @@ -75,6 +91,51 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler this(url, clientId, new DefaultMqttPahoClientFactory()); } + /** + * Set to true if you don't want to block when sending messages. Default false. + * When true, message sent/delivered events will be published for reception + * by a suitably configured 'ApplicationListener' or an event + * inbound-channel-adapter. + * @param async true for async. + * @since 4.1 + */ + public void setAsync(boolean async) { + this.async = async; + } + + /** + * When {@link #setAsync(boolean)} is true, setting this to true enables + * publication of {@link MqttMessageSentEvent} and {@link MqttMessageDeliveredEvent} + * to be emitted. Default false. + * @param asyncEvents the asyncEvents. + * @since 4.1 + */ + public void setAsyncEvents(boolean asyncEvents) { + this.asyncEvents = asyncEvents; + } + + /** + * Set the completion timeout for async operations. Not settable using the namespace. + * Default 30000 milliseconds. + * @param completionTimeout The timeout. + * @since 4.1 + */ + public void setCompletionTimeout(int completionTimeout) { + this.completionTimeout = completionTimeout; + } + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + Assert.state(getConverter() instanceof MqttMessageConverter, + "MessageConverter must be an MqttMessageConverter"); + } + @Override protected void doStart() { } @@ -83,7 +144,7 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler protected void doStop() { try { if (this.client != null) { - this.client.disconnect(); + this.client.disconnect().waitForCompletion(this.completionTimeout); this.client.close(); this.client = null; } @@ -102,9 +163,10 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions(); Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null, "If no 'url' provided, connectionOptions.getServerURIs() must not be null"); - this.client = this.clientFactory.getClientInstance(this.getUrl(), this.getClientId()); - this.client.connect(connectionOptions); + this.client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId()); + incrementClientInstance(); this.client.setCallback(this); + this.client.connect(connectionOptions).waitForCompletion(this.completionTimeout); if (logger.isDebugEnabled()) { logger.debug("Client connected"); } @@ -124,9 +186,25 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler } @Override - protected void publish(String topic, Object mqttMessage) throws Exception { + protected void publish(String topic, Object mqttMessage, Message message) throws Exception { Assert.isInstanceOf(MqttMessage.class, mqttMessage); - this.client.publish(topic, (MqttMessage) mqttMessage); + IMqttDeliveryToken token = this.client.publish(topic, (MqttMessage) mqttMessage); + if (!this.async) { + token.waitForCompletion(this.completionTimeout); + } + else if (this.asyncEvents && this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent( + new MqttMessageSentEvent(this, message, topic, token.getMessageId(), getClientId(), + getClientInstance())); + } + } + + private void sendDeliveryComplete(IMqttDeliveryToken token) { + if (this.async && this.asyncEvents && this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent( + new MqttMessageDeliveredEvent(this, token.getMessageId(), getClientId(), + getClientInstance())); + } } @Override @@ -142,7 +220,7 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler @Override public void deliveryComplete(IMqttDeliveryToken token) { - + sendDeliveryComplete(token); } } diff --git a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd index 9e8a5d280c..e7bbc5c187 100644 --- a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd +++ b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd @@ -43,6 +43,15 @@ + + + + Specifies the QoS to use when subscribing to topics; default '1'. This can be single + value (applying to all topics); otherwise it must be a comma-delimited list corresponding + to the provided topics (the name number of elements must be provided). + + + + + + + Specifies that sends should not block, with the thread returning + immediately the message is sent. When 'true', message + sent and message delivery events can be published; see 'async-events'. + Default: 'false'. + + + + + + + When 'async' is true, specifies that message + sent and message delivery events will be published for reception + by a suitably configured 'ApplicationListener' or an event + inbound-channel-adapter. + Default: 'false'. + + + diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml index ad8622c558..18318a209d 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/integration/mqtt http://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt.xsd"> - diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java new file mode 100644 index 0000000000..62bfa212b1 --- /dev/null +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java @@ -0,0 +1,293 @@ +/* + * Copyright 2002-2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.mqtt; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; + +import java.io.File; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.eclipse.paho.client.mqttv3.MqttClientPersistence; +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; +import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent; +import org.springframework.integration.mqtt.event.MqttMessageSentEvent; +import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; +import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; +import org.springframework.integration.mqtt.support.MqttHeaders; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class BackToBackAdapterTests { + + @ClassRule + public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883); + + @Autowired + private MessageChannel out; + + @Autowired + private PollableChannel in; + + @Test + public void testSingleTopic() { + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-foo"); + adapter.setBeanFactory(mock(BeanFactory.class)); + adapter.afterPropertiesSet(); + adapter.start(); + MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", + "si-test-in", "mqtt-foo"); + QueueChannel outputChannel = new QueueChannel(); + inbound.setOutputChannel(outputChannel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + inbound.setTaskScheduler(taskScheduler); + inbound.setBeanFactory(mock(BeanFactory.class)); + inbound.afterPropertiesSet(); + inbound.start(); + adapter.handleMessage(new GenericMessage("foo")); + adapter.stop(); + Message out = outputChannel.receive(1000); + assertNotNull(out); + inbound.stop(); + assertEquals("foo", out.getPayload()); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + } + + @Test + public void testTwoTopics() { + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-foo"); + adapter.setBeanFactory(mock(BeanFactory.class)); + adapter.afterPropertiesSet(); + adapter.start(); + MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", + "si-test-in", "mqtt-foo", "mqtt-bar"); + QueueChannel outputChannel = new QueueChannel(); + inbound.setOutputChannel(outputChannel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + inbound.setTaskScheduler(taskScheduler); + inbound.setBeanFactory(mock(BeanFactory.class)); + inbound.afterPropertiesSet(); + inbound.start(); + adapter.handleMessage(new GenericMessage("foo")); + Message message = MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build(); + adapter.handleMessage(message); + adapter.stop(); + Message out = outputChannel.receive(1000); + assertNotNull(out); + inbound.stop(); + assertEquals("foo", out.getPayload()); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + out = outputChannel.receive(1000); + assertNotNull(out); + inbound.stop(); + assertEquals("bar", out.getPayload()); + assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC)); + } + + @Test + public void testAsync() throws Exception { + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-foo"); + adapter.setBeanFactory(mock(BeanFactory.class)); + adapter.setAsync(true); + adapter.setAsyncEvents(true); + EventPublisher publisher = new EventPublisher(); + adapter.setApplicationEventPublisher(publisher); + adapter.afterPropertiesSet(); + adapter.start(); + MqttPahoMessageDrivenChannelAdapter inbound = + new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo"); + QueueChannel outputChannel = new QueueChannel(); + inbound.setOutputChannel(outputChannel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + inbound.setTaskScheduler(taskScheduler); + inbound.setBeanFactory(mock(BeanFactory.class)); + inbound.afterPropertiesSet(); + inbound.start(); + GenericMessage message = new GenericMessage("foo"); + adapter.handleMessage(message); + verifyEvents(adapter, publisher, message); + adapter.stop(); + Message out = outputChannel.receive(10000); + assertNotNull(out); + inbound.stop(); + assertEquals("foo", out.getPayload()); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + } + + @Test + public void testAsyncPersisted() throws Exception { + DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); + String tmpDir = System.getProperty("java.io.tmpdir") + File.separator + "mqtt_persist"; + new File(tmpDir).mkdirs(); + MqttClientPersistence persistence = new MqttDefaultFilePersistence(tmpDir); + factory.setPersistence(persistence); + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out", factory); + adapter.setDefaultTopic("mqtt-foo"); + adapter.setBeanFactory(mock(BeanFactory.class)); + adapter.setAsync(true); + adapter.setAsyncEvents(true); + adapter.setDefaultQos(1); + EventPublisher publisher1 = new EventPublisher(); + adapter.setApplicationEventPublisher(publisher1); + adapter.afterPropertiesSet(); + adapter.start(); + + MqttPahoMessageDrivenChannelAdapter inbound = + new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo", "mqtt-bar"); + QueueChannel outputChannel = new QueueChannel(); + inbound.setOutputChannel(outputChannel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + inbound.setTaskScheduler(taskScheduler); + inbound.setBeanFactory(mock(BeanFactory.class)); + inbound.afterPropertiesSet(); + inbound.start(); + Message message1 = new GenericMessage("foo"); + adapter.handleMessage(message1); + verifyEvents(adapter, publisher1, message1); + + Message message2 = MessageBuilder.withPayload("bar") + .setHeader(MqttHeaders.TOPIC, "mqtt-bar") + .build(); + EventPublisher publisher2 = new EventPublisher(); + adapter.setApplicationEventPublisher(publisher2); + adapter.handleMessage(message2); + verifyEvents(adapter, publisher2, message2); + + verifyMessageIds(publisher1, publisher2); + int clientInstance = publisher1.delivered.getClientInstance(); + + adapter.stop(); + adapter.start(); // new client instance + + publisher1 = new EventPublisher(); + adapter.setApplicationEventPublisher(publisher1); + adapter.handleMessage(message1); + verifyEvents(adapter, publisher1, message1); + + publisher2 = new EventPublisher(); + adapter.setApplicationEventPublisher(publisher2); + adapter.handleMessage(message2); + verifyEvents(adapter, publisher2, message2); + + verifyMessageIds(publisher1, publisher2); + + assertNotEquals(clientInstance, publisher1.delivered.getClientInstance()); + adapter.stop(); + + Message out = null; + for (int i = 0; i < 4; i++) { + out = outputChannel.receive(10000); + assertNotNull(out); + if ("foo".equals(out.getPayload())) { + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + } + else if ("bar".equals(out.getPayload())) { + assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC)); + } + else { + fail("unexpected payload " + out.getPayload()); + } + } + inbound.stop(); + } + + private void verifyEvents(MqttPahoMessageHandler adapter, EventPublisher publisher1, Message message1) + throws InterruptedException { + assertTrue(publisher1.latch.await(10, TimeUnit.SECONDS)); + assertNotNull(publisher1.sent); + assertNotNull(publisher1.delivered); + assertEquals(publisher1.sent.getMessageId(), publisher1.delivered.getMessageId()); + assertEquals(adapter.getClientId(), publisher1.sent.getClientId()); + assertEquals(adapter.getClientId(), publisher1.delivered.getClientId()); + assertEquals(adapter.getClientInstance(), publisher1.sent.getClientInstance()); + assertEquals(adapter.getClientInstance(), publisher1.delivered.getClientInstance()); + assertSame(message1, publisher1.sent.getMessage()); + } + + private void verifyMessageIds(EventPublisher publisher1, EventPublisher publisher2) { + assertNotEquals(publisher1.delivered.getMessageId(), publisher2.delivered.getMessageId()); + assertEquals(publisher1.delivered.getClientId(), publisher2.delivered.getClientId()); + assertEquals(publisher1.delivered.getClientInstance(), publisher2.delivered.getClientInstance()); + } + + @Test + public void testMultiURIs() { + out.send(new GenericMessage("foo")); + Message message = in.receive(10000); + assertNotNull(message); + assertEquals("foo", message.getPayload()); + } + + private class EventPublisher implements ApplicationEventPublisher { + + private volatile MqttMessageDeliveredEvent delivered; + + private MqttMessageSentEvent sent; + + private final CountDownLatch latch = new CountDownLatch(2); + + @Override + public void publishEvent(ApplicationEvent event) { + if (event instanceof MqttMessageSentEvent) { + this.sent = (MqttMessageSentEvent) event; + } + else if (event instanceof MqttMessageDeliveredEvent){ + this.delivered = (MqttMessageDeliveredEvent) event; + } + latch.countDown(); + } + + } + +} diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java deleted file mode 100644 index 7867c8c7a3..0000000000 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java +++ /dev/null @@ -1,129 +0,0 @@ -/* - * Copyright 2002-2014 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.integration.mqtt; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.mockito.Mockito.mock; - -import org.junit.Ignore; -import org.junit.Rule; -import org.junit.Test; -import org.junit.runner.RunWith; - -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; -import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; -import org.springframework.integration.mqtt.support.MqttHeaders; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.PollableChannel; -import org.springframework.messaging.support.GenericMessage; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -/** - * @author Gary Russell - * @since 4.0 - * - */ -@Ignore //TODO transiently -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) -@DirtiesContext -public class BackTobackAdapterTests { - - @Rule - public final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883); - - @Autowired - public MessageChannel out; - - @Autowired - public PollableChannel in; - - @Test - public void testSingleTopic() { - MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); - adapter.setDefaultTopic("mqtt-foo"); - adapter.setBeanFactory(mock(BeanFactory.class)); - adapter.afterPropertiesSet(); - adapter.start(); - MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo"); - QueueChannel outputChannel = new QueueChannel(); - inbound.setOutputChannel(outputChannel); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - inbound.setTaskScheduler(taskScheduler); - inbound.setBeanFactory(mock(BeanFactory.class)); - inbound.afterPropertiesSet(); - inbound.start(); - adapter.handleMessage(new GenericMessage("foo")); - adapter.stop(); - Message out = outputChannel.receive(1000); - assertNotNull(out); - inbound.stop(); - assertEquals("foo", out.getPayload()); - assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); - adapter.stop(); - } - - @Test - public void testTwoTopics() { - MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); - adapter.setDefaultTopic("mqtt-foo"); - adapter.setBeanFactory(mock(BeanFactory.class)); - adapter.afterPropertiesSet(); - adapter.start(); - MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo", "mqtt-bar"); - QueueChannel outputChannel = new QueueChannel(); - inbound.setOutputChannel(outputChannel); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - inbound.setTaskScheduler(taskScheduler); - inbound.setBeanFactory(mock(BeanFactory.class)); - inbound.afterPropertiesSet(); - inbound.start(); - adapter.handleMessage(new GenericMessage("foo")); - Message message = MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build(); - adapter.handleMessage(message); - adapter.stop(); - Message out = outputChannel.receive(1000); - assertNotNull(out); - inbound.stop(); - assertEquals("foo", out.getPayload()); - assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); - out = outputChannel.receive(1000); - assertNotNull(out); - inbound.stop(); - assertEquals("bar", out.getPayload()); - assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC)); - adapter.stop(); - } - - @Test - public void testMultiURIs() { - out.send(new GenericMessage("foo")); - Message message = in.receive(10000); - assertNotNull(message); - assertEquals("foo", message.getPayload()); - } -} diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java index d11c6e15b2..27972f2d48 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java @@ -32,7 +32,6 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.junit.ClassRule; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.invocation.InvocationOnMock; @@ -58,7 +57,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext -@Ignore //TODO transiently public class DownstreamExceptionTests { @ClassRule diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java index 42dc8cccc4..0b83b4c25f 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java @@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -34,10 +35,12 @@ import java.util.concurrent.atomic.AtomicReference; import javax.net.SocketFactory; +import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; -import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; +import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.eclipse.paho.client.mqttv3.MqttToken; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.junit.Test; import org.mockito.invocation.InvocationOnMock; @@ -107,25 +110,27 @@ public class MqttAdapterTests { factory.setWill(will); factory = spy(factory); - final MqttClient client = mock(MqttClient.class); - doAnswer(new Answer() { + final MqttAsyncClient client = mock(MqttAsyncClient.class); + doAnswer(new Answer() { @Override - public MqttClient answer(InvocationOnMock invocation) throws Throwable { + public MqttAsyncClient answer(InvocationOnMock invocation) throws Throwable { return client; } - }).when(factory).getClientInstance(anyString(), anyString()); + }).when(factory).getAsyncClientInstance(anyString(), anyString()); MqttPahoMessageHandler handler = new MqttPahoMessageHandler("foo", "bar", factory); handler.setDefaultTopic("mqtt-foo"); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); handler.start(); + + final MqttToken token = mock(MqttToken.class); final AtomicBoolean connectCalled = new AtomicBoolean(); - doAnswer(new Answer(){ + doAnswer(new Answer(){ @Override - public Object answer(InvocationOnMock invocation) throws Throwable { + public MqttToken answer(InvocationOnMock invocation) throws Throwable { MqttConnectOptions options = (MqttConnectOptions) invocation.getArguments()[0]; assertEquals(23, options.getConnectionTimeout()); assertEquals(45, options.getKeepAliveInterval()); @@ -137,19 +142,22 @@ public class MqttAdapterTests { assertEquals("bar", new String(options.getWillMessage().getPayload())); assertEquals(2, options.getWillMessage().getQos()); connectCalled.set(true); - return null; + return token; } }).when(client).connect(any(MqttConnectOptions.class)); + doReturn(token).when(client).subscribe(any(String[].class), any(int[].class)); + + final MqttDeliveryToken deliveryToken = mock(MqttDeliveryToken.class); final AtomicBoolean publishCalled = new AtomicBoolean(); - doAnswer(new Answer() { + doAnswer(new Answer() { @Override - public Object answer(InvocationOnMock invocation) throws Throwable { + public MqttDeliveryToken answer(InvocationOnMock invocation) throws Throwable { assertEquals("mqtt-foo", invocation.getArguments()[0]); MqttMessage message = (MqttMessage) invocation.getArguments()[1]; assertEquals("Hello, world!", new String(message.getPayload())); publishCalled.set(true); - return null; + return deliveryToken; } }).when(client).publish(anyString(), any(MqttMessage.class)); @@ -177,15 +185,16 @@ public class MqttAdapterTests { factory.setWill(will); factory = spy(factory); - final MqttClient client = mock(MqttClient.class); - doAnswer(new Answer() { + final MqttAsyncClient client = mock(MqttAsyncClient.class); + doAnswer(new Answer() { @Override - public MqttClient answer(InvocationOnMock invocation) throws Throwable { + public MqttAsyncClient answer(InvocationOnMock invocation) throws Throwable { return client; } - }).when(factory).getClientInstance(anyString(), anyString()); + }).when(factory).getAsyncClientInstance(anyString(), anyString()); + final MqttToken token = mock(MqttToken.class); final AtomicBoolean connectCalled = new AtomicBoolean(); doAnswer(new Answer() { @@ -202,9 +211,10 @@ public class MqttAdapterTests { assertEquals("bar", new String(options.getWillMessage().getPayload())); assertEquals(2, options.getWillMessage().getQos()); connectCalled.set(true); - return null; + return token; } }).when(client).connect(any(MqttConnectOptions.class)); + doReturn(token).when(client).subscribe(any(String[].class), any(int[].class)); final AtomicReference callback = new AtomicReference(); doAnswer(new Answer() { diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml index da463b9b7e..e30bb881bf 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml @@ -28,6 +28,19 @@ client-id="foo" url="tcp://localhost:1883" topics="bar, baz" + qos="0, 2" + converter="myConverter" + client-factory="clientFactory" + send-timeout="123" + channel="out" /> + + diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java index cca4047a3d..0b075c18b1 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java @@ -16,7 +16,11 @@ package org.springframework.integration.mqtt.config.xml; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import org.hamcrest.Matchers; import org.junit.Test; @@ -70,6 +74,8 @@ public class MqttOutboundChannelAdapterParserTests { assertEquals("bar", TestUtils.getPropertyValue(withConverterHandler, "defaultTopic")); assertSame(converter, TestUtils.getPropertyValue(withConverterHandler, "converter")); assertSame(clientFactory, TestUtils.getPropertyValue(withConverterHandler, "clientFactory")); + assertFalse(TestUtils.getPropertyValue(withConverterHandler, "async", Boolean.class)); + assertFalse(TestUtils.getPropertyValue(withConverterHandler, "asyncEvents", Boolean.class)); Object handler = TestUtils.getPropertyValue(this.withConverterEndpoint, "handler"); @@ -96,6 +102,8 @@ public class MqttOutboundChannelAdapterParserTests { assertEquals(1, TestUtils.getPropertyValue(defaultConverter, "defaultQos")); assertTrue(TestUtils.getPropertyValue(defaultConverter, "defaultRetained", Boolean.class)); assertSame(clientFactory, TestUtils.getPropertyValue(withDefaultConverterHandler, "clientFactory")); + assertTrue(TestUtils.getPropertyValue(withDefaultConverterHandler, "async", Boolean.class)); + assertTrue(TestUtils.getPropertyValue(withDefaultConverterHandler, "asyncEvents", Boolean.class)); } } diff --git a/src/reference/docbook/mqtt.xml b/src/reference/docbook/mqtt.xml index cdc7d22581..764e8a0162 100644 --- a/src/reference/docbook/mqtt.xml +++ b/src/reference/docbook/mqtt.xml @@ -9,7 +9,7 @@ Spring Integration provides inbound and outbound channel adapters supporting the MQ Telemetry Transport (MQTT) protocol. The current implementation uses the Eclipse Paho MQTT Client - library. + library. Configuration of both adapters is achieved using the @@ -44,7 +44,8 @@ A comma delimited list of topics from which this adapter will receive messages. + + A comma delimited list of QoS values. Can be a single value that is applied to all + topics, or a value for each topic (in which case the lists must the same length). + An MqttMessageConverter (optional). The default DefaultPahoMessageConverter produces a message with a String @@ -103,6 +108,11 @@ is wrapped in a ConsumerEndpoint. For convenience, it can be configured using the namespace. + + Starting with version 4.1, the adapter supports asynchronous sends, avoiding + blocking until the delivery is confirmed; application events can be emitted to enable applications + to confirm delivery if desired. + Attributes: @@ -114,6 +124,8 @@ default-qos="1"]]> ]]> @@ -132,7 +144,7 @@ An MqttMessageConverter (optional). The default - DefaultPahoMessageConverter + DefaultPahoMessageConverter recognizes the following headers: mqtt_topic - the topic to which the message will be sent @@ -154,6 +166,25 @@ The default topic to which the message will be sent (used if no mqtt_topic header is found). + + When true, the caller will not block waiting for delivery confirmation when a message is + sent. + Default:false (the send blocks until delivery is confirmed). + + + When async and async-events are both true, an + MqttMessageSentEvent is emitted, containing the message, the + topic, the messageId generated by the client library, the clientId + and the clientInstance (incremented each time the client is connected). + When the delivery is confirmed by the client library, an + MqttMessageDeliveredEvent is emitted, containing the the messageId, + clientId and the clientInstance, enabling + delivery to be correlated with the send. These events can be received by any + ApplicationListener, or by an event inbound channel adapter. Note that + it is possible that the MqttMessageDeliveredEvent might be received before + the MqttMessageSentEvent. + Default: false. + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index e99b780ef8..5f37c4bbc2 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -54,13 +54,21 @@ See for more information. -
+
MQTT Adapter Changes The MQTT channel adapters can now be configured to connect to multiple servers, for example, to support High Availability (HA). See for more information. + + The MQTT message-driven channel adapter now supports specifying the QoS setting for each + subscription. See for more information. + + + The MQTT outbound channel adapter now supports asynchronous sends, avoiding blocking + until delivery is confirmed. See for more information. +