* GH-8625: Add Duration support for `<poller>`
Fixes https://github.com/spring-projects/spring-integration/issues/8625
The duration can be represented in a ISO 8601 format, e.g. `PT10S`, `P1D` etc.
The `<poller>` and `@Poller` don't support such a format.
* Introduce a `PeriodicTriggerFactoryBean` to accept string values for
trigger options and parse them manually before creating the target `PeriodicTrigger`
* Use this `PeriodicTriggerFactoryBean` in the `PollerParser` and `AbstractMethodAnnotationPostProcessor`
where we parse options for the `PeriodicTrigger`
* Modify tests to ensure that feature works
* Document the duration option
* Add more cross-links into polling docs
* Fix typos in the affected doc files
* Add `-parameters` for compiler options since SF 6.1 does not support `-debug` anymore
for method parameter names discovery
* Fix typos
Co-authored-by: Gary Russell <grussell@vmware.com>
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
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`
* 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`
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)
```
