The `Flux.takeWhile()` only works if there is data in the `Publisher`
to consume.
We still need to be able to cancel subscription and stop producing even if
there is no data at the moment.
* Change `takeWhile()` to the `doOnSubscribe()` and store `subscription`
in the `volatile` property of the `MessageProducerSupport`
* Cancel such a subscription in the `doStop()` impl
* Propagate `doStop()` to super in the `ZeroMqMessageProducer`
which is only one reactive channel adapter overriding `doStop()`
* Verify in the `ReactiveMessageProducerTests` that subscription is cancelled
for delayed data in the `Publisher`
**Cherry-pick to `5.5.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/3506
Revise the `@Poller` usage in the messaging annotations from an array value
to a single entry with a `ValueConstants.DEFAULT_NONE` for default value
Code review feedback
- Make annotation utils check for NONE more generic
- Move docs to whats-new.adoc
- Polish javadoc
Code review feedback
- simplify docs
- simplify isProvided checks
* Increase a `default-delay="1000"` for `GatewayInterfaceTests`
to test the late reply scenario
* Use a `ConcurrentHashMap` for `JdbcMessageStore.queryCache`
to avoid a possible `ConcurrentModificationException`
We cannot use a `DFA` to set value to the `record` property:
it is marked as `trustedFinal` and cannot be modified via reflection.
Therefore, rework `ControlBusTests` to do other logic than reflection.
The reflection in the unit test is not a justification to stay away
from the `record` structure
* 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`
* Remove `Disposables` since it is a package protected therefore could not be used outside
* Remove the logic relevant ot the `Disposables` in the `MessagingAnnotationPostProcessor` logic
* Replace a `registerSingleton()` and `initializeBean()` with the proper `registerBeanDefinition()`
and `getBean()` usage.
This way bean are going to be destroyed properly
* Remove `MessagingAnnotationPostProcessorChannelCreationTests` since its mocking logic is too vague
and really covered with many other real tests.
* Add `toReactivePublisher(autoStartOnSubscribe)`
The `IntegrationFlowBuilder.toReactivePublisher()` returns a `Publisher<Message<?>>`
which may be subscribed somewhere late in the application logic, e.g.
when WebSocket (or RSocket) subscription is initiated by the external client.
In between application context startup and that subscription moment, the `IntegrationFlow`
must not try to produce messages since there is nothing to consumer them from the
`Publisher<Message<?>>` side.
One of the way is to have a source endpoint not started automatically and control its
lifecycle from the point fo reactive subscription
* Introduce an `IntegrationFlowBuilder.toReactivePublisher(boolean autoStartOnSubscribe)`
to let the framework do a job for an `IntegrationFlow` lifecycle control.
This way end-user doesn't need to know autowire a starting endpoint and use `doOnSubscribe()`
and similar callbacks
* Change `ConsumerEndpointFactoryBean` log message about a `FixedSubscriberChannel`
to `INFO` since an `autoStartup = false` really does not have any effect and there is nothing
for end-user to worry about.
The `IntegrationFlow` knows nothing about each endpoint internals and cannot control which
mark as `autoStartup = false` and which not
* Fix languge in JavaDocs
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
Resolves https://github.com/spring-projects/spring-integration/issues/3648
When configuring a gateway proxy with XML, but specifying the payload expression
on the method `@Gateway` annotation, the expression was ignored, even though it
had been parsed.
`@Payload` worked.
With this change, if both `@Payload` and `@Gateway` are defined on a gateway method,
`@Gateway.payloadExpression` wins.
* Fix doc links.
* Fix Kotlin deprecation warning in the test
* Replace wrong `IntervalTrigger` mentioning in the docs to the proper `PeriodicTrigger`
* Fix code snippet in the `channel-adapter.adoc`
* To align with the rest of Spring portfolio, based on Spring Framework 5.3.x generation,
it is better to upgrade to the Kotlin version supported by SF
* Fix `ClassUtils.isLambda()` to treat `$inlined$` classes as Kotlin lambdas and let
Java DSL to delegate to the `LambdaMessageProcessor` for better expected type handling.
Looks like Kotlin does not carry type info in its lambdas anymore (at least for functions).
* Add `-parameters` to the compiler args for possible method argument names discovery.
Essentially realign with the rest of portfolio
* Fix Kotlin tests for deprecated API
* Introduce high-level API for flows composition
For better end-user experience and more smooth integration logic
decomposition and distribution introduce an `IntegrationFlows.from(IntegrationFlow)`
to let to start the current flow from existing one.
On the other hand introduce an `BaseIntegrationFlowDefinition.to(IntegrationFlow)`
to let to continue the flow logic in the other existing one.
This way we can extract some templating logic into separate `IntegrationFlow` definitions
allowing at the same time to decompose a complex flow definition into logical reusable parts
* * Add more tests
* * Fix Checkstyle violation
* Add `@SuppressWarnings("overloads")` to new `from(IntegrationFlow)` and existing `from(Publisher)`.
Technically it does not make sense since `PublisherIntegrationFlow` is not a `public` class
* * Add docs
* Fix language in JavaDocs according review
* Fix language in the docs after review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
The `spring.integration.properties` can come with the `noAutoStartup`
property where we can specify a source polling channel adapter endpoint to
not start automatically.
Turns out the `SourcePollingChannelAdapterFactoryBean` propagates its
`autoStartup` property unconditionally which will skip the `noAutoStartup` value
because an `AbstractEndpoint.setAutoStartup()` sets an `autoStartupSetExplicitly` state
* Fix `SourcePollingChannelAdapterFactoryBean` to rely on a `Boolean` object state
and don't call target endpoint `setAutoStartup()` if it was not set
* Adjust `spring.integration.properties` in tests to use `noAutoStartup`
for some `SourcePollingChannelAdapterFactoryBean`
* Verify that property was applied in the `IntegrationFlowTests.testWithSupplierMessageSourceImpliedPoller()`
**Cherry-pick to `5.4.x` & `5.3.x`**
* Rework `MessagingGatewayRegistrar` to parse messaging gateway annotation
an `<gateway>` XML using a supplier variant for bean definition.
Such a feature is required by Spring Native - otherwise we would need to
register reflection info for to many internal Spring Integration classes
* Such a change should benefit from regular JDK perspective, too -
we don't do reflection for this kind of bean registrations
* Rework `DefaultMessageAggregatorIntegrationTests` to Junit 5
* Add `@DirtiesContext` to release a `TaskScheduler` resources
when the `ApplicationContext` is closed
Fixes https://github.com/spring-projects/spring-integration/issues/3558
Kotlin lambdas mostly used to configure endpoints in DSL manner
are not really Java lambdas, but rather anonymous classes implementing
respective Java interfaces.
While in most cases such classes carry generic info for their method impls
properly in Java, it is somehow doesn't work well for `GenericHandler`
implemented by Kotlin lambdas
* Wrap provided `GenericHandler` in the `BaseIntegrationFlowDefinition.handle()`
into a Java lambda and call `handle()` recursively to carry an expected type to
the `LambdaMessageProcessor`
* Fix `LambdaMessageProcessor` to handle `ClassUtils.isKotlinUnit()` result of
an invocation as a `null` reply
**Cherry-pick to `5.4.x` & `5.3.x`**
* Make `@IntegrationConverter` Native compatible
* Add `BASE_PACKAGE` into an `IntegrationContextUtils`;
deprecate similar in the `IntegrationConfigUtils`.
This fixes a package tangle between `config` and `context`
* Move `ConverterRegistrar` and `CustomConversionServiceFactoryBean`
into a `config` package since they are package protected and
created their instances in the `IntegrationConverterInitializer`
functional way instead of reflection
* Use new `IntegrationContextUtils.BASE_PACKAGE` constant instead of
deprecated one
* Make `DefaultConfiguringBeanFactoryPostProcessor` `public` to
make it available for Spring Native `trigger` option in the `@NativeHint`
declaration
* Simplify logic around `JsonPath` to just a `ClassUtils.isPresent()`
* Move the `@IntegrationConverter` processing logic into the `ConverterRegistrar`
to avoid reflection via `BeanDefinition` ctor arg manipulation
* Move the reflection logic into a `ConverterParser` which, being a part of XML
configuration, is not going to be compatible with native any way
* Mark `JsonNodeWrapperToJsonNodeConverter` with an `@IntegrationConverter`
since it is not registered via reflection any more
* Expose `MicrometerMetricsCaptorRegistrar.METER_REGISTRY_PRESENT` and use
it in the `IntegrationGraphServer`
* Extract `UnmarshallingTransformer.MIME_MESSAGE_PRESENT` for less
reflection at runtime
* Use `null` for a `ClassLoader` arg in the `ClassUtils.isPresent()`
relying on the default one internally
* * Fix Checkstyle violations
Currently the `MessageHistoryRegistrar` can parse several sources for message history -
`@EnableMessageHistory` and/or `<message-history>`.
Since its logic relies on the reflection it is not compatible with Spring Native.
Plus it causes confusion when several sources are declared so when time comes to
change something in that configuration, we may miss some place to re-align with
our new requirements.
Better to reject extra configurations and enforce end-users to use only one
`@EnableMessageHistory` or `<message-history>`.
This is actually a preferences for many other `@Enable...` in Spring portfolio.
Plus we got a benefit with a Spring Native compatibility
* Fix `MessageHistoryRegistrar` to parse only one `@EnableMessageHistory`.
Register `MessageHistoryConfigurer` function way for Spring Native compatibility
* Clean up `PublisherRegistrar` for better readability
* Fix message history tests which exposed several configurations
* Add JavaDoc into `EnableMessageHistory`
* Refactor `MessageHistoryConfigurer` to let to override patterns configuration
at runtime
* Clean up `message-history.adoc`
* Fix complexity in the `DefaultJmsHeaderMapper` according Sonar report
* Optimize all JMS tests to rely on a shared `ActiveMQConnectionFactory` resource
and disable JMX & statistics for embedded ActiveMQ broker
* Increase timeout for some sporadically failing tests
Related to: https://github.com/spring-cloud/stream-applications/issues/156
The `IntegrationReactiveUtils` uses a general `Flux.retry()`
operator to always retry for all the errors.
On the other hand it has only a `.doOnError(MessagingException.class)`
which leads swallowing all the other exceptions from logs and handling.
* Replace with `retryWhen()` for the `MessagingException` predicate
failing for all other exceptions.
The end-user may add their own `retry` or error handling mechanism to
the returned `Flux` from the `IntegrationReactiveUtils`
**Cherry-pick to 5.4.x**
* GH-3502: More refactoring to avoid reflection
Fixes https://github.com/spring-projects/spring-integration/issues/3502
* Move `ChannelInitializer` bean registration into an `AbstractIntegrationNamespaceHandler` -
it was never used for annotations and Java DSL...
* Rework `IntegrationFlows.fromSupplier()` to call a provided `Supplier` directly -
not via reflection in the `MethodInvokingMessageSource`
* Resolve new Sonar smells
* Rework `EndpointSpec` to accept an expected factory bean instance via ctor arg
instead of reflection
* Rework `Jackson2JsonObjectMapper` to use well-known module instances directly -
not via reflection from their class names
* * Revert `DefaultMethodInvokingMethodInterceptor.methodHandleCache` property definition wrap
* GH-3521: Delayer: schedule release task with TX
Fixes https://github.com/spring-projects/spring-integration/issues/3521
There is a race condition when transactional `MessageStore` is used
for `DelayHandler`, so the message is not visible for reads until after TX is
committed, but a scheduled release task may be already ready after delay
* Register a `TransactionSynchronization` with scheduling a releasing task
when TX is committed
**Cherry-pick to `5.4.x` & `5.3.x`**
* Fix language in delayer.adoc
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* INT-4377: aggregator groupTimeout as Date
JIRA: https://jira.spring.io/browse/INT-4377
Change the `groupTimeoutExpression` logic to let it to be evaluated to `Date`
instance for some fine-grained scheduling use-case, e.g. to determine a
scheduling moment from the group creation time (`timestamp`) instead of a
current message arrival
* Fix language in docs accoridng PR review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* INT-4444: Introduce `@Reactive` & `reactive()`
JIRA: https://jira.spring.io/browse/INT-4444
Right now the high-level API creates a `ReactiveStreamsConsumer`
only when the input channel is a `Publisher<?>` impl or target handler
is a `ReactiveMessageHandler`
* Add `@Reactive[] reactive()` attribute to messaging annotations
* Add `ConsumerEndpointSpec.reactive()`
Both options point to the same `ConsumerEndpointFactoryBean.setReactiveCustomizer()`
making the target endpoint always as a `ReactiveStreamsConsumer` independently of
the input channel and target handler
* Use the `Function` to customize a source `Flux` from the channel
* Test and document a new feature
* * Fix links in docs
* * Fix `ReactiveStreamsTests`
* * Rework `reactive()` attribute of messaging annotations ot a single `@Reactive` value
with default as `@Reactive(ValueConstants.DEFAULT_NONE)`
* Fix language in docs
* Fix `MessagingAnnotationUtils.resolveAttribute()` to use `requiredType.isInstance()`
instead of comparing classes since annotation instances are `Proxy` at runtime
* Miscellaneous improvement for end-user experience
Related to https://stackoverflow.com/questions/66428239/ineligiblemethodexception-found-more-than-one-parameter-type-candidate-spring
Related to https://stackoverflow.com/questions/66397350/spring-integration-how-do-i-call-the-spring-integration-from-spring-controller
The exception thrown from gateway method call for `null` arg is not clear that it cannot be `null`
* Improve `GatewayMethodInboundMessageMapper.mapArgumentsToMessage()` exception message to show args
and explain that it cannot be resolved to `null`
When more than one params are candidates for method invocation, the `MessagingMethodInvokerHelper`
throws an `IneligibleMethodException` which outcome is not clear what to do
* Add `@Payload` annotation recommendation into an exception message to make the reason more cleaner
* Some other code style clean up in the `MessagingMethodInvokerHelper`
* * Fix `PseudoTransactionalMessageSourceTests.testCommit()` for better coverage
* Apply suggestions from code review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
It looks like `OrderedAwareCopyOnWriteArraySetTests` uses
extra repetitions in its tests for nothing relevant to the
tested functionality.
Only the problem that it spawns too many threads
* Use a tmp file in the `FileTests.testFileReadingFlow()`
for writing content and then rename it to the target file.
It looks like we may already have a file just with a short content
and it is picked up by the watch service for processing in the flow
* Rework `AggregatorWithCustomReleaseStrategyTests` to JUnit 5 and remove
extra looping logic since it just does not add any extra coverage just
performance overhead via crating and destroying the same application ctx
* Rework `ChannelAdapterParserTests` to `@SpringJUnitConfig` managed test
JIRA: https://jira.spring.io/browse/INT-3585
Even if this fix doesn't provide an order support,
it does cover an original premise of an issue
With this fix the order of accessors doesn't matter.
* Catch an `AccessException` in the `JsonPropertyAccessor.canRead()` and return `false`
to let the rest accessors to deal with the property requested
Having a bean as a `java.util.Properties` is not confusing
and may lead to some conflicts in the real application.
Plus it is not so easy to configure: need to know all
the possible integration properties - bad end-user experience
* Make an `IntegrationProperties` as a public POJO
for easy configuration of global properties
* Deprecate a presence of the `java.util.Properties`
* Leave framework-created `integrationGlobalProperties` as
a `Properties` instance for backward compatibility
* Fix tests to expose an `IntegrationProperties` bean instead
of deprecated `Properties`
* Fix docs according the change and current recommendations
The logic in the `FluxMessageChannelTests.testFluxMessageChannelCleanUp()`
relies on the `finishLatch.countDown()` which happens before we
release the lock for a group in the aggregator.
So, when we call a `destroy()` as the next statement, we interrupt
the lock instead of unlocking.
This cause a race condition in the Reactor's `Sink` to terminate properly
* Expose `CorrelationHandlerSpec.releaseLockBeforeSend()` and set it to `true`
in the test to properly unlock before we call `finishLatch.countDown()`
* Deprecate a couple introduced before options in favor of their variants
with `set` prefix
* Fix new Sonar smells
* GH-3464: Treat 0 as special for polling endpoint
Fixes https://github.com/spring-projects/spring-integration/issues/3464
The `maxMessagePerPoll <= 0` is considered as an unbound `receive()` call.
End-users would like to have a special treatment for `0` value -
skip the `receive()` call altogether for the current polling cycle.
* Change the logic for a scheduled poller to not call `pollForMessage()`
and just log an INFO when `maxMessagePerPoll == 0`
* Fix reactive poller to deal with `maxMessagePerPoll == 0` properly
* Expose `maxMessagesPerPoll` as a `@ManagedAttribute` to let it to be
modified via Control Bus and JMX
* Test and document a new behavior
* * Fix unused imports in the test class
* Fix language in docs accoridng PR review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
The current `Flux.take()` doesn't allow an arg `< 0` treating
it as an unbound request.
* Change `take()` to `limitRequest()` according strict `MessageSource.receive()`
producing expectations
* Treat `maxMessagesPerPoll < 0` as a `Long.MAX_VALUE` for unbound requests;
`0` is treated in the `limitRequest()` as "no more requests - cancel"
* Revise `AbstractPollingEndpoint` for `LogAccessor` usage
* Add `AbstractPollingEndpoint` class JavaDocs
* Fix tests according `AbstractPollingEndpoint` changes
**Cherry-pick to `5.4.x`**
* Expose `error.ignoreFailures` global property
The global default `errorChannel` can have more then only one logging subscriber.
There is no guarantee in which order subscribers are called and one of them fails
the rest are not called.
* Expose `spring.integration.channels.error.ignoreFailures` global property for
default `errorChannel` and make it `true` by default for better end-user experience
when it is expected to have error being handled by all the subscribers
* * Fix language in the `whats-new.adoc`
* * Fix `JmsTests` for error channel to propagate an exception
* More language fixes for whats-new.adoc
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/3474
The `AbstractTransformer` implementation just wraps a thrown exception into
a `MessageTransformationException`.
It is expected with any method transformer impl (POJO, Lambda direct etc.)
the exception is thrown as is without extra wrapping into an `IllegalStateException`
for consistency
* Fix `LambdaMessageProcessor` to rethrow `RuntimeException` as is
and only wrap into an `IllegalStateException` for all other exceptions
* Fix tests for missed extra stack trace
* Copy request headers when reply is not a Message
Related to https://stackoverflow.com/questions/65887787/spring-integration-unexpected-behavior-on-error-handling-in-splitter-aggregator
Typically `AbstractMessageProducingHandler` implementation with the `shouldCopyRequestHeaders = false`
returns a `Message`(or `MessageBuilder`) as a reply, but when some request handler `Advice` is involved,
we may produce a plain payload and the reply message does not have request headers
* Improve `AbstractMessageProducingHandler` logic to copy request headers into a reply message builder,
when the original reply is not a `Message` or `MessageBuilder`.
This way we improve end-user experience and set them free from copying those headers manually
* Justify a new behavior with the `TransformerTests.testFailedTransformWithRequestHeadersCopy()`
* * Fix Checkstyle violations
Fixes https://github.com/spring-projects/spring-integration/issues/3446
* For better resources utilization provide a `Stream<Message<?>>` API
on the `MessageGroupStore`, `MessageGroup` and `MessageGroupQueue`
* Use this API in the `DelayHandler` when it reschedules persisted messages
* add `Comparable` contract to the initial wrapper so that expressions
containing selection/projection with JsonNode filtering are now possible
* render directly the value of a ValueNode. SpEL parser can now evaluate
correctly the object value when the expression deals with filtering on
`JsonNode` values (<, >, ==...)
* the 2 above changes make this possible:
`property.^[name.getTarget().asText() == 'value1'].name` can now be
written as simply as `property.^[name == 'value1'].name`. The expression
is now fully compatible with other PropertyAccessor types
* rename `ToStringFriendlyJsonNode` to `JsonNodeWrapper` as it reflects
more the capability of the new wrapper
* add a GenericConverter to be able to convert the `JsonNodeWrapper`
class into a `JsonNode` (or its derivative classes) directly while
calling `expression.getValue(..., JsonNode.class)`
* add the possibility to access list items in a json-path like index
=> a negative number will begin from the end of the list <=
* return null value when `JsonNode` cannot handle the property name
(or index)
* Change Json Converter
* Clean up the code style
* Convert JsonPropertyAccessor tests to JUnit 5
* GH-3450: Expose PubSub requireSubscribers option
Fixes https://github.com/spring-projects/spring-integration/issues/3450
* Add `PublishSubscribeChannel` ctors with the `requireSubscribers` option
with a direct delegation to the same option of underlying `BroadcastingDispatcher`
* Add factory methods to the `Channels` & `MessageChannels` to expose
this new `requireSubscribers` option
* Expose an XML `require-subscribers` attribute for the `<publish-subscribe-channel`
* Document this new `requireSubscribers` option
* Introduce a global `spring.integration.channels.error.requireSubscribers` property
for a default `errorChannel`
* Document this property and explain its default `true` in the `error-handling.adoc`
* Remove docs for `spring.integration.postProcessDynamicBeans` since it purpose was
removed since version `5.1` in favor of "always post-process behavior" as it was
always with all the `BeanPostProcessor`s
* * Fix language in docs according review
* * Add more docs about `requireSubscribers` and cross-links between chapters
* GH-3452: Warn Publisher.onError in NullChannel
Fixes https://github.com/spring-projects/spring-integration/issues/3452
The `NullChannel` can subscribe to the `Publisher` payload fully ignoring
the possible data according the `NullChannel` nature.
However an error thrown from the reactive stream processing is also ignored.
* Log WARN message from the `Subscriber.onError()` when `NullChannel`
subscribes to the `Publisher` produced to this channel.
* Mention the logic in the `NullChannel` docs; point to the
`ReactiveRequestHandlerAdvice` for further possible error handling in
the target application
**Cherry-pick to `5.4.x`**
* * Remove unused imports
* * Fix `PseudoTransactionalMessageSourceTests` for `LogAccessor`
from `NullChannel`
