StreamListenerMethodSetupOrchestrator is an API hook that allows
downstram binder implementations or applications to inject custom
strategies to alter the default StreamListener adapter method invocations.
This PR primarily focuses on allowing customizations on the oubound side.
For the inbound, StreamListenerMethodSetupOrchestrator interface provides
a default implementation.
Corresponding refactoring in StreamListenerAnnotationBeanPostProcessor.
By default StreamListenerAnnotationBeanPostProcessor will use the current
strategies used for invoking the StreamListener adapters. If beans are provided
for orchestration, they take precedence however.
Test changes.
Further polishing in StreamListenerAnnotationBeanPostProcessor.
Resolves#1177Resolves#1179
- missing `}` in one code snippet caused syntax-highlighting problems in asciidoc-aware editors such as atom
- use spaces, not tabs for code snippets
- Change markdown to asciidoc for one snippet
Fixed support for partition properties to be Spring configured
- Fixed support for 'partitionKeyExtractor' and 'partitionSelector' to be Spring configured
- Added new producer properties 'partitionKeyExtractorName' and 'partitionSelectorName'
- Deprecated 'partitionKeyExtractorClass' and 'partitionSelectorClass' properties
- Removed InitilaizingBean from MessageConverterConfigurer
- Updated documentation
- Added additional tests
- Removed PartitionedProducerTest as redundant
- Fixed unrelated to this effort BinderPropertiesTests due to recent Boot changes
polishing
Resolves: https://github.com/spring-cloud/spring-cloud-stream/issues/1174
Introduce `PollableMessageSource` for polled consumers.
The `@Input` annotation can now be applied to interface methods returning
`PollableMessageSource` and binders that can support polled consumers
will wire up an implementation `DefaultPollableMessageSource` which can
then be `poll()`ed.
The `poll()` method takes a `MessageHandler` callback to handle any message
returned by the poll; the method returns true if a message was found and
handled, false otherwise.
By default, acknowledging the message is deferred until the message handler
returns but that auto-acknowledging can be disabled if the user wishes to
defer the acknowledgment until later. Refer to the Spring Integration
documentation for more information (internally, a `MessageSourcePollingTemplate`
is used, which performs the ack/nack when the handler exits).
Usage:
```java
public interface PolledConsumer extends Processor {
@Input("pollableSource")
PollableMessageSource pollableSource();
}
```
and
```java
@Bean
public ApplicationRunner runner(PollableMessageSource pollableSource) {
return args -> pollableSource.poll(message -> {
System.out.println("Polled payload: " + message.getPayload());
});
}
```
Polishing
Fix test; add support to test binder.
Add error channel handling and retry
More polishing driven by Rabbit implementation
- can't use the MessageSourcePollingTemplate within retry since it will fetch more messages
- populate the retry context with data for the error message strategy
Polishing - PR Comments
Fix NPE.
Revert finally in DefaultPollableMessageSource
Catch Throwable - JUnit throws Errors
Change method name to reflect it's only used for polled consumers.
Fix errors when no retry
Remove stack trace print
Fix error message strategy context with no retry
Add separate error MessageHandler for polled consumers.
Resolves#1174Resolves#1176
After changing the `propertyAccessorInjector` `BeanPostProcessor`
just to a simple `InitializingBean` there is no guarantee which bean
will be accessed from the autowire phase.
There are some cases like `@PostConstruct` on the `@Configuration`
wich may be called before this new `propertyAccessorInjector` bean
and we end up with `EvaluationContext` not supplied with additional
`PropertyAccessors`
* For proper `IntegrationEvaluationContextFactoryBean` initialization
add `SpelPropertyAccessorRegistrar` for custom `PA` configuration
* Remove `propertyAccessorInjector` bean altogether
* Resolve deprecation in the `BindingServiceConfiguration`
* Modify `SpelExpressionConverterConfigurationTests` to confirm that
even `@PostConstruct` on `@Configuration` works as expected
Resolves#1178
- added SI provided argumentResolvers to MessageHandlerMethodFactory configuration to ensure that StreamListener annotated methods are consistent with equivalent SI config annotations (e.g., ServiceActivator)
Resolves#1135
Fixes: spring-cloud/spring-cloud-stream#1170
The `messageHandlerHeaderPropagationBeanPostProcessor` causes early
`BeanFactory` initialization via its `SpringIntegrationProperties`
dependency.
This triggers an initialization all the `@ConfigurationProperties`
infrastructure in Spring Boot, when we might not have all the necessary
beans, e.g. `Converter` for some configuration property
* Eliminate the "early `BeanFactory` initialization syndrome" through
the `SmartInitializingSingleton`
**Cherry-pick to 1.3.x**
Some tests in `AbstractBinderTests` are rely on the partition header,
but that one can be populated only by the `PartitionInterceptor`.
To allow it we need to provide appropriate producer properties, like
`partitionKeyExpression`.
* Populate test-based `ProducerProperties` to the
`createBindableChannel()` where `PartitionInterceptor` is populated
* Fix `checkstyle.xml` static imports rule for Mockito-2.x as the
current base line
Fix for the proper `BindingProperties` usage
Fixes: spring-cloud/spring-cloud-stream#1159
* Extract `EmbeddedHeadersChannelInterceptor` class for better navigation
* Use only one shared instance of this interceptor since it is stateless
* Add this `EmbeddedHeadersChannelInterceptor` in the position `0` to
extract embedded headers first
The fix confirmed by Kafka and Kinesis binders
- Deprecated BindableAdapter in favor of providing default operations in Bindable
- Removed 'postProcessBeforeInitialization(..)' from 'BinderAwareRouterBeanPostProcessor' as it is no longer required
Continue the work started in f53ff9585a
- simplified 'isDeclarativeMethodParameter(..) operation of StreamListenerAnnotationBeanPostProcessor
- aligned MessageChannelToInputFluxParameterAdapter and MessageChannelToInputObservableParameterAdapter around the actual Message
- Removed dependency and callbacks to 'BeanFactoryUtils.beansOfTypeIncludingAncestors' in
StreamListenerAnnotationBeanPostProcessor in favor of more appropriate and lazy DI
mechanism provided by Spring.
- Polishing
- Refactored Input/OutputBindingLifecycle to ensure they properly use Spring's dependency injection mechanisms
- Refactored additional tests to use new SI-backed binder
polishing
- changed BinderProperties environment to be Map to support appropriate semantics where keys and values may be non-Strings
- deprecated `setEnvironment(Properties environment)` operation
Resolves#1136
polishing
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/12
Support late binding for binders that don't inherently support it.
Schedule attempts at 30 second intervals (default, configurable).
`IllegalState` and `IllegalArgument` exceptions are fatal.
Add test cases (mocks) to verify rebind after failure
Docs
Polishing - always use the setter on LateBinding
Remove volatile from the fields since they are only accessed from synchronized methods.
Resolves: spring-cloud/spring-cloud-stream#1127
- Added initial implementation of SI-backed stub binder to facilitate
more consistent testing
- Improved BinderAwareChannelResolverTests and ExtendedPropertiesBinderAwareChannelResolverTests
to make use of it
- Added sample Application that demonstrates the binder usage
Addressed PR comments
- Removed MessageSerializationUtils
- Consolidated Message contentType conversion refactoring that was started with d0be34f7cb commit
At this point Message conversion is consolidated in either MessageConverters or In/Out channel interceptors configured in MessageConverterConfigurer
- Separated InboundContentTypeConvertingInterceptor and OutboundContentTypeConvertingInterceptor to isolate logic that needs to be performed by such interceptors for *inbound* and *outbound* messages.
- Documented their purpose via javadocs
- Moved InboundMessageConvertingInterceptor to TestSupportBinder (for now) to only support MessageCollector's 1.3 behavior
Fixes#1109
- Disable outbound content type interceptor when nativeEndoding is enabled on the producer
- Disable inbound content type interceptor when nativeDecoding is enabled on the consumer
- Add a new flag in ConsumerProperties for useNativeDecoding
- Apply ContentType interceptor before the InboundInterceptor to ensure
that content type set on the channel is applied on the message if content type
is missing on the Message.
- Add tests for verifying interceptors are not applied when native encoding/decoding is enabled
- Fix a kryo integration test
- Remove Inbound interceptor from MessageCollector when native encoding is enabled on the producer
- Polishing
Fixes#1106
- This PR builds on the previous PR with commit hash 6c259be6... (pr/1112)
- Added support in AbstractBinderTests to create bindable channel based on determining channel input type based on it's name (i.e., *input*)
- Added LegacyContentTypeHeaderInterceptor to TestSupportBinder to restor the previous behavior of MessageCollector for cases
where Message's payload content type is a variant of 'text'.
- Restored tests that use MessageCollector to depend on proper payload type
- Added 'deserialize' routine back to MessageSerializationUtils
- Polished MessageConverterConfigurer.LegacyContentTypeHeaderInterceptor to remove conditional original-content-type header logic
- Removed default contentType from BindingProperties
- Restored tests that use MessageCollector to depend on proper payload type
- polishing
- fixed Kryo/Java serialization
- Fixed ser/de for "application/json" and contentType equals
- Fixed of ser/de of JSON strings to ensure that Strings are not re-quoted
- Fixed how we comparing contentTypes
- more polishing
- Fixed NPE in MessageSerializationUtils
- Make bindings and consumer groups in new tests added to AbstractBinderTests mutually exclusive
- Remove binding config property `legacyContentTypeHeaderEnabled` that was
introduced to enable legacy content type handling
- Enable LegacyContentTypeInterceptor in 2.0 always, but bypass any
legacy content type handling if the message received is from
a 2.0 producer by checking on the version header
- Introdce a new BinderHeader property for version
- Fix the LegacyContentType related tests
- Remove the check for originalContentType in ReceivingHandler when
the payload received is a byte[]
- Remove unnecessary deserializePayloadIfNecessary calls in ReceivingHandler
- Remove deprecated deserializePayload methods in AbstractBinder
and MessageSerializationUtils
Partly fixes#1106Fixes#1110
Added assertion to org.springframework.cloud.stream.binder.AbstractTestBinder to ensure that MessageChannels that are passed to any of the bind methods are configured with Message Interceptors
