diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java index 103d6782f4..ecb2ac38b4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java @@ -65,6 +65,11 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { public static final String DELIVERY_ATTEMPT = "deliveryAttempt"; + /** + * A callback to acknowledge message delivery. The type of the header value depends on + * the context in which the header is used. See the reference manual for more + * information. + */ public static final String ACKNOWLEDGMENT_CALLBACK = "acknowledgmentCallback"; /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java index 93cd0a96b7..7172cc85c4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -21,6 +21,7 @@ import java.util.UUID; import java.util.concurrent.atomic.AtomicInteger; import org.springframework.integration.acks.AcknowledgmentCallback; +import org.springframework.integration.acks.SimpleAcknowledgment; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; @@ -107,6 +108,12 @@ public final class StaticMessageHeaderAccessor { AcknowledgmentCallback.class); } + @Nullable + public static SimpleAcknowledgment getAcknowledgment(Message message) { + return message.getHeaders().get(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, + SimpleAcknowledgment.class); + } + @SuppressWarnings("unchecked") @Nullable public static T getSourceData(Message message) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/acks/AcknowledgmentCallback.java b/spring-integration-core/src/main/java/org/springframework/integration/acks/AcknowledgmentCallback.java index e5ecc137ca..8fe527a83c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/acks/AcknowledgmentCallback.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/acks/AcknowledgmentCallback.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -25,7 +25,7 @@ package org.springframework.integration.acks; * */ @FunctionalInterface -public interface AcknowledgmentCallback { +public interface AcknowledgmentCallback extends SimpleAcknowledgment { /** * Acknowledge the message. @@ -33,6 +33,11 @@ public interface AcknowledgmentCallback { */ void acknowledge(Status status); + @Override + default void acknowledge() { + acknowledge(Status.ACCEPT); + } + /** * Implementations must implement this to indicate when the ack has been * processed by the user so that the framework can auto-ack if needed. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/acks/SimpleAcknowledgment.java b/spring-integration-core/src/main/java/org/springframework/integration/acks/SimpleAcknowledgment.java new file mode 100644 index 0000000000..07f9a0f10f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/acks/SimpleAcknowledgment.java @@ -0,0 +1,34 @@ +/* + * Copyright 2020 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 + * + * https://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.acks; + +/** + * Opaque object for manually acknowledging. + * + * @author Gary Russell + * @since 5.3 + * + */ +@FunctionalInterface +public interface SimpleAcknowledgment { + + /** + * Acknowledge the message delivery. + */ + void acknowledge(); + +} 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 cc05cffa58..91b75f1c3b 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -48,6 +48,7 @@ public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapte IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "qos"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "manual-acks"); return builder.getBeanDefinition(); } 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 226b56d021..23e69a6906 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 @@ -30,11 +30,14 @@ import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.acks.SimpleAcknowledgment; import org.springframework.integration.mqtt.core.ConsumerStopAction; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent; import org.springframework.integration.mqtt.event.MqttSubscribedEvent; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; @@ -71,6 +74,8 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv private long disconnectCompletionTimeout = DISCONNECT_COMPLETION_TIMEOUT; + private boolean manualAcks; + private volatile IMqttClient client; private volatile ScheduledFuture reconnectFuture; @@ -152,6 +157,15 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv this.recoveryInterval = recoveryInterval; } + /** + * Set the acknowledgment mode to manual. + * @param manualAcks true for manual acks. + * @since 5.3 + */ + public void setManualAcks(boolean manualAcks) { + this.manualAcks = manualAcks; + } + /** * @since 4.2.2 */ @@ -263,6 +277,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv String[] topics = getTopic(); try { this.client.connect(connectionOptions); + this.client.setManualAcks(this.manualAcks); int[] requestedQos = getQos(); int[] grantedQos = Arrays.copyOf(requestedQos, requestedQos.length); this.client.subscribe(topics, grantedQos); @@ -365,7 +380,12 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv @Override public void messageArrived(String topic, MqttMessage mqttMessage) { - Message message = this.getConverter().toMessage(topic, mqttMessage); + AbstractIntegrationMessageBuilder builder = getConverter().toMessageBuilder(topic, mqttMessage); + if (this.manualAcks) { + builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, + new AcknowledgmentImpl(mqttMessage.getId(), mqttMessage.getQos(), this.client)); + } + Message message = builder.build(); try { sendMessage(message); } @@ -379,4 +399,46 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv public void deliveryComplete(IMqttDeliveryToken token) { } + /** + * Used to complete message arrival when {@link AckMode#MANUAL}. + * + * @since 5.3 + */ + private static class AcknowledgmentImpl implements SimpleAcknowledgment { + + private final int id; + + private final int qos; + + private final IMqttClient ackClient; + + /** + * Construct an instance with the provided properties. + * @param id the message id. + * @param qos the message QOS. + * @param client the client. + */ + AcknowledgmentImpl(int id, int qos, IMqttClient client) { + this.id = id; + this.qos = qos; + this.ackClient = client; + } + + @Override + public void acknowledge() { + if (this.ackClient != null) { + try { + this.ackClient.messageArrivedComplete(this.id, this.qos); + } + catch (MqttException e) { + throw new IllegalStateException(e); + } + } + else { + throw new IllegalStateException("Client has changed"); + } + } + + } + } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java index 91cb51d255..eb40a5e343 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -206,6 +206,11 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa @Override public Message toMessage(String topic, MqttMessage mqttMessage) { + return toMessageBuilder(topic, mqttMessage).build(); + } + + @Override + public AbstractIntegrationMessageBuilder toMessageBuilder(String topic, MqttMessage mqttMessage) { try { AbstractIntegrationMessageBuilder messageBuilder; if (this.bytesMessageMapper != null) { @@ -219,13 +224,14 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa .withPayload(mqttBytesToPayload(mqttMessage)); } messageBuilder + .setHeader(MqttHeaders.ID, mqttMessage.getId()) .setHeader(MqttHeaders.RECEIVED_QOS, mqttMessage.getQos()) .setHeader(MqttHeaders.DUPLICATE, mqttMessage.isDuplicate()) .setHeader(MqttHeaders.RECEIVED_RETAINED, mqttMessage.isRetained()); if (topic != null) { messageBuilder.setHeader(MqttHeaders.RECEIVED_TOPIC, topic); } - return messageBuilder.build(); + return messageBuilder; } catch (Exception e) { throw new MessageConversionException("failed to convert object to Message", e); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaderAccessor.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaderAccessor.java new file mode 100644 index 0000000000..5e42b0b85f --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaderAccessor.java @@ -0,0 +1,84 @@ +/* + * Copyright 2020 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 + * + * https://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.support; + +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; + +/** + * Helper for typed access to incoming MQTT message headers. + * + * @author Gary Russell + * @since 5.3 + * + */ +public final class MqttHeaderAccessor { + + private MqttHeaderAccessor() { + } + + /** + * Return the received topic header. + * @param message the message. + * @return the header. + */ + @Nullable + public static String receivedTopic(Message message) { + return message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class); + } + + /** + * Return the MQTT message id. + * @param message the message. + * @return the header. + */ + @Nullable + public static Integer id(Message message) { + return message.getHeaders().get(MqttHeaders.ID, Integer.class); + } + + /** + * Return the received QOS header. + * @param message the message. + * @return the header. + */ + @Nullable + public static Integer receivedQos(Message message) { + return message.getHeaders().get(MqttHeaders.RECEIVED_QOS, Integer.class); + } + + /** + * Return the received retained header. + * @param message the message. + * @return the header. + */ + @Nullable + public static Boolean receivedRetained(Message message) { + return message.getHeaders().get(MqttHeaders.RECEIVED_RETAINED, Boolean.class); + } + + /** + * Return the duplicate header. + * @param message the message. + * @return the header. + */ + @Nullable + public static Boolean duplicate(Message message) { + return message.getHeaders().get(MqttHeaders.DUPLICATE, Boolean.class); + } + +} diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java index 015cf0bd5e..727f95cf69 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -30,6 +30,8 @@ public final class MqttHeaders { public static final String QOS = PREFIX + "qos"; + public static final String ID = PREFIX + "id"; + public static final String RECEIVED_QOS = PREFIX + "receivedQos"; public static final String DUPLICATE = PREFIX + "duplicate"; diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java index 7ec6a2bd12..287163e113 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -19,6 +19,7 @@ package org.springframework.integration.mqtt.support; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.integration.handler.MessageProcessor; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.converter.MessageConverter; @@ -35,12 +36,20 @@ public interface MqttMessageConverter extends MessageConverter { /** * Convert to a Message. * - * @param topic The topic. - * @param mqttMessage The MQTT message. - * @return The Message. + * @param topic the topic. + * @param mqttMessage the MQTT message. + * @return the Message. */ Message toMessage(String topic, MqttMessage mqttMessage); + /** + * Convert to a message builder. + * @param topic the topic. + * @param mqttMessage the MQTT message. + * @return the builder. + */ + AbstractIntegrationMessageBuilder toMessageBuilder(String topic, MqttMessage mqttMessage); + static MessageProcessor defaultQosProcessor() { return message -> message.getHeaders().get(MqttHeaders.QOS, Integer.class); } diff --git a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt.xsd b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt.xsd index 06b6a57473..2a31b8c9f0 100644 --- a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt.xsd +++ b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt.xsd @@ -89,6 +89,14 @@ + + + + + + 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 a9077e742f..4226bac5d2 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 @@ -74,6 +74,7 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.MessageProcessor; @@ -85,6 +86,7 @@ import org.springframework.integration.mqtt.event.MqttSubscribedEvent; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; +import org.springframework.integration.mqtt.support.MqttHeaderAccessor; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; @@ -264,6 +266,7 @@ public class MqttAdapterTests { MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory, "baz", "fix"); + adapter.setManualAcks(true); QueueChannel outputChannel = new QueueChannel(); adapter.setOutputChannel(outputChannel); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); @@ -290,6 +293,11 @@ public class MqttAdapterTests { assertThat(outMessage).isNotNull(); assertThat(outMessage.getPayload()).isEqualTo("qux"); + StaticMessageHeaderAccessor.getAcknowledgment(outMessage).acknowledge(); + verify(client).setManualAcks(true); + verify(client) + .messageArrivedComplete(MqttHeaderAccessor.id(outMessage), MqttHeaderAccessor.receivedQos(outMessage)); + MqttIntegrationEvent event = events.poll(10, TimeUnit.SECONDS); assertThat(event).isInstanceOf(MqttSubscribedEvent.class); assertThat(((MqttSubscribedEvent) event).getMessage()).isEqualTo("Connected and subscribed to [baz, fix]"); 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 85885a705a..d8da6b510f 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 @@ -12,6 +12,7 @@ >. +| If an inbound endpoint supports it, a call back to accept, reject, or requeue a message. +See <<./polling-consumer.adoc#deferred-acks-message-source,Deferred Acknowledgment Pollable Message Source>> and <<./mqtt.adoc#mqtt-ack-mode,MQTT Manual Acks>>. |=== Convenient typed getters for some of these headers are provided on the `IntegrationMessageHeaderAccessor` class, as the following example shows: diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index 591967512a..cb46dfa1a8 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/src/reference/asciidoc/mqtt.adoc @@ -74,6 +74,7 @@ The following listing shows the available attributes: send-timeout="123" <7> error-channel="errors" <8> recovery-interval="10000" <9> + manual-acks="false" <10> channel="out" /> ---- @@ -97,6 +98,7 @@ The payload is a `MessagingException` that contains the failed message and cause <9> The recovery interval. It controls the interval at which the adapter attempts to reconnect after a failure. It defaults to `10000ms` (ten seconds). +<10> The acknowledgment mode; set to true for manual acknowledgment. ==== NOTE: Starting with version 4.1, you can omit the URL. @@ -145,6 +147,23 @@ A new application context reverts to the configured settings. Changing the topics while the adapter is stopped (or disconnected from the broker) takes effect the next time a connection is established. +[[mqtt-ack-mode]] +==== Manual Acks + +Starting with version 5.3, you can set the `manualAcks` property to true. +Often used to asynchronously acknowledge delivery. +When set to `true`, header (`IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK`) is added to the message with the value being a `SimpleAcknowledgment`. +You must invoke the `acknowledge()` method to complete the delivery. +See the Javadocs for `IMqppClient` `setManualAcks()` and `messageArrivedComplete()` for more information. +For convenience a header accessor is provided: + +==== +[source, java] +---- +StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge(); +---- +==== + ==== Configuring with Java Configuration The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration: @@ -278,7 +297,7 @@ The default is `headers['mqtt_topic']`. <11> When `true`, the caller does not block. Rather, it waits for delivery confirmation when a message is sent. The default is `false` (the send blocks until delivery is confirmed). -<12> When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted (See <>). +<12> When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted (See <>). It contains 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. It contains the the `messageId`, the `clientId`, and the `clientInstance`, enabling delivery to be correlated with the send. @@ -373,7 +392,7 @@ public class MqttJavaApplication { ---- ==== -[[events]] +[[mqtt-events]] === Events Certain application events are published by the adapters. @@ -381,5 +400,6 @@ Certain application events are published by the adapters. * `MqttConnectionFailedEvent` - published by both adapters if we fail to connect or a connection is subsequently lost. * `MqttMessageSentEvent` - published by the outbound adapter when a message has been sent, if running in asynchronous mode. * `MqttMessageDeliveredEvent` - published by the outbound adapter when the client indicates that a message has been delivered, if running in asynchronous mode. +* `MqttSubscribedEvent` - published by the inbound adapter after subscribing to the topics. These events can be received by an `ApplicationListener` or with an `@EventListener` method. diff --git a/src/reference/asciidoc/polling-consumer.adoc b/src/reference/asciidoc/polling-consumer.adoc index 04828a4a1a..d1160d837a 100644 --- a/src/reference/asciidoc/polling-consumer.adoc +++ b/src/reference/asciidoc/polling-consumer.adoc @@ -60,7 +60,7 @@ Starting with version 5.0.1, certain modules provide `MessageSource` implementat This is currently limited to the `AmqpMessageSource` and the `KafkaMessageSource` provided by the `spring-integration-kafka` https://github.com/spring-projects/spring-integration-kafka[extension project]. With these message sources, the `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header (see <<./message.adoc#message-header-accessor,`MessageHeaderAccessor` API>>) is added to the message. -The value of the header is an instance of `AcknowledgmentCallback`, as the following example shows: +When used with pollable message sources, the value of the header is an instance of `AcknowledgmentCallback`, as the following example shows: [source, java] ---- diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 66c2d9ca1c..409a671f14 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -116,3 +116,13 @@ See <<./rsocket.adoc#rsocket-inbound,RSocket Inbound Gateway>> for more informat A `LeaderInitiatorFactoryBean` (as well as its XML ``) exposes a `candidate` option for more control over a `Candidate` configuration. See <<./zookeeper.adoc#zk-leadership,Leadership event handling>> for more information. + +[[x5.3-mqtt]] +=== MQTT Changes + +The inbound channel adapter can now be configured to provide user control over when a message is acknowledged as being delivered. +See <<./mqtt.adoc#mqtt-ack-mode,Manual Acks>> for more information. + +The outbound adapter now publishes a `MqttConnectionFailedEvent` when a connection can't be created, or is lost. +Previously, only the inbound adapter did so. +See <<./mqtt.adoc#mqtt-events,MQTT Events>>.