* 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/3473
When `AbstractEndpoint.start()` and `AbstractEndpoint.isRunning()`
are called from different thread on `synchronized` methods,
we may end up with a dead lock: one thread waits for monitor on
`synchronized` and another waits for the `lifecycleLock`
* Change `AbstractEndpoint.isRunning()` to a plain `return this.running;`
- there is no reason in a lock around returning this `volatile` property state
**Cherry-pick to 5.4.x & 5.3.x**
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`
* Subscribe to Publisher in the NullChannel
If a payload of the message in the flow is a `org.reactivestreams.Publisher`,
it must be subscribed somewhere downstream to initiate reactive processing.
The `NullChannel` just ignores the message altogether and therefore `Publisher`
is lost
* Check the payload of the message in the `NullChannel` for `Publisher` type
and subscribe to it
* Verify in the test and mention this logic in the docs
**Cherry-pick to 5.4.x**
* Apply suggestions from code review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* Fix compatibility with upgraded dependencies
* Move `What's New` for `5.4` to `history`
* Add `XmlUnit` dependency since the latest AssertJ has deprecated its `isXmlEqualTo()`
* Remove `javax.annotation-api` and `apiguardian-api` as redundant dependencies
Related to https://build.spring.io/browse/INT-MASTER-2240
Turns out the `doOnRequest()` is still not enough to be sure
that subscriber is accepted into the `Publisher`.
Probably because `doOnRequest()` maybe really asked for the
subscription by itself for some logic
* Use `Mono` with `repeatWhenEmpty()` until
`this.sink.currentSubscriberCount() > 0`
Related to https://build.spring.io/browse/INT-MASTERSPRING40-JOB1-1271
* Fix race condition with subscription in the `FluxMessageChannel`:
rely on the `doOnRequest()` to start producing from upstream publishers
* Remove `SourcePollingChannelAdapterTests` since
`Class.getDeclaredConstructor().newInstance()` doesn't rethrow an exception as is
but wrap it into the `InvocationTargetException`.
This is probably the main reason to deprecate a regular `Class.newInstance()`
* Use `LettuceConnectionFactory.setEagerInitialization(true)` in the `RedisAvailableRule`
to avoid possible dead lock with lazy init
* Adjust Redis tests for new `RedisAvailableRule` behavior
The `IntegrationManagementConfigurer` is a `BeanPostProcessor`
so it must not have direct dependency injection for other beans.
In our case it is a `MetricsCaptor` injected from the
`IntegrationManagementConfiguration`
* Fix `IntegrationManagementConfiguration` and `IntegrationManagementConfigurer`
to rely on the `ObjectProvider<MetricsCaptor>` instead
Tested against latest Spring Boot
**Cherry-pick to `5.3.x` & `5.2.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/3425
It turns out that `IntegrationManagementConfigurer.destroy()` is still too
late to remove `gauges` and Micrometer tries to gather them on application
context close, when many beans are already destroyed
* Catch `ContextClosedEvent` in the `IntegrationManagementConfigurer`
and removed `gauges` from there
* Remove `destroy()` impl since it is out of use already:
the `IntegrationManagementConfigurer` is not supposed to be in the
target application directly, so it looks safe to remove the `DisposableBean`
altogether
**Cherry-pick to `5.3.x` & `5.2.x`**
* Rely on `MProducerSupport.active` for `Flux`
* Fix `MessageProducerSupport` to extract an `active` flag and set it before
`isRunning` - the `Flux` subscription relies on the `takeWhile()`
where in case of `autoStartup = false` we will never start consume because
it is set to `true` already after `doStart()`
* Refactor all the `MessageProducerSupport` implementation with similar
`active` state to use already one from the super class
**Cherry-pick to 5.3.x**
* * Remove `MessageProducerSupport.setActive()`
to not let to mutate it from the implementations
* Set `active` to `false` in the `destroy()`
* Clean up and fix typos in the affected `JmsMessageDrivenEndpoint`
* * Pull `active` flag down to the `AbstractEndpoint`
* Set `active = true` in the `start()` before calling `doStart()`
* Do same for `active = false` in the `stop()`
* Clean up `AbstractEndpoint` impls to not call `doStart/doStop` for nothing
* Refactor endpoints to rely on the `active` state from the `AbstractEndpoint`
not their own
See SO for more info:
https://stackoverflow.com/questions/64456946/handle-exceptions-errors-other-than-messagingexception-ie-other-error-excepti
In the versions before `5.2.x` the `Error` was wrapped to the `MessagingException`
in the `AbstractRequestHandlerAdvice` and this one was handled properly
on the gateway level with its `errorChannel` configured
* Fix `MessagingGatewaySupport` to catch all the `Throwable` and try to send them as `ErrorMessage`
to the `errorChannel`.
Otherwise unwrap returned `MessagingException` and re-throw an `Error` as is to keep
the current behavior for non-handled exceptions
* Do not wrap `Error` into a `MessagingException` and re-throw as is if there is no `errorChannel`
**Cherry-pick to 5.3.x & 5.2.x**
With the `time-source` in Spring Cloud Stream we noticed
that first two items have the same (or close) value
when the `time-supplier` is used with the reactive output channel.
Turns out the `contactMap()` with `prefetch == 1` asks two
upstream elements immediately not waiting for a completion
for the fist one.
This way we were not able to update `triggerContext` for the proper
`lastCompletionTime`
* Change `AbstractPollingEndpoint` to `prefetch = 0` for the
"fair" upstream request
* Verify the behavior with the new
`ReactiveInboundChannelAdapterTests.testTimeSupplierConsistency()`
* Mark `ReactiveInboundChannelAdapterTests` as `@LongRunningTest`
since it is now pretty long waiting for all the 3 dates to verify
* Fix `InboundChannelAdapterAnnotationPostProcessor` to properly
register `MethodInvokingMessageSource` when we have more than one
`Supplier` with the `@InboundChannelAdapter` in the same configuration
class
If there is a `MeterRegistry` bean (any) in the application context,
we add a `MicrometerMetricsCaptor` bean which populates meters further
from the components.
In some case we may load a `MicrometerMetricsCaptor` bean too early
so, not all the stuff around `MeterRegistry` maybe ready.
See Spring Boot and its `MetricsAutoConfiguration`
* Fix the `MicrometerMetricsCaptorRegistrar` the way to rely on the
`ObjectProvider<MeterRegistry>` instead
* Add package protected ctor to the `MicrometerMetricsCaptor` to
provide a target `MeterRegistry` on demand
All of that will ensure that we use an already post-processed `MeterRegistry`
including Spring Boot auto-configuration
**Cherry-pick to 5.3.x & 5.2.x**
* Use 'onErrorContinue' to continue receiving messages from Stream
* Send an `ErrorMessage` to the provided `errorChannel` (if any)
* And `@Nullable` to some `MessageProducerSupport` API which definitely
may accept `null`
* Extract common `buildMessageFromRecord()` method in the `ReactiveRedisStreamMessageProducer`,
so all the headers from the stream `Record` are carried to the message independently of the
record state - normal send or error sending
Related to https://github.com/spring-projects/spring-framework/issues/25884
* Don't use `MonoProcessor` & `FluxProcessor` since they are deprecated now
* Fix RSocket components to use an `AtomicReference` header instead of deprecated
`MonoProcessor`
* Introduce an internal `IntegrationRSocketPayloadReturnValueHandler` to handle
properly a Spring Integration case for inbound requests with its
`IntegrationRSocketEndpoint` implementation
* Don't do response handling in the `RSocketInboundGateway` any more
since it is deferred now to the `IntegrationRSocketPayloadReturnValueHandler`
* Remove internal `ChannelSendOperator` since it is out of use from now on
Fixes https://github.com/spring-projects/spring-integration/issues/3306
To prevent API misuse (one can set ID after doing a registration which
will do nothing actually), it is safer to remove set* methods from
`StandardIntegrationFlowRegistration` object.
Providing some related refactoring like making fields 'final',
adjusting usages to keep correct behavior of the builder.
* INT-3045: Add in & out ZeroMq channel adapters
JIRA: https://jira.spring.io/browse/INT-3045
* Add `ZeroMqMessageHandler` to produce messages into one-way ZeroMq sockets
* Add `ZeroMqMessageProducer` to consumer messages from one-way ZeroMq sockets
* Add `ConvertingBytesMessageMapper` impl for the `BytesMessageMapper` to
delegate an actual conversion into the provided `MessageConverter`
* Add `ZeroMqHeaders` for message headers constants representing ZeroMq message attributes
* Fix `ZeroMqChannel` for the proper deferred `zeroMqProxy` evaluation
* Add more JavaDocs
* Fix `ZeroMqChannelTests.testPubSubBind()` to be sure that really all the
subscribed channels get the same message from the `PUB` socket
* * Fix typo in the `ConvertingBytesMessageMapper`
* Add `this` for `doOnError()` in the `ZeroMqChannel` & `ZeroMqMessageProducer`
* Change the bind logic in the `ZeroMqMessageProducer` to `port` and let it to
bind to random port.
The actual port is available later via `getBoundPort()`
* Introduce a `ZeroMqMessageProducer.receiveRaw()` to let received `ZMsg` to
be produce as a `payload`
* Add a logic into `ZeroMqMessageHandler` to treat `ZMsg` in the payload of
request message as is without any conversion
* Fix race condition in the `ZeroMqMessageProducer` to destroy `consumerScheduler`
when the main `Flux` is complete
* * Add Java DSL for ZeroMq components
* Extract `ReactiveMessageHandlerSpec` for `ReactiveMessageHandler` impls
* Add debug message into `EmbeddedJsonHeadersMessageMapper` when cannot `decodeNativeFormat()`
* Make `ReactiveMongoDbMessageHandlerSpec` extending `ReactiveMessageHandlerSpec`
* Make `ZeroMqProxy` `autoStartup` by default
* Add `ZeroMqDslTests` to cover all the Java DSL for ZeroMq
* Introduce a `MimeTypeSerializer` to serialize a `MimeType` into JSON as a plain string;
use it as extra serializer in the `JacksonJsonUtils.messagingAwareMapper()`
* Fix typo for the `AllowListTypeResolverBuilder` inner class
* * Add some docs
* Fix Checkstyle violations
* * More docs
* Fix language in Docs
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
The `Mono.doOnSuccess()` is always called for completed `MonoSink`
even if the value is `null`
* Fix `IntegrationReactiveUtils.messageSourceToFlux()` to check a
message for `null` before calling `AckUtils.autoAck()`
* Add an `logger.error()` message when `doOnError()` is invoked for that `Mono`
**Cherry-pick to 5.3.x**
* GH-3386: Add method signature for gateway proxy
Fixes https://github.com/spring-projects/spring-integration/issues/3386
All the gateway proxy method invokers are supplied with the same
bean name inherited from the proxy.
* Add method signature for proxy method bean to fine-grain the
management for those bean in the logs, message history and metrics
* * Use `Class.getSimpleName()` instead to make gateway names less verbose
* GH-3376: Remove gauges on application ctx close
Fixes https://github.com/spring-projects/spring-integration/issues/3376
The `MeterRegistry` may request meters on application shutdown.
The gauges for channels, handlers and message sources don't make sense
at the moment since all those beans are going to be destroyed.
* Remove gauges for channel, handler and message source numbers from the
`IntegrationManagementConfigurer.destroy()`
**Cherry-pick to 5.3.x & 5.2.x**
* * Add `MicrometerImportSelector` to conditionally load
a `MicrometerMetricsCaptorConfiguration` when `MeterRegistry`
is on class path.
* Make `MicrometerMetricsCaptorConfiguration.integrationMicrometerMetricsCaptor()`
bean dependant on the `ObjectProvider<MeterRegistry>`
* Make `IntegrationManagementConfiguration.managementConfigurer()`
dependant on the `ObjectProvider<MetricsCaptor>`.
This way the `IntegrationManagementConfigurer` is destroyed before
`MeterRegistry` when application context is closed
* Deprecate `MicrometerMetricsCaptor.loadCaptor()` in favor of
`@Import(MicrometerImportSelector.class)`
* * Add `MicrometerMetricsCaptorRegistrar` to register a `MICROMETER_CAPTOR_NAME`
bean when `MeterRegistry` is on class path and no `MICROMETER_CAPTOR_NAME` bean yet.
* Make `IntegrationManagementConfiguration.managementConfigurer()`
dependant on the `ObjectProvider<MetricsCaptor>`.
This way the `IntegrationManagementConfigurer` is destroyed before
`MeterRegistry` when application context is closed
* Deprecate `MicrometerMetricsCaptor.loadCaptor()` in favor of
`@Import(MicrometerMetricsCaptorRegistrar.class)`
* Fix test to make a `MeterRegistry` bean as `static` since
`@EnableIntegrationManagement` depends on this bean definition now
* Fix unused import in the `IntegrationRequestMappingHandlerMapping`
* Fix deprecations from Reactor
* Fix race condition in the `AbstractCorrelatingMessageHandlerTests`:
the discard message is sent much earlier than group is removed from the store.
Iterate group count call until it pass or 10 seconds timeout
* Remove list size assert in the `FtpServerOutboundTests`: looks like
it is not updated properly even if we have an expected content in the collection
* Increase timeout to assert remote files removal in the `FtpRemoteFileTemplateTests`
* Use `tryEmitNext` on Reactor `Sink` since `emitNext` is deprecated
* Add `MessageDeliveryException` emission when `send()` returns `false`
in the `FluxMessageChannel` for `subscribeTo` provided `Publisher`
* GH-3374: Fix scan for BF propagation
Fixes https://github.com/spring-projects/spring-integration/issues/3374
An internal `ClassPathScanningCandidateComponentProvider` instance in the `IntegrationComponentScanRegistrar`
does not propagate a provided `registry`.
* Implement `getRegistry()` on the internal `ClassPathScanningCandidateComponentProvider` to propagate
a provided into the `registerBeanDefinitions()` a `BeanDefinitionRegistry`
* Add `@Conditional` on some scanned `@MessagingGateway` in the `EnableIntegrationTests`
**Cherry-pick to 5.3.x & 5.2.x**
* * Remove unused import
* Restore `unused` warning on the unused registry arg
* Implement `Emission` handling in the `FluxMessageChannel` and `IntegrationReactiveUtils`
* Upgrade to Spring Kafka `2.6.0`
* Fix R2DBC components for deprecation in the Spring Data R2DBC
* Implement `StatementMapper.SelectSpec` for query expression
* Clean up for some sporadic test failures
Fixes https://github.com/spring-projects/spring-integration/issues/3367
Introduce a `requestTimeout` and `triggerTimeout` for `BarrierMessageHandler`
For instance, if an HTTP request sends a message to the barrier,
it should time out after 1min if no trigger message is received.
If the trigger message then arrives late and the HTTP request is no longer waiting,
it shouldn't wait for 1min before discarding the request but do so immediately.
* Enable `MessageChannelReactiveUtilsTests.testOverproducingWithSubscribableChannel()` back
* Disable `StompServerIntegrationTests` again since build on CI is stalled again
