The `this.messagePrefix + groupId + '_' + messageId` logic
is introduced since `6.2` via: 64f8ed5bab
The current fix for new `getMessageFromGroup()` and `removeMessageFromGroupById()`
is to retrieve group and then perform plain `get` and `delete` against just message id
Related to: https://github.com/spring-projects/spring-integration/issues/8773
Fixes https://github.com/spring-projects/spring-integration/issues/8773
The https://github.com/spring-projects/spring-integration/issues/8732 introduced a filtering for messages in group.
So, plain `removeMessage()` doesn't work any more if message is connected to some group yet.
Therefore, `DelayHandler` is failing.
* Introduce `getMessageFromGroup()` and `removeMessageFromGroupById()` into `MessageGroupStore` API
and implement it respectively in all the stores
* Remove `@LongRunningTest` from delayer integration tests and adjust its config to delay not for a long
**Cherry-pick to `6.1.x`**
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
* 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.
**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`
* GH-8732 Don't remove JDBC message if other groups
Fixes https://github.com/spring-projects/spring-integration/issues/8732
When same message is added into different groups,
its record in the `INT_MESSAGE` must remain until the last group is removed
* Improve `JdbcMessageStore.DELETE_MESSAGES_FROM_GROUP` SQL
to ignore those messages for removal which has other group records in the `INT_GROUP_TO_MESSAGE` table
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
* * Fix `JdbcMessageStore.removeMessage()` and `removeMessagesFromGroup()`
to remove from `INT_MESSAGE` only if there is no other records in `INT_GROUP_TO_MESSAGE`
* * Improve `MessageStore.removeMessage()` Javadoc
mentioning `MessageGroupStore` specifics
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
Fixes https://github.com/spring-projects/spring-integration/issues/8664
When an `Observation` is turned to `NoopObservation`,
the `observeWithContext()` fails with `ClassCastException`
since `NoopObservation` serves just plain `Context` not the one we supplied
* Fix `MessagingGatewaySupport.sendAndReceiveWithObservation()` same way
as it is in version `6.0.x`
* Modify `IntegrationObservabilityZipkinTests` to reject some `Observation`
via `observationPredicate()` configuration
**Cherry-pick to `6.1.x`**
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`**
Even if the `PartitionedDispatcher.populatedPartitions()`
is fast, in-memory, non-blocking operation, its active call from the `dispatch()`
on every message sent to the channel may pin the virtual thread.
* Optimize the `populatedPartitions()` for double `if`
where we will step into a `synchronized` block only for first several concurrent messages
**Cherry-pick to `6.1.x`**
The `IntegrationFlowBeanPostProcessor.processIntegrationComponentSpec()`
uses a wrong `generateBeanName()` for component to register making the
provided `id` as a prefix
* Use `generateBeanName(Object instance, String prefix, @Nullable String fallbackId, boolean useFlowIdAsPrefix)`
instead to properly "fallback" to the provided name
**Cherry-pick to `6.1.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`**
* Introduce `PartitionedChannel`
* Implement a `PartitionedChannel` as an extension of the `AbstractExecutorChannel`
* Supply this channel with a `PartitionedDispatcher` which is an extension of the `AbstractDispatcher`
* The target partition is essentially a `UnicastingDispatcher` with a single thead executor
* * Fix language in Javadocs
* Add docs for `PartitionedChannel` into `channel.adoc`
* Pre-populate partitions in the `PartitionedDispatcher`
to ensure a thread number reflection of the partition it is used for (for default `ThreadFactory`)
* Add Javadocs to `public` methods
* Add Java DSL `PartitionedChannelSpec` and respective factory methods into `Channels`
* Use `IntegrationMessageHeaderAccessor.CORRELATION_ID` header as a default partition key
* * Fix language in Javadocs
* Fix Checkstyle violations
* Mark `PartitionedDispatcher.prePopulatedPartitions()` with `synchronized`
* * Populate partitions from `PartitionedDispatcher` ctor
and when a custom `ThreadFactory` is set
* * Clear `executors` in `populatedPartitions()`
* * Bring back `synchronized populatedPartitions()` and use it in the `dispatch()`
Turns out `done` in the `SinkManyEmitterProcessor`
is set to `true` when we already processed all the data.
Therefore, it is better to `await().until()` for `done`
condition in the end of test
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`**
When we configure an output channel for the handler as a `FluxMessageChannel`,
in most cases we assume an async processing for the reply.
Especially this is critical when reply is a reactive type, so in the async
mode it is "flattened" by the mentioned `FluxMessageChannel`.
Therefore, it is a bit awkward to require to set async explicitly,
when we have already configured output channel as a `FluxMessageChannel`
* Remove redundant config for async from `R2dbcDslTests`
* Mention the change in the docs
* GH-3763: Add `handleReactive()` for Java DSL
Fixes https://github.com/spring-projects/spring-integration/issues/3763
Add a convenient terminal operator to `BaseIntegrationFlowDefinition`
based on a `ReactiveMessageHandler`.
Also add an overload like `handleReactive(ReactiveMessageHandlerSpec)`
to let end-user to choose a protocol-specific channel adapter
* * Fix `Namespace Factory` wording in the `BaseIntegrationFlowDefinition` Javadocs
* Fix language in Docs
Co-authored-by: Gary Russell <grussell@vmware.com>
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
* Fix `IntegrationGraphServerTests` to not use `net.minidev.json.JSONArray` API
since it is not available as transitive from `json-path` in the `testCompileClasspath`
* Fix NPEs in DSL Specs
The `BaseWsInboundGatewaySpec` and `TailAdapterSpec` don't override super methods
and when we call them we fail with NPE since `target` was not populated.
* Fix `BaseWsInboundGatewaySpec` and its inheritors to populate the `target`
from their ctors.
* Remove redundant methods from `BaseWsInboundGatewaySpec` hierarchy
* Propagate `AbstractWebServiceInboundGateway` properties directly to the target
from the `BaseWsInboundGatewaySpec` inheritors
* Override `MessageProducerSpec` methods in the `TailAdapterSpec`
to populate respective option into the `FileTailInboundChannelAdapterFactoryBean`
* Expose more missed options for producer endpoint in the `FileTailInboundChannelAdapterFactoryBean`
**Cherry-pick to `6.0.x` & `5.5.x`**
* * `acceptIfNotNull()` for `errorMessageStrategy` in the `FileTailInboundChannelAdapterFactoryBean`
* * Fix `Unmarshaller` population logic in the `MarshallingWsInboundGatewaySpec`
Currently, the `MessageTriggerAction.trigger` is configured
on the `BaseIntegrationFlowDefinition` to be called via reflection
in the `MessagingMethodInvokerHelper`
* Represent a `MessageTriggerAction.trigger` as a `Consumer<Message<?>>` method reference
and use a `LambdaMessageProcessor` for direct call
* Add a `Consumer` support for `LambdaMessageProcessor`
* GH-8586: Deprecate IntegrationComponentSpec.get()
Fixes https://github.com/spring-projects/spring-integration/issues/8586
The `IntegrationComponentSpec` is not a plain wrapper around single component.
Sometimes it comes with several components where all of them must be registered
as beans.
If `IntegrationComponentSpec.get()` is called from end-user code, we may lose
other related components, for example filters in the `FileInboundChannelAdapterSpec`.
* Deprecate `IntegrationComponentSpec.get()` with no-op for end-user,
rather encourage to leave it as is and let the framework take care about its lifecycle
and related components registration
* Fix `IntegrationComponentSpec` logic to deal as a simple `FactoryBean` instead of
extra overhead via `AbstractFactoryBean`
* Use `IntegrationComponentSpec.getObject()` in the framework code where `get()` was called
* Fix tests to expose `IntegrationComponentSpec` as beans instead of previously called `get()`
* Some other clean up and typos fixes in the affected classes
* Document the change
* * Revert `ObjectStringMapBuilder` in the `KafkaInboundGatewaySpec.getComponentsToRegister()`
* Fix language in docs
Co-authored-by: Gary Russell <grussell@vmware.com>
* * Remove trailing whitespace in the `ScriptMessageSourceSpec`
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
* 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>
Fixes https://github.com/spring-projects/spring-integration/issues/3869
* Move `ContextHolderRequestHandlerAdvice` to the `core` module for more general purposes
* Add `ContextHolderRequestHandlerAdviceTests`
* Rework `DelegatingSessionFactoryTests` to rely on the `ContextHolderRequestHandlerAdvice`.
This allows us to remove unnecessary XML configuration for this test class
* Document the feature
Some tests deliberately don't expect a reply, but they still block
on a gateway's `sendAndReceive()`
* Improve `Jsr223ScriptExecutingMessageProcessorTests`
to verify that script variables work
Currently, many timeouts in the project are like `-1` or other negative value
with a meaning to wait indefinitely.
According to distributed systems design and bad demo developing experience
it is not OK to block forever.
* Rework most of the timeouts in the framework to be `30` seconds.
Only one remained as `1` seconds is a `PollingConsumer` where it is
better to not block even for those 30 seconds when no messages in the queue,
but let the polling task be rescheduled.
* Remove the `MessagingGatewaySupport.replyTimeout` propagation down to the
`PollingConsumer` correlator where it was a `-1` before and blocked
the polling thread on the `Queue.poll()`.
This fixed the problem with a single thread in a pool for auto-configured `TaskScheduler`.
Now with 1 seconds wait time we are able to switch to other scheduled tasks
even with only 1 thread in the pool
* Warn about dropped message in filter
Buy default the `MessageFilter` just drops a discarded message silently.
If a request-reply gateway is used upstream, then it becomes unclear
why the flow sometimes doesn't work.
* Add a waring log ot emit for default behavior.
This still doesn't fix the request-reply problem, but at least it can
give a clue what is going on
* * Mention `nullChannel` variant to ignore even warn
* Fix language in docs
Co-authored-by: Gary Russell <grussell@vmware.com>
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
It is not obvious from the `@MessagingGateway` Javadocs how to configure
an individual service method
* Add links and explanation about per-method `@Gateway`
annotation into the `@MessagingGateway`
* 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`**
Fixes https://github.com/spring-projects/spring-integration/issues/3772
* Add basic To/From Protocol Buffer's com.google.protobuf.Message transformers.
* Allow the proto_type header to specify the type.
* Add tests.
* add Protobuf docs
* Leverage the Spring ProtobufMessageConverter
* move protobuf-java-util to test dependecies as optional
* protobuf docs improvements
* improve the expected type handling
* expected type expression
* fix indentation
* fix indentation for generated test classes
* suppress style check for proto generated classes
* address the transformer doc format
* fix whats new merge conflict
* fix doc sample code
* Some code clean up; fixing typos
* GH-3957: Add JmsInboundGateway.replyToExpression
Fixes https://github.com/spring-projects/spring-integration/issues/3957
Sometimes we cannot use a standard `JmsReplyTo` property for sending replies from the server.
A `DestinationResolver` API does not have access to the request message.
* Introduce a `ChannelPublishingJmsMessageListener.replyToExpression` property to evaluate
a reply destination against request JMS `Message`
* Use this expression only of no `JmsReplyTo` property
* Expose this property on Java DSL level
* To simplify end-user experience with lambda configuration for this property, introduce a `CheckedFunction`
which essentially re-throws exception "sneaky" way
* Fix Javadoc for `CheckedFunction`
* * Fix language in docs
* Fix Javadocs lines length
* Regular `catch` and re-throw in the `CheckedFunction`
