Fixes: #9597
Issue link: https://github.com/spring-projects/spring-integration/issues/9597
When input channel for the endpoint with a `Mqttv5PahoMessageHandler` is an `ExecutorChannel` and a lot of concurrent messages are sent,
the `Connect already in progress (32110)` exception can be thrown
* Wrap `this.mqttClient.connect()` in the `Mqttv5PahoMessageHandler.publish()` into a `Lock`
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
Fixes: #9428
Issue link: https://github.com/spring-projects/spring-integration/issues/9428
When `ClientManager` is used for MQTT channel adapters, a `MqttMessageDeliveredEvent`
is not emitted since callback for the `ClientManager` is not aware about `deliveryComplete`
* Use a `MqttActionListener` abstraction for the `publish` operation instead of a `deliveryComplete`
from a common callback
* Make some other refactoring into the `MqttPahoMessageHandler` and `Mqttv5PahoMessageHandler`
extracting a common logic into their `AbstractMqttMessageHandler` superclass
* Introduce an `MqttMessageNotDeliveredEvent` to be emitted from the `MqttActionListener.onFailure()` callback
* Adapt mocks in the `MqttAdapterTests` for a new code flow
* Add delivery events verification into the `ClientManagerBackToBackTests`
* Fix race condition in the `ClientManagerBackToBackTests`
Looks like the message can be consumed even before we just emit that `MqttMessageSentEvent`
Fixes: #9276
The current Eclipse Paho client has wrong removal from map logic which leads to the
`ConcurrentModificationException` when we unsubscribe from topic.
* Catch a `ConcurrentModificationException` `this.mqttClient.unsubscribe()` and just log it as an `error`
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
Fixes: #9214
The `MqttAsyncClient` can set `subscriptionIdentifier` from session if it is enabled
and available from connection
* Remove manual `subscriptionIdentifierCounter` from the `Mqttv5PahoMessageDrivenChannelAdapter`.
Instead, use `subscriptionProperties.setSubscriptionIdentifiers(List.of(0));`
to make the `this.mqttSession.getNextSubscriptionIdentifier();` to work
when `if (connOpts.useSubscriptionIdentifiers() && this.mqttConnection.isSubscriptionIdentifiersAvailable()) {` is `true`
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
Fixes: #9112
`Mqttv5ClientManager` hangs in `stop()` if never was connected.
The scheduled reconnect timer in the client is never cancelled.
* Call `stopReconnectCycle()` on the client via reflection when we disconnect
from the client in Spring Integration MQTT components
**Auto-cherry-pick to `6.2.x` & `6.1.x`**
Fixes: #8879
To workaround the problem with `$share/` subscriptions the `Mqttv5PahoMessageDrivenChannelAdapter` must provide a `subscriptionIdentifier` into `MqttProperties` on `subscribe()`
* Introduce a `Mqttv5PahoMessageDrivenChannelAdapter.subscriptionIdentifierCounter` according to the MQTT specification:
> 3.8.2.1.2 Subscription Identifier: [..]The Subscription Identifier is associated with any subscription created or modified as the result of this SUBSCRIBE packet. If there is a Subscription Identifier, it is stored with the subscription.
This one is associated with the MQTT session for the current subscriber and does not interfere into other sessions even if identifier is same from the counter.
It works because the Subscription identifier is per session and because you cannot have multiple connection with the same client ID.
**Cherry-pick to `6.2.x` & `6.1.x`**
Fixes: #8873
The `mqttClient.subscribe()` API does not check if properties are provided and fails on the
`subscriptionProperties.getSubscriptionIdentifiers().get(0)` call with an `IndexOutOfBoundsException`
* Use another `mqttClient.subscribe()` API in the `Mqttv5PahoMessageDrivenChannelAdapter` where there is not such a check
* Ensure that `addTopic(NAME)` works as expected in the `Mqttv5BackToBackTests`
**Cherry-pick to `6.2.x` & `6.1.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/8654
Spring Framework has an ability to start dependant beans automatically
when we start the current one.
The `AbstractMethodAnnotationPostProcessor` is missing a bean dependency
registration causing errors in target applications when Messaging Annotations configuration is used.
* Add `registerDependentBean()` into an `AbstractMethodAnnotationPostProcessor` when we generate
and register a `ConsumerEndpointFactoryBean`
* Change one of the `ClientManagerBackToBackTests` configuration to rely on a `@ServiceActivator`
for `Mqttv5PahoMessageHandler` bean to ensure that change in the `AbstractMethodAnnotationPostProcessor`
has a proper effect
**Cherry-pick to `6.1.x` & `6.0.x`**
We cannot always reset `readyToSubscribeOnStart` to `false` in the `stop()`
since we may be connected to `ClientManager` and have `cleanStart` option,
so we `unsubscribe()` in the stop and then when we start there is no way to subscribe back
* Reset `readyToSubscribeOnStart` in the `AbstractMqttMessageDrivenChannelAdapter.doStop()`
only if we don't unsubscribe for a `cleanSession` reason.
* Also reset it in the `connectionLost`, so next `connectComplete` will take care about subscription
or `readyToSubscribeOnStart` state change
* Use `isActive()` instead of `isRunning()` in the `connectComplete()` since there is a race
condition when connection is established but `doStart()` has not returned yet, but already passed
`if (this.readyToSubscribeOnStart)` line
Fixes https://github.com/spring-projects/spring-integration/issues/4014
The `ClientManager` implementations uses by mistake a `connectionTimeout` for operations with completion wait
* Introduce `completionTimeout` and `disconnectCompletionTimeout` for `ClientManager`
abstraction to realign the logic with existing channel adapters and Paho Client by itself.
* Deprecate `DEFAULT_COMPLETION_TIMEOUT` and `DISCONNECT_COMPLETION_TIMEOUT` constants
in the `AbstractMqttMessageDrivenChannelAdapter` in favor of respective replacement in the `ClientManager`
* Pull `disconnectCompletionTimeout` property from the `MqttPahoMessageDrivenChannelAdapter` to its superclass
* Use new `disconnectCompletionTimeout` in the `Mqttv5PahoMessageDrivenChannelAdapter` for similar `disconnectForcibly()` call
* Fix Lifecycle race condition when `ClientManager` is started by the outbound channel adapter
(`Integer.MIN_VALUE` phase and auto-startup - see `DefaultLifecycleProcessor.doStart()` and logic around `dependencies`)
which is much earlier than `MessageProducerSupport` (`Integer.MAX_VALUE / 2` phase) and there a `connectComplete()` callback
might be called before the `MessageProducerSupport.start()`.
For that purpose check for an `isRunning()` in the `connectComplete()` before subscribing and set `readyToSubscribeOnStart` flag
to `subscribe()` in a `doStart()` of this `MqttMessageDrivenChannelAdapter`
* Remove redundant `MqttPahoMessageDrivenChannelAdapter.cleanSession` property in favor of
`this.clientFactory.getConnectionOptions().isCleanSession()` call
GH-8550: MQTT: Always re-subscribe on re-connect
Fixes https://github.com/spring-projects/spring-integration/issues/8550
Turns out the Paho MQTT client does not re-subscribe when connection
re-established on automatic reconnection
* Fix `AbstractMqttMessageDrivenChannelAdapter` to always subscribe to their topics
in the `connectComplete()` independently of the `reconnect` status
* Verify behavior with `MOSQUITTO_CONTAINER` image restart in Docker
Fixes https://github.com/spring-projects/spring-integration/issues/4002
Turns out the `ConsumerStopAction` was introduced in a point version to mitigate an unsubscription bug
and let to preserve a previous behaviour.
* Deprecate `ConsumerStopAction` in favor of just `cleanSession` flag in the `MqttConnectOptions`
**Cherry-pick to `5.5.x`**
* GH-3959: MqttConFailedEvent for normal disconnect
Fixes https://github.com/spring-projects/spring-integration/issues/3959
The `MqttCallback.disconnected(MqttDisconnectResponse)` in Paho v5 client is also
called when server initiates a disconnection.
In this case that `MqttDisconnectResponse` does not have a `cause` value
* Modify `MqttConnectionFailedEvent` to make a `cause` property optional
* Fix `Mqttv5PahoMessageDrivenChannelAdapter` & `Mqttv5PahoMessageHandler`
to not check for `cause`, but emit an `MqttConnectionFailedEvent` for any `disconnected()` calls
Unfortunately current Paho v3 client does not call `connectionLost()` for normal disconnections
and we cannot react for this callback with an `MqttConnectionFailedEvent`
**Cherry-pick to `5.5.x`**
* Fix language in doc
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
Fixes https://github.com/spring-projects/spring-integration/issues/3955
`Mqttv5PahoMessageDrivenChannelAdapter` unsubscribes from all topics, even if `cleanStart/cleanSession` is set to `false`, thus not receiving offline messages after restart.
* unsubscribe `Mqttv5PahoMessageDrivenChannelAdapter` only when `cleanStart`
* add tests
**Cherry-pick to `5.5.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/3686
* Add `src/idea` with respective editor config for IntelliJ IDEA.
Must be imported into an IDE
* Add `src/eclipse` with respective editor config for Eclipse/STS
* Reformat imports in source code according a new editor config
* Some code style improvement for SMB classes
* Make an `SmbSessionTests` based on the `SmbTestSupport` for faster execution, but not blocking on fake URL connection attempt
* Remove `AbstractMqttMessageDrivenChannelAdapter.Topic` model in favor of `LinkedHashMap` handling
Fixes https://github.com/spring-projects/spring-integration/issues/3685
Introduce some initial design.
Add a new interface `ClientManager` which will manage clients and
connections.
Use this manager in v3 topic adapter and message handler.
Add a new interface `ClientManager` which will manage clients and
connections.
Add different implementations for v3 and v5 MQTT clients.
Use this manager in v3/v5 topic adapters and message handlers.
Add a couple of unit/integration tests to cover client manager usage.
Several small code improvements after the code review:
* Improve client manager usage via providing several mutual exclusive
constructors, whether the users provides `url` or `connectionOptions`
or `clientFactory` for v3.
* Move the logger to `AbstractMqttClientManager`
* Do not inject TaskScheduler in constructor for v3 client manager
but use lazy init via `BeanFactory` and `IntegrationContextUtils`
* Other smaller code readability improvements
Add new tests with reconnect cases.
Other code improvements after the code review:
* Adjust javadocs according to standards
* Remove `setClientManager` and use exclusive ctors
* Make automatic reconnects using the v3 client instead of manually
using task scheduler
Some fixes and improvements after another code review iteration:
* Rearrange the code according to the code style guides
* Move client instance to `AbstractClientManager` with `isRunning`
method
* Fix abstract adapter/handler fields visibility and `final`ize them
where we can
* Send application event if automatic reconnect is not enabled for the
client manager
Other fixes and improvements after code review:
* Changes around fields, methods, ctors visibility
* Removed contradictory ctors
* Reduce amount of unnecessary `getClientManager() != null` checks
in logic and make it as similar as possible for client manager and the
old approach
* Use auto-reconnect where possible
* Remove manual reconnect trigger and rely on events instead to know
where to subscribe
* Do not close the connection in adapter to be able to use reconnect
logic without lose of subscriptions
* Make `ClientManager` extend `MqttComponent` so that it knows about
connection options as part of its contract
* Remove not relevant auto test cases (relying on connection close or
manual reconnect)
* Other code style smaller changes
Other fixes and improvements after code review:
* Get manual `reconnect` invocation back for v3/v5 adapters and client
managers (see bug GH-3822 for a reasoning)
* Remove unnecessary getters/setter for a listener and use adapter
class as listener instead
* Optimize MessageListener: remove redundant inner class and use a
single method reference instead of N instances per each subscribe
* Javadocs improvements
* Add Javadocs to abstract client manager
* Extract common callback add/rm logic to abstract adapter class
* Small code cleanups/fixes related to code style & simplicity, ctor
inits and unnecessary methods; eliminate unnecessary logs noise
* Remove `@LongRunningTest` for `ClientManagerBackToBackTests` as test
run time is ~6-7 secs
* Remove client factory as dependency for v3 client manager and use
plain connection properties and client persistence instead
* Add missed javadocs
* Other code style & cleanup improvements
* More code cleanup
* More Javadocs
* Mostly changes are related to the `TaskScheduler` and `Trigger` APIs
* Migrate to `micrometer-tracing` dependency
* Rework `SocketTestUtils` to use a `InetAddress.getLocalHost()`
for more stability and performance on Windows
* Fix docs for new `PeriodicTrigger` API
Fixes https://github.com/spring-projects/spring-integration/issues/3623
* `IntegrationFlow` refactoring
* Apply several code style improvements and good practices
* Code style: no empty lines for methods javadocs
* make deprecated implementation reuse actual one instead of the copy-paste approach
* add whats-new comments
* Fix whats-new page according to standards
Fixes: https://github.com/spring-projects/spring-integration/issues/3822
* Apply spring-framework code style on modified class
* Remove unwanted formatting
* Take pull request comments into account
* Code and JavaDocs clean up
* Improve `Mqttv5BackToBackAutomaticReconnectTests` removing non-related code
* Improve `mqtt.adoc` for this new manual reconnection feature
**Cherry-pick to `5.5.x`**
The `withReuse(true)` and `testcontainers.reuse.enable=true` don't work together with `@Container`.
The JUnit extension gathers those containers and stop them in the end of test class unconditionally.
* Remove `@Container` annotation usage
* Use `@BeforeAll` and `GenericContainer.start()` manually
This way the container ensures to reuse existing running container and don't start a fresh one.
Since the container instance is stored in a `static` property, it is really started only once.
The rest tests in a suite just reuse that existing container.
Ryuk container will take care about their stopping and removal eventually afte JVM exit.
**Cherry-pick to `5.5.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/3732
The `Mqttv5PahoMessageDrivenChannelAdapter` unconditionally tries to (un)subscribe
to/from topics when the `mqqtClient` might not be initialized yet.
* Add `mqqtClient` initialization check before adding or removing topic.
Re-align logic with the `MqttPahoMessageDrivenChannelAdapter`
**Cherry-pick to `5.5.x`**
* Upgrade to Java 17, SF-6.0, Gradle 7.2
* Upgrade to Jakarta dependencies and respective namespaces
* Fix some tests for Java 17 compatibility
* Fix wrong Javadocs
* Add some missed Javadocs
* Fix more `jakarta` namespace
* Fix WS & XML modules to use Jakarta EE
* `--add-opens` in some modules for their reflection-based tests
* Disable Kafka tests which does not work on Windows; see Apache Kafka `3.0.1`
* Upgrade to JUnit `5.8.1`
* Migrate JMS tests to Artemis
* Remove RMI module as it was deprecated before
* Fix `pr-build-workflow.yml` for Java 17
* Fix JavaDocs warnings using `Xdoclint:syntax` per module, not in the top-level `api` task
* Move docs for version `6.0`
Add a logic into the `TestUtils` to take a Docker registry URL
from the `DOCKER_REGISTRY_URL` ENV variable.
Use the value in the `GenericContainer` for image to pull.
Fallback to an official registry if no value
* Use `MosquittoContainerTest` for all the MQTT integration tests
* Remove MQTT `BrokerRunning` JUnit 4 rule
* Mark `BackToBackAdapterTests` with `@LongRunningTest` as it is really too slow
* GH-3432: Add MQTT v5 channel adapters
Fixes https://github.com/spring-projects/spring-integration/issues/3432
* Add `optional` dependency for `org.eclipse.paho:org.eclipse.paho.mqttv5.client`
* Add `MqttProtocolErrorEvent` and emit it from the `mqttErrorOccurred()` callback of the MQTT v5 client
* Add `MqttHeaderMapper` since MQTT v5 has introduced user properties pair to transfer over the protocol
* Add `Mqttv5PahoMessageHandler` as one more extension of the `AbstractMqttMessageHandler`
* Add more convenient `MqttHeaders` constants for easier headers mapping configuration
* Ensure via `Mqttv5BackToBackTests` that MQTT v5 is supported by the provided components
* Change `pr-build-workflow.yml` to use `eclipse-mosquitto` container for testing all the MQTT interactions
* Change `cyrilix/rabbitmq-mqtt` service to the `rabbitmq:management` since RabbitMQ does not support MQTT v5
* * Handle manual acks
* Add `Mqttv5PahoMessageDrivenChannelAdapter.persistence` property
* * Add documentation
* Add `MosquittoContainerTest` for TestContainers support with Mosquitto image
* Fix language in the docs after review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
Fixes https://github.com/spring-projects/spring-integration/issues/3627
The `destroy()`, and therefore `stop()` could be called from the `MqttConnectionFailedEvent` handling
in the same thread resetting the `client` property to `null`.
* Check for `this.client != null` in the next block of the `connectAndSubscribe()` to avoid NPE
* Check for `isActive()` in the `scheduleReconnect()` to be sure do not reconnect if channel adapter
has been stopped already
**Cherry-pick to `5.4.x`**
