* Add parameterized ExtendedConsumerProperties and ExtendedProducerProperties;
* Added ExtendedPropertiesBinder with the ability of managing per-binding property extensions and interacting the core service;
* Add extensions for Kafka and Rabbit
Addressing PR comments
Changed binder-specific binding prefix to `spring.cloud.stream.binderType.bindings`
Removing kafka-binder.properties
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
- When deserialising the message, replace `contentType` with the `originalContentType` only if the `originalContentType` isn't null which means the original content type was set by the producer before serialising.
- Add and fix tests
This resolves#404
- When deserializing the payload at the consumer endpoint, the non-byte stream payload type requires to use `String` object when the underlying message content-type is of any `text` type contentType (text/plain, text/xml and text/html).
- This fix is only needed to support any non-SCSt applications that will have the 'text/*` contentType of the message that is being deserialized
- Add test
This resolves#403
- Since the `originalContentType` header is preserved across the wire while the serialization/de-serialization processes take place using the deduced `contentType`,
it is convenient to set the `String` value of the `originalContentType` instead of setting the actual `MimeType` object in the header.
- Once the `de-serialization` has taken place, the deduced `contentType` is always replaced with the `originalContentType` as the `contentType` in the message header and this preserved contentType can be resolved to its corresponding MimeType using `ContentTypeResolver` anytime. Hence, using the `String` object for `originalContentType` doesn't do any harm.
- Fix the issue while setting the deduced contentType based on the `payload` when the `originalContentType` is `application/json`
- Currently, the contentType is set to `text/plain` which seems incorrect
- Also, add the `application/json` contentType to return `String` object during de-serialization process.
This resolves#383
Use `text/plain` contentType when the payload is of type `String`
- This is for the intermediate contentType that is used for serialization/de-serialization
- 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
- Add `SmartLifecycle` to BinderAwareChannelResolver
- Add resolved destinations' bindings to the list of `Bindings` that need to be onbound
- Add lowest phase for its lifecycle so that bindings are unbound at the latest
This resolves#370
Use a BindableAdapter to register the output channels
- Add `DyanamicBindable` which is a BindableAdapter that stores the dynamic destination names and handle their unbinding
- Inject `ChannelBindingService` into BinderAwareChannelResolver so that the dynamic binding gets registered with `producerBindings` and can subsequently be used when unbinding
- Fix the `BinderAwareChannelResolver` to resolve the `transport:name` correctly by setting the bean name of the channel to use `name` only (without including transport:)
- Update tests
Use Binding to unbind instead of ChannelBindingService
- subclasses should implement the new onInit() method instead of overriding
- avoids the potential for a NPE in case an overriding subclass did not call the superclass method
- 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
So you can add a parent for instance, like this:
new AggregateApplicationBuilder().from(TwitterStreamApplication.class)
.args(new String[] { "--language=en" }).to(StreamTransformer.class)
.parent(
SpringApplication.run(ModuleApplication.class, args))
.run();
