* GH-3813: ClientWebSocketContainer URI setting
Fixes https://github.com/spring-projects/spring-integration/issues/3813
Introduce a `ClientWebSocketContainer(WebSocketClient client, URI uri)` ctor
to let end-user to decide what and how should be encoded the URI for WebSocket connection
* Fix language in docs
Co-authored-by: Gary Russell <grussell@vmware.com>
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
* Upgrade dependencies whenever it is reasonable
* Fix deprecations and some new API paths
* Move docs to a new version with respective branching for the `changes-5.5-6.0.adoc`
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
For better performance by default it is better to not pull
a `io.micrometer:context-propagation` a hard dependency.
* Remove `io.micrometer:context-propagation` dependency management
* It is pulled transitively by the `io.micrometer:micrometer-tracing-integration-test` in test scope
* Rework all the `ContextSnapshot` usage in the reactive code to respective recommended `handle()` API in `Flux` and `Mono`
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/4008
Impossible to retrieve current leader id from ZK `LeaderInitiator`
* change return type of `LeaderInitiator.getContext()` from `Context` to `CuratorContext`
* add `getLeader()` method to `CuratorContext`
* add `getParticipants()` method to `CuratorContext`
* GH-4001: Doc for cooperation with some HZ objects
Fixes https://github.com/spring-projects/spring-integration/issues/4001
The `IQueue`, `ITopic` and `IExecutorService` can be used with Spring Integration
channel as is without any extra component implementations.
* Document the cooperation feature with Hazelcast objects via samples
* * Add a sample about an Inbound Channel Adapter on the `IQueue`
* Fix language in docs
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/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`**
To propagate an `Observation` from reactive stream (e.g. WebFlux)
we have to capture its context and set it into the current thread scope.
* Add a `io.micrometer:context-propagation` dependency to support reactive context propagation
* Populate a `parentObservation` in the `IntegrationObservation.PRODUCER.observation()`
since it is not available for tracing on `Observation.onStart()`.
Might be tentative until upcoming fix in Micrometer Observation
* Populate from reactive context in the `WebFluxInboundEndpoint` where we use just `send()` operation downstream
* Populate from reactive context in the `MessagingGatewaySupport` where we use `send()` operation downstream or `FluxMessageChannel.subscribeTo()`
* Use `contextCapture()` in the `FluxMessageChannel` to gather a `ThreadLocal` info into a Reactor context
and then set back to `ThreadLocal` in the `transformDeferredContextual()` which really happens on a different thread
* Verify a trace propagation from WebFlux to an integration flow via Brave instrumentation in the `WebFluxObservationPropagationTests`
* GH-3998: Fix gateway docs for `@Payload`
Fixes https://github.com/spring-projects/spring-integration/issues/3998
The `GatewayProxyFactoryBean` now deals only with a `MethodArgsHolder` as a root evaluation context object.
There is no `#args` and `#method` SpEL variables anymore.
However, the `gateway.adoc` still refer to those.
* Fix `gateway.adoc` for a proper expressions and actual wording
* * Cross link to `gateway-expressions` paragraph for better context
Fixes https://github.com/spring-projects/spring-integration/issues/3994
Essentially, migrate those hints from Spring Boot Actuator:
an `IntegrationGraphServer` can be used without Spring Boot
* Introduce `IntegrationGraphRuntimeHints` - implementation of `RuntimeHintsRegistrar`
to register reflection hints for `Graph` and top-level `IntegrationNode` types.
* Use `@ImportRuntimeHints` on the `IntegrationGraphServer` to make those hints
conditional.
* GH-3993: Fix async race condition in TcpOutGateway
Fixes https://github.com/spring-projects/spring-integration/issues/3993
When `TcpOutboundGateway` is in an `async` mode and `CCF` is configured
not for `singleUse` an `Semaphore` around an obtained `TcpConnection` is involved.
If we fail on `TcpConnection.send()`, resources are not clean up, including the
mentioned `Semaphore`: in async mode this happens only when we receive a reply.
* Catch an exception on the `TcpConnection.send()` and perform `cleanUp()` in async mode.
* Add `cleanUp()` into a scheduled task from the `TcpOutboundGateway.AsyncReply`
when no reply arrives in time.
* Optimize `TcpOutboundGateway.AsyncReply` behavior to cancel no-reply scheduled task
when reply arrives into a `CompletableFuture`
**Cherry-pick to `5.5.x`**
* * Call `cleanUp()` from no response scheduled task
only if `future.completeExceptionally()` is `true`
The `ChannelInterceptor.postSend()` may be called when
`MessageChannel.receive()` has already done its job polling from
the queue where message appears in the `MessageChannel.doSend()`
* Use `ChannelInterceptor.preSend()` contract instead to obtaine info about the current thread
Related to https://stackoverflow.com/questions/75109345/spring-integration-webflux-inboundgateway-replychannel-for-error-response
The `MessagingGatewaySupport.doSendAndReceiveMessageReactive()` uses `MutableMessageBuilder`
to build a new message for next `send` operation.
Even if this is an error flow, the provided `ErrorMessage` becomes a plain `MutableMessage`.
This may break some downstream logics, like `TracingChannelInterceptor` from Spring Cloud Sleuth,
which checks for the `Message` class to rebuild or reuse a message content.
Therefore, an error handling flow is not able to extract error info because it is just lost.
* Fix `MessagingGatewaySupport.doSendAndReceiveMessageReactive()` to check for message type before
choosing an `AbstractIntegrationMessageBuilder` impl for building a new message.
The regular `MessageBuilder` just builds a new `ErrorMessage` for an exception payload.
* Add `filter()` into a `WebFluxDslTests` error handling flow to be sure that message for
error sub-flow is really an `ErrorMessage`
Fixes https://github.com/spring-projects/spring-integration/issues/3990
Turns out just being on the classpath doesn’t make the type reachable.
If it’s only accessed reflectively then it’s not reachable.
This is exaclt what happened with our `JsonPathUtils` which is used via
reflection from SpEL when that calls its method via function reference
* Change `onReachableType()` logic for `com.jayway.jsonpath.JsonPath` type
to `ClassUtils.isPresent()` on `JsonPathUtils` reflection hint registration.
Currently, when `!resolutionRequired` not resolved channel name
is ignored silently, and we go ahead to `defaultOutputChannel` or
exception thrown
* For better end-user experience `DEBUG` not resolved and ignored channel name if `!resolutionRequired`
* Revise JS support for regular JVM
Turns out that we just need to have `org.graalvm.sdk:graal-sdk` and `org.graalvm.js:js`
dependencies for regular JVM to enable JavaScript support for Spring Integration Scripting module
* Add respective `providedImplementation` dependencies
* Remove `@EnabledIfSystemProperty` from unit tests
* Mention those dependencies in the doc
* Rework sentence in the doc for better English
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/3986
The `ReactiveStreamsConsumer` in addition to plain `Subscriber` and `MessageHandler`
also supports a `ReactiveMessageHandler`, but `MockIntegrationContext` doesn't
handle a scenario when `ReactiveMessageHandler` is provided for consumer
* Rework the logic in the `MockIntegrationContext` to substitute a `ReactiveMessageHandler`
in the `ReactiveStreamsConsumer` which has a precedence in its logic over plain `Subscriber`.
* Wrap a plain `MessageHandler` mock into a `ReactiveMessageHandler` before substitution
in the `ReactiveStreamsConsumer`
* Reset `ReactiveStreamsConsumer.reactiveMessageHandler` with source `ReactiveMessageHandler`
or `null` respectively to an original `ReactiveStreamsConsumer` configuration
* Mention `ReactiveMessageHandler` use-case in the `testing.adoc`
**Cherry-pick to `5.5.x`**
* Do not build exception instances in advance
When no `cause` and no clean context of the exception thrown,
the code place where an exception has been created is confusing.
* Fix `SimpleMessageStore` to create an `out of capacity` exception in the
`addMessagesToGroup()` exactly at the point where it is thrown
* Fix `FunctionExpression` and `SupplierExpression` do not pre-create `readOnlyException`:
it is unlikely these kind of expressions are going to be used in the SpEL `write` context,
so we save some time and memory not creating extra object and the plain call `throw new`
gives us a clean context what method call has ended up with such an exception
* * Extract `outOfCapacityException` to the method in the `SimpleMessageStore`
to avoid duplication.
Even if we got in the end extra line in th stack trace, it is still clear from
where it is thrown:
```
org.springframework.messaging.MessagingException: SimpleMessageStore was out of capacity (1) for group 'foo', try constructing it with a larger number.
at org.springframework.integration.store.SimpleMessageStore.outOfCapacityException(SimpleMessageStore.java:324)
at org.springframework.integration.store.SimpleMessageStore.addMessagesToGroup(SimpleMessageStore.java:302)
at org.springframework.integration.store.AbstractMessageGroupStore.addMessageToGroup(AbstractMessageGroupStore.java:189)
at org.springframework.integration.store.SimpleMessageStoreTests.shouldNotHoldMoreThanGroupCapacity(SimpleMessageStoreTests.java:128)
```
Fixes https://github.com/spring-projects/spring-integration/issues/3980
When not `isSharedSession`, the `initClient()` is called for every session
we request from the factory and in concurrent calls we end up with not initialized SSH client
in some threads.
* Add `synchronized` double check logic to the `initClient()` to block other threads while the first one
initialize the client
* Use `volatile boolean` instead of `AtomicBoolean` and change its state when `SshClient` is created and started
* Fix `pr-build-workflow.yml` removing unused services: now they are pulled by Testcontainers
* Use official `rabbitmq:management` image instead of `MQTT`-based.
The `MQTT` is now tested with Testcontainers and Mosquitto broker image
Looks like sometime Spring Framework XML parser can properly determine the target factory method to chose,
but sometimes it does that in a wrong order.
* Add `type="java.lang.Class"` to the ctor args everywhere for `class="org.mockito.Mockito" factory-method="mock"`
`<bean>` definitions
