* Use `StandardCharsets.UTF_8` as a charset configuration instead of string.
Later remove the necessity of handling encoder errors.
* Use regular expressions to validate the `UUID` string standard representation.
Later obsolete the need for try/catch exceptions.
* Deprecate `UUIDConverter.DEFAULT_CHARSET` since it is out of use.
**Cherry-pick to `6.1.x` & `6.0.x`**
Code clean up for `UUIDConverter`
We use an aggregate handler that uses a group timeout.
It's expected that this timeout will be hit frequently.
However, this causes a info log message emitted every time.
I'd propose to log this message on debug.
**Cherry-pick to `6.1.x` & `6.0.x`**
(cherry picked from commit f89ca99bde)
Fixesspring-projects/spring-integration#8754
* `AbstractIntegrationMessageBuilder.setExpirationDate(@Nullable Long expirationDate)`
* `MessageBuilder.setExpirationDate(@Nullable Long expirationDate)`
* `MessageBuilder.setExpirationDate(@Nullable Date expirationDate)`
**Cherry-pick to `6.1.x` & `6.0.x`**
(cherry picked from commit 75fcddac00)
* Fix maxMessagesPerPoll for SourcePollingChAdapter
The `AbstractMethodAnnotationPostProcessor` does not check
for `PollerMetadata.MAX_MESSAGES_UNBOUNDED` before setting
`maxMessagesPerPoll` into a `SourcePollingChannelAdapter`
which in this case must be `1`
Also fix `SourcePollingChannelAdapterFactoryBean` to not mutate
the provided `PollerMetadata` (which might be global default)
with a new `maxMessagesPerPoll`
**Cherry-pick to `6.1.x` & `6.0.x`**
* * Fix `this.` prefix in `SourcePollingChannelAdapterFactoryBean`
The `JmsMessageDrivenEndpoint` delegates all the hard work to the
`ChannelPublishingJmsMessageListener`, but missed to propagate an `ObservationRegistry`
and other related options.
The `JmsInboundGateway` is worse: it delegated to the `JmsMessageDrivenEndpoint`
* Add `IntegrationObservation.HANDLER` observation to the `MessagingGatewaySupport.send()`
operation: used by the delegate in the `ChannelPublishingJmsMessageListener`
* Expose and propagate observation-related options from `JmsInboundGateway`
and `JmsMessageDrivenEndpoint`
* Expose `observationConvention()` option on the `MessagingGatewaySpec`
and `MessageProducerSpec`
* Remove unused imports
* Do not start a new `RECEIVER` observation if there is already `SERVER` one
* Fix `MessagingGatewaySupport` for `Observation.NOOP` check.
The parent process may still use `ObservationRegistry.NOOP` which sets
`Observation.NOOP` instance into the current context and thread local.
* fix merge conflicts
**Cherry-pick to `6.1.x` & `6.0.x`**
* Fix ThreadSPropagationChInterceptor for stacking
Related SO thread: https://stackoverflow.com/questions/77058188/multiple-threadstatepropagationchannelinterceptors-not-possible
The current `ThreadStatePropagationChannelInterceptor` logic is to wrap one
message to another (`MessageWithThreadState`), essentially stacking contexts.
The `postReceive()` logic is to unwrap a `MessageWithThreadState`,
therefore we deal with the latest pushed context which leads to the `ClassCastException`
* Rework `ThreadStatePropagationChannelInterceptor` logic to reuse existing `MessageWithThreadState`
and add the current context to its `stateQueue`.
Therefore, the `postReceive()` will `poll()` the oldest context which is, essentially,
the one populated by this interceptor before, according to the interceptors order
* Fix `AbstractMessageChannel.setInterceptors()` to not modify provided list of interceptors
* The new `ThreadStatePropagationChannelInterceptorTests` demonstrates the problem
described in that mentioned SO question and verifies that context are propagated
in the order they have been populated
**Cherry-pick to `6.1.x` & `6.0.x`**
* * Fix `ThreadStatePropagationChannelInterceptor` for publish-subscribe scenario.
Essentially, copy the state queue to a new decorated message
* Fix `BroadcastingDispatcher` to always decorate message, even if not `applySequence`
* * Fix unused import in the `BroadcastingDispatcher`
* * Fix unused import in the `ThreadStatePropagationChannelInterceptor`
Fixes https://github.com/spring-projects/spring-integration/issues/8697
The `IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME` is not designed
for end-user code autowire, rather internal framework requirement
* Mark `BeanDefinition` for `IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME`
as `setAutowireCandidate(false)`
(cherry picked from commit f9dc75c739)
* GH-8685: Re-fetch group after setting condition
Fixes https://github.com/spring-projects/spring-integration/issues/8685
The `AbstractCorrelatingMessageHandler` updates the group metadata in DB
not only for provided `condition`, but also a `lastModified` field.
A subsequent scheduling for group timeout takes the `lastModified`
to compare with the value in the store after re-fetching group in task.
This does not reflect reality since adding `condition` modifies the data in DB,
but in-memory state remains the same.
* Re-fetch a group from the store in the `AbstractCorrelatingMessageHandler.setGroupConditionIfAny()`.
* Verify expected behavior via new `ConfigurableMongoDbMessageGroupStoreTests.groupIsForceReleaseAfterTimeoutWhenGroupConditionIsSet()`
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
* * Fix Checkstyle violation in the test
The `ConsumerEndpointSpec` extensions for Kotlin
don't delegate to the provided `endpointFactoryBean`
* Introduce `KotlinConsumerEndpointSpec` extension for `ConsumerEndpointSpec`
with the proper delegation to the provided spec
* Use `KotlinConsumerEndpointSpec` in the Kotlin-specific `Spec` classes
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.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`**
The `NumberFormatException` flow control is costly
operation
* Use `Character.isDigit()` check iterating through property String
instead of `NumberFormatException` flow control
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
The `FluxMessageChannel` can subscribe to any volatile `Publisher`.
For example, we can call Reactor Kafka `Sender.send()` for
input data and pass its result to the `FluxMessageChannel`
for on demand subscription.
These publishers are subscribed in the `FluxMessageChannel`
and their `Disposable` is stored in the internal `Disposable.Composite`
which currently only cleared on `destroy()`
* Extract `Disposable` from those internal `subscribe()` calls
into an `AtomicReference`.
* Use this `AtomicReference` in the `doOnTerminate()`
to remove from the `Disposable.Composite` and `dispose()`
when such a volatile `Publisher` is completed
**Cherry-pick to `6.0.x` & `5.5.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/8613
If `JsonPropertyAccessor` is registered for SpEL, it would be great
to have it working in native images as well.
Since SpEL is fully based on reflection, expose
`JsonPropertyAccessor$ComparableJsonNode` and `JsonPropertyAccessor$ArrayNodeAsList`
reflection hints for their method invocations from SpEL
**Cherry-pick to `6.0.x`**
* Propagate Reactor context over headers
When we do something like `Flux.from(Publisher)`
and don't compose it with the one involved in the `Subscriber` context,
we lose this context.
* Provide a mechanism to propagate a Reactor context over message header
produce within that context.
* Restore this context in the `FluxMessageChannel` for a new publisher
we use in this channel
**Cherry-pick to `6.0.x`**
* Fix language in docs
Co-authored-by: Gary Russell <grussell@vmware.com>
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
* Some `LockRegistryLeaderInitiator` improvements
It is better to not go to the target lock provider at all
if the current thread is already interrupted.
* Check for the `Thread.currentThread().isInterrupted()` in the `while` loop
and `restartSelectorBecauseOfError()` immediately without checking for a lock
* Fix some other simple typos in the `LockRegistryLeaderInitiator`
**Cherry-pick to `6.0.x`**
* * Introduce a `LeaderSelector.yielding` flag
to revoke leader smoothly.
Turns out just canceling the `Future` may lead to a broken lock
where we cannot unlock it because the target lock repository may not work
with interrupted threads.
This way a new leader must wait until the lock is expired in the store
* Remove the use of instance suppliers on bean definitions
that are processed during the AOT phase.
* The remaining areas that use instance suppliers do so
at runtime and do not use reflection, but instead
are passed the configured bean to register.
See https://github.com/spring-cloud/spring-cloud-stream/issues/2655
**Cherry-pick to `6.0.x`**
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`
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`
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.
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`
* 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)
```
* Obtain observed timers in the Integration Graph
The `Observation` populates slightly different Micrometer timer names and their tags
* Fix `MicrometerNodeEnhancer` to search timers in the registry according to the `Observation`
convention if the provided component is observed
* For that purpose expose an `IntegrationManagement.isObserved()` option to check
of component is instrumented with an `Observation`
* Improve the logic in the `IntegrationGraphServer` to delegate into a `Function`
to call `micrometerEnhancer.enhance()` instead of `static` property which might not
be OK in the environment where several integration applications are ran in the same JVM
* * Collect times into graph for inbound gateways & producers
* Fix language in javadoc of new `IntegrationManagement.isObserved()`
* Add `@Nullable` to `MicrometerNodeEnhancer.buildTimerStats()` param to satisfy this method contract
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
The time-based tests are very sensitive to an activity on the target machine.
So, strict test sometimes fail.
* Fix `DelayerUsageTests` to deal with approximate values to compare by some percentage
Fixes https://github.com/spring-projects/spring-integration/issues/3953
The replying `MessageHandler` can resolve the target output channel in on-demand manner.
A new messaging annotations on `@Bean` parsing algorithm is missing the lazy-load
opportunity and uses a `RuntimeBeanReference` for output channel options
loading its bean eagerly.
* Fix `AbstractMethodAnnotationPostProcessor` to use a `outputChannelName` and
`defaultOutputChannelName` target properties for channel names to set.
* Expose more reflection hints
* Bring back `@Reflective` on `Pausable` - for possible end-user usage
* Add `ReactiveMessageHandler` hint since its method is used reflectively
in the `IntegrationRSocketMessageHandler`
* Add `@Reflective` on the `ServerRSocketMessageHandler.handleConnectionSetup()`
since it is used reflectively for a `registerHandlerMethod()`
* Add `KafkaRuntimeHints` to expose `Pausable` contract on Kafka inbound endpoints
for SpEL invocation via Control Bus
* Document native images support
* Fix language in docs
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* Also register `Pausable` explicitly and remove its `@Reflective`.
This type is needed for Control Bus SpEL execution, but it might not
be available at runtime because not all endpoints implement it
The `Pausable` & `ManageableSmartLifecycle` types are eligible to be used
from Control Bus.
This one is based on SpEL and therefore requires a reflection for those
types to be available in native image
* GH-3555: Change logger order for errorChannel
Fixes https://github.com/spring-projects/spring-integration/issues/3555
The default global `errorChannel` has a `LoggingHandler` as a subscriber.
It is subscribed without any `order` which may lose logging messages,
when another subscriber with re-throw is present.
* Set default `LoggingHandler` on the default `errorChannel` to `Ordered.LOWEST_PRECEDENCE - 100`
to give a room for custom subscribers without an `order` and still get error logged
* Add extra note in docs about an order for custom subcribers
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* GH-3946: Revise Router channelKeyFallback option
Fixes https://github.com/spring-projects/spring-integration/issues/3946
The `AbstractMappingMessageRouter` has both `resolutionRequired` and `channelKeyFallback`
as `true` by default.
End-users expects them to back off when they set a `defaultOutputChannel`.
They really want something similar to Java `switch` statement
* Change the logic in the `AbstractMappingMessageRouter` to reset `channelKeyFallback`
to `false` when `defaultOutputChannel` to avoid attempts to resolve channel from name,
but rather fallback to `defaultOutputChannel` as it states from th mentioned
Java `switch` statement experience
* Deprecate `RouterSpec.noChannelKeyFallback()` in favor of newly introduced `channelKeyFallback(boolean)`
* Call `channelKeyFallback(false)` from an overloaded `defaultOutputToParentFlow()`
to reflect the mentioned expected behavior in Java DSL as well.
* Respectively, deprecate `KotlinRouterSpec.noChannelKeyFallback()` wrapper
in favor of newly introduced `channelKeyFallback(channelKeyFallback: Boolean)`
* Remove redundant already `noChannelKeyFallback()` option in the `NoFallbackAllowedTests`
* Document the change and new behavior
* Fix `IntegrationGraphServerTests` to `setChannelKeyFallback(true)` explicitly
* Remove not relevant `default-output-channel` from the `DynamicRouterTests-context.xml`
* Reject an `AbstractMappingMessageRouter` configuration where `defaultOutputChannel` is provided
and both `channelKeyFallback` & `resolutionRequired` are set to `true`.
Such a state makes `defaultOutputChannel` as not reachable and may cause some confusions in target
applications.
* Remove `&` symbol from JavaDocs
* Fix `boolean` expression for `AbstractMappingMessageRouter` configuration check
* Fix `IntegrationGraphServerTests` for new router behavior
* Improve language in docs
