- 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
