Fix#1080
When class with `@StreamListener` method is proxied we end up
with the double target subscribers registration because we meet
the same method multiple times during `ReflectionUtils.doWithMethods()`
* Use `ReflectionUtils.getUniqueDeclaredMethods()` instead to extract
the list of method candidates
**Cherry-pick to 1.2.x**
Fixes spring-cloud/spring-cloud-stream/#1028
Adding unit test to verify message conversion for error channel
Instead of directly binding on the error channel, use a bridge channel
and apply custom converters on that channel and then bind
Make corresponding test changes
renaming key used for error
Adding tests for exceptions on error channel when content type is set
Fixes#1010
Propagate headers by default.
Add a `copyHeaders` property to `@StreamListener` to allow suppression of header propagation.
Honour header propagation settings
Add `SpringIntegrationProperties` class with
`spring.cloud.stream.integration` prefix for managing
Spring Integration properties.
Add `spring.cloud.stream.integration.messageHandlerNotPropagatedHeaders`
for controlling headers propagated by reply producing handlers.
Fix#943
Remove SI version override
Cleanup
* Simple polishing according PR comments
Add 'eclipse' folder containing Eclipse code
formatter configuration and instructions how to use
it.
Update rule for join_wrapped_lines
- Set to `false`
Resolves#930
Update README
Address review comments
- At StreamListenerBeanPostProcessor, avoid eager creation of inbound/outbound target beans whenever possible
- Use `applicationContext.getType(targetName)` instead of `applicationContext.getBean(targetName)` which checks the type of target beans from bean factory instead of creating the bean eagerly.
- For declarative method invocation, `applicationContext.getBean` is needed as it requires the actual bean as its argument
Resolves#767
Update the test
- ignore bean definition not found exception when the bean definition doesn't exist
See spring-cloud/spring-cloud-stream-binder-kafka#109
See #851
Add Prefix to Partition Headers
originalContentType remains as-is for backwards-compatibility.
Enhance Javadoc
- Remove conditionalOnMissingBean check on the necessary beans that are required for exernal bindings of the child application
- Add test to verify
Resolves#816
`@StreamListener` has support for a `condition` parameter,
that contains a SpEL expression that is evaluated before the
method is invoked.
Fix#682
Move StreamListenerMessageHandler as a top level class
Use dispatching and add test
Refactor dispatching mechanism
Throw error when conditions are used in declarative mode
Make StreamListenerAnnotationBeanPostProcessor overridable
Remove unused field in test
Address some PR comments
Add placeholder resolution
Update how multiple matches work with return values
- Methods with return values are not allowed to specify conditions
- If multiple matches are detected (e.g. multiple methods without
conditions, or a mix of methods with and without conditions)
checks that all of them have no return value;
Fixes#519
Introduces some internal changes to the framework allowing the use
of other types than MessageChannel/SubscribableChannel as bindable
types (e.g. Flux, Observable, KStream, etc.)
- Modify BinderFactory to allow the retrieval and lookup of a
binder not only by name, but also by binding target type
- Subsequent changes to ChannelBindingService and tests to account
for the modified signature
- Introduce BindingTargetFactory as the contract for creating bound
elements
- Remove any references to chanels and bound elements and use
'binding target' systematically across the board
Reinstate our own Checkstyle checks with a reduced set of rules so that header
validation can be performed automatically at compile time.
Use ${project.version} for the checkstyle plugin configuration
Renaming some occurences of 'boundElement' to 'bindingTarget'
Renamed a stray occurence of 'channel'
Fix some occurences of String concatenation in the same line after reformatting
- Use the same mapping of method/parameters for both declarative and message handler Stream Listeners
- Declarative mode of the StreamListener method is determined if at least one of the the method parameters is annotated with @Input or @Output with either bound elements (e.g. channels) or conversion targets from bound elements via a registered StreamListenerParameterAdapter
- If the method is non-declarative then it is considered to be in message handler mode
- Declarative mode now accepts @Output annotation at method level as well along with @SendTo
- Declarative mode also accepts @Input annotation value at the method level via @StreamListener valuue.
- Message handler mode accepts @Input annotation value at the method parameter level and @Output annotation at the method annotation level
- Both @Output and @SendTo annotations are supported to specify the outbound target value while @SendTo is allowed only as a method level annotation
- Add assertions on allowable use of method and parameter annotations
- Add parameterized tests to cover all possible cases
- Message handler tests
- Reactor and RxJava tests
Resolves#664
Add more error handling and simplify the usage patterns
Address review comments
- Support @Input/@Output only for declarative StreamListener methods
- Support multiple @Output only when there is no return type in the StreamListener method
- Add/Update tests
WIP
Refactor PR based on the review comments
- Add schema server implementation
- Add schema client abstraction
- Add schema client implementation for own schema registry server
- Add schema client supporting Confluent schema registry
- Add Avro-based message converter supporting a static schema resource
- Add Avro-based message converter with schema evolution support, via
schema registry client.
- On serialization, the converter register writer schemas with the schema
registry server and augment the content type of outbound message with
schema information.
On deserialization, the reading converter will fetch the schema from the server
if not available locally.
Use class information if schema is not specified
In the case of SpecificRecord and Reflective readers/writers, the class information can be used instead
Make subtype prefix configurable and shorten the subject
- Subtype prefix is now configurable and subject is the lowercase schema name
- Enhance/correct javadoc
Refine AbstractAvroMessageConverter
- distinguish between writer and reader schema when reader is created
Add schema registry and schema registry client docs
Fixes#520Fixes#458 (for non-reactive binders)
Adds spring-cloud-stream-reactive module.
Introduces support for declarative @StreamListener and @Input and @Output annotated parameters.
Add StreamListenerArgumentAdapter and StreamListenerResultAdapter for wrapping bindable inputs
and outputs when passing arguments to declarative @StreamListener.
Adds support for using reactive types (Flux/Observable) with traditional binders.
Introduce FluxSender and ObservableSender for handling multiple streaming outputs per method.
Ensure that errors are caught and logged.
Fixing constructor assertions and Javadoc
Addressing PR comments
- rework @Input/@Output parameter validation
- renamed StreamListenerArgumentAdapter to StreamListenerParameterAdapter
- ensure that parameter direction is accounted for in the current adapters
Fixes#505
The goals are twofold: to simplify the registration of new converters and
to add some consistency to the conversion process, and align them with the
way converters are used in generic Spring Messaging listeners, by describing
two possible transformations: inbound, message (including contentType)->targetClass,
and outbound payload+headers -> message. The idea is for the two
transformations to match the input and output directions of the bound channels.
List of changes:
- Create and configure input/output channels distinctly - reflected
in the definitions of `BindableChannelFactory` and `MessageChannelConfigurer`;
- Replace AbstractFromMessageConverter with bidirectional converters;
- Use the channel direction (input/output) to determine whether
`toMessage` or `fromMessage` will be invoked;
- Use contentType support from `AbstractMessageConverter` to map
converters to mime types instead;
Making converters more robust
Addressing comments
Convert content type headers to String before serializing
Addressing further PR comments
Handle common aspects of message-channel binders:
- added generic, customizable ReceivingHandler and SendingHandler
- made doBindProducer delegate to a series of template methods
- made doBindConsumer delegate to a series of template methods
- moved partitioning to an interceptor, thus fixing #493
Removed unused manual ack handling
- For the message converters that use Jackson ObjectMapper, autowire the ObjectMapper from JacksonAutoConfiguration
- This will allow configuration of ObjectMapper using boot configuration properties `JacksonProperties`
- Add test
This resolves#466
Fixes#156
- Supports Spring Messaging infrastructure for argument mapping and type conversion, sharing the same configured converters as the ones configured on input/output channels;
- Also: refactored AbstractFromMessageConverter to return only payload - conversion to a message enriched with the content type of the channel is done by a separate wrapping converter;
- Update MessageConverterConfigurer that uses a channel interceptor to check if the message contentType header is missing and if there is a binding property for `contentType` header is provided, then set that value as the message contentType header
This resolves#397
Add test
- renamed MessageConverterUtils.getJavaType to MessageConverterUtils.getJavaTypeForJavaObjectContentType and restricted usage only to 'application/x-java-object' types
- Autowire custom message converters into MessageConverterConfigurer
- When configuring the message channel, set the `datatypes` of the channel based on the `supported` datatypes of all the matching message converters
This resolves#381
- Change the Binder interface to support ConsumerProperties/ProducerProperties beans and subclasses
- Binders can subclass the property beans to add new supported properties that will be automatically populated
- Spring Cloud Stream will infer the target type and populate the beans from the environment based on a `spring.cloud.stream.bindings..<bindingName>` prefix
- Remove binder defaults and retain only general binder configurations
TODO: a) decide on instanceIndex/partitionIndex alignment (we do not need both)
b) support `defaultProducer`/`defaultConsumer` properties
c) add leniency control on binding (fail/ignore for unknown properties)
d) add a `requiredProperties` configuration for consumer/producer properties to finely tune the mandatory properties expected to be supported by a bound application
Changes made during review:
- Add support for consumer and producer defaults
- Remove partitionIndex, keeping only instanceIndex
- Fix default properties for Kafka binder
- Move batching properties to Rabbit only
- Modify the integration tests to verify the usecases per app basis (that uses the TestSupportBinder for its channel binding)
- Remove redis binder usage in integration tests
This resolves#380
- bind Spring Integration `errorChannel` as a message producer when the property `spring.cloud.stream.bindings.error.destination` is set
- The error channel will use this property value as the destination name
- Add test
This resolves#329
Rename NoopBindable -> BindableAdapter
Return error channel name for getOutputs
Fix review comments
- Modified integration test to use `Sink` interface
- Add unit test
- The `spring-cloud-stream-tuple` is now split into `spring-tuple`, `spring-integration-tuple` along with `spring-batch-tuple`
- Update dependencies to use the new ones
This resolves#326
- Per-binding properties defined in `BindingProperties` need to hold the binder specific configurations needed when binding producer/consumers.
Previously, these were added in `ChannelBindingServiceProperties` as producer/consumer properties. Since these configuration properties are for
per-binding based on the input/output (consumer/producer) channel being bound it is better to move these properties into `BindingProperties`.
- Add some of the supported producer/consumer properties into BindingProperties
- This fixes#256
Rename CommonBinderProperties -> BinderPropertyKeys
Rename class AbstractBinderPropertiesAccessor.java -> AbstractBindingPropertiesAccessor.java
- Add `MessageChannelConfigurer` that any class that configures the message channel would implement
- Have `message-converter` and `message-history-tracking` configurer implement this interface
- Autowire all available configurers and use them to configure the message channel
- For the message history tracking
- add `channel-binding` properties as part of message header along with `timestamp`
- Introduce CompositeMessageChannelConfigurer
Remove field injection of messageChannelConfigurer
Add tests for message track history configurer
Add content-type configurer test
Address review comments
- Add `toString` to BindingProperties
- Add test case to depict producer/consumer properties
Remove processor channels from tests
