@SpringBootApplication
public class LoggingConsumerApplication {

	public static void main(String[] args) {
		SpringApplication.run(LoggingConsumerApplication.class, args);
	}

	@Bean
	public Consumer<Person> log() {
	    return person -> {
	        System.out.println("Received: " + person);
	    };
	}

	public static class Person {
		private String name;
		public String getName() {
			return name;
		}
		public void setName(String name) {
			this.name = name;
		}
		public String toString() {
			return this.name;
		}
	}
}

	--- [ main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg, bound to: input
	--- [ main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [localhost:5672]
	--- [ main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#2a3a299:0/SimpleConnection@66c83fc8. . .
	. . .
	--- [ main] o.s.i.a.i.AmqpInboundChannelAdapter      : started inbound.input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg
	. . .
	--- [ main] c.e.l.LoggingConsumerApplication         : Started LoggingConsumerApplication in 2.531 seconds (JVM running for 2.897)

@SpringBootApplication
public class SimpleStreamApplication {

	public static void main(String[] args) throws Exception {
		SpringApplication.run(SimpleStreamApplication.class,
		  "--spring.cloud.function.definition=uppercase",
		  "--spring.cloud.stream.bindings.uppercase-in-0.destination=uppercase",
		  "--spring.cloud.stream.bindings.uppercase-in-0.group=myGroup",
		  "--spring.cloud.stream.rabbit.bindings.uppercase-in-0.consumer.auto-bind-dlq=true"
		);
	}

	@Bean
	public Function<Person, Person> uppercase() {
		return personIn -> {
		   throw new RuntimeException("intentional");
	      });
		};
	}
}

. . . .
x-exception-stacktrace:	org.springframework.messaging.MessageHandlingException: nested exception is
      org.springframework.messaging.MessagingException: has an error, failedMessage=GenericMessage [payload=byte[15],
      headers={amqp_receivedDeliveryMode=NON_PERSISTENT, amqp_receivedRoutingKey=input.hello, amqp_deliveryTag=1,
      deliveryAttempt=3, amqp_consumerQueue=input.hello, amqp_redelivered=false, id=a15231e6-3f80-677b-5ad7-d4b1e61e486e,
      amqp_consumerTag=amq.ctag-skBFapilvtZhDsn0k3ZmQg, contentType=application/json, timestamp=1522327846136}]
      at org.spring...integ...han...MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:107)
      at. . . . .
Payload: blah

--spring.cloud.stream.bindings.uppercase-in-0.consumer.max-attempts=1

@StreamRetryTemplate
public RetryTemplate myRetryTemplate() {
    return new RetryTemplate();
}

spring.cloud.stream.bindings.<foo>.consumer.retry-template-name=<your-retry-template-bean-name>

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>

BindingsLifecycleController bindingsController = context.getBean(BindingsLifecycleController.class);
Binding binding = bindingsController.queryState("echo-in-0");
assertThat(binding.isRunning()).isTrue();
bindingsController.changeState("echo-in-0", State.STOPPED);
//bindingsController.stop("echo-in-0") alternative way of changing state. For convenience we expose start/stop and pause/resume operations.
assertThat(binding.isRunning()).isFalse();

==== Actuator
Since actuator and web are optional, you must first add one of the web dependencies as well as add the actuator dependency manually.
The following example shows how to add the dependency for the Web framework:

[source,xml]

The following example shows how to add the dependency for the WebFlux framework:

[source,xml]

You can add the Actuator dependency as follows:
[source,xml]

NOTE: To run Spring Cloud Stream 2.0 apps in Cloud Foundry, you must add `spring-boot-starter-web` and `spring-boot-starter-actuator` to the classpath. Otherwise, the
application will not start due to health check failures.

You must also enable the `bindings` actuator endpoints by setting the following property: `--management.endpoints.web.exposure.include=bindings`.

Once those prerequisites are satisfied. you should see the following in the logs when application start:

	: Mapped "{[/actuator/bindings/{name}],methods=[POST]. . .
	: Mapped "{[/actuator/bindings],methods=[GET]. . .
	: Mapped "{[/actuator/bindings/{name}],methods=[GET]. . .

To visualize the current bindings, access the following URL:
`http://<host>:<port>/actuator/bindings`

Alternative, to see a single binding, access one of the URLs similar to the following:
`http://<host>:<port>/actuator/bindings/<bindingName>`

You can also stop, start, pause, and resume individual bindings by posting to the same URL while providing a `state` argument as JSON, as shown in the following examples:

    curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName
    curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName
    curl -d '{"state":"PAUSED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName
    curl -d '{"state":"RESUMED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName

NOTE: `PAUSED` and `RESUMED` work only when the corresponding binder and its underlying technology supports it. Otherwise, you see the warning message in the logs.
Currently, only Kafka binder supports the `PAUSED` and `RESUMED` states.

=== Binder Configuration Properties

The following properties are available when customizing binder configurations. These properties exposed via `org.springframework.cloud.stream.config.BinderProperties`

They must be prefixed with `spring.cloud.stream.binders.<configurationName>`.

type::
The binder type.
It typically references one of the binders found on the classpath -- in particular, a key in a `META-INF/spring.binders` file.
+
By default, it has the same value as the configuration name.
inheritEnvironment::
Whether the configuration inherits the environment of the application itself.
+
Default: `true`.
environment::
Root for a set of properties that can be used to customize the environment of the binder.
When this property is set, the context in which the binder is being created is not a child of the application context.
This setting allows for complete separation between the binder components and the application components.
+
Default: `empty`.
defaultCandidate::
Whether the binder configuration is a candidate for being considered a default binder or can be used only when explicitly referenced.
This setting allows adding binder configurations without interfering with the default processing.
+
Default: `true`.

[[spring-cloud-stream-overview-custom-binder-impl]]
=== Implementing Custom Binders

In order to implement a custom `Binder`, all you need is to:

- Add the required dependencies
- Provide a ProvisioningProvider implementation
- Provide a MessageProducer implementation
- Provide a MessageHandler implementation
- Provide a Binder implementation
- Create a Binder Configuration
- Define your binder in META-INF/spring.binders

***Add the required dependencies***

Add the `spring-cloud-stream` dependency to your project _(eg. for Maven)_:

[source,xml]

***Provide a ProvisioningProvider implementation***

The `ProvisioningProvider` is responsible for the provisioning of consumer and producer destinations, and is required to convert the logical destinations included in the application.yml or application.properties file in physical destination references.

Below an example of ProvisioningProvider implementation that simply trims the destinations provided via input/output bindings configuration:

[source,java]

***Provide a MessageProducer implementation***

The `MessageProducer` is responsible for consuming events and handling them as messages to the client application that is configured to consume such events.

Here is an example of MessageProducer implementation that extends the `MessageProducerSupport` abstraction in order to poll on a file that matches the trimmed destination name and is located in the project path, while also archiving read messages and discarding consequent identical messages:

[source,java]

NOTE: When implementing a custom binder, this step is not strictly mandatory as you could always resort to using an already existing MessageProducer implementation!

***Provide a MessageHandler implementation***

The `MessageHandler` provides the logic required to produce an event.

Here is an example of MessageHandler implementation:

[source,java]

NOTE: When implementing a custom binder, this step is not strictly mandatory as you could always resort to using an already existing MessageHandler implementation!

***Provide a Binder implementation***

You are now able to provide your own implementation of the `Binder` abstraction. This can be easily done by:

- extending the `AbstractMessageChannelBinder` class
- specifying your ProvisioningProvider as a generic argument of the AbstractMessageChannelBinder
- overriding the `createProducerMessageHandler` and `createConsumerEndpoint` methods

_eg.:_
[source,java]

***Create a Binder Configuration***

It is strictly required that you create a Spring Configuration to initialize the bean for your binder implementation _(and all other beans that you might need)_:

[source,java]

***Define your binder in META-INF/spring.binders***

Finally, you must define your binder in a `META-INF/spring.binders` file on the classpath, specifying both the name of the binder and the full qualified name of your Binder Configuration class:

[source]

== Configuration Options

Spring Cloud Stream supports general configuration options as well as configuration for bindings and binders.
Some binders let additional binding properties support middleware-specific features.

Configuration options can be provided to Spring Cloud Stream applications through any mechanism supported by Spring Boot.
This includes application arguments, environment variables, and YAML or .properties files.

=== Binding Service Properties

These properties are exposed via `org.springframework.cloud.stream.config.BindingServiceProperties`

spring.cloud.stream.instanceCount::
The number of deployed instances of an application.
Must be set for partitioning on the producer side. Must be set on the consumer side when using RabbitMQ and with Kafka if `autoRebalanceEnabled=false`.
+
Default: `1`.

spring.cloud.stream.instanceIndex::
The instance index of the application: A number from `0` to `instanceCount - 1`.
Used for partitioning with RabbitMQ and with Kafka if `autoRebalanceEnabled=false`.
Automatically set in Cloud Foundry to match the application's instance index.

spring.cloud.stream.dynamicDestinations::
A list of destinations that can be bound dynamically (for example, in a dynamic routing scenario).
If set, only listed destinations can be bound.
+
Default: empty (letting any destination be bound).

spring.cloud.stream.defaultBinder::
The default binder to use, if multiple binders are configured.
See <<multiple-binders,Multiple Binders on the Classpath>>.
+
Default: empty.

spring.cloud.stream.overrideCloudConnectors::
This property is only applicable when the `cloud` profile is active and Spring Cloud Connectors are provided with the application.
If the property is `false` (the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors).
When set to `true`, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the `spring.rabbitmq.*` properties provided in the environment for the RabbitMQ binder).
The typical usage of this property is to be nested in a customized environment <<multiple-systems, when connecting to multiple systems>>.
+
Default: `false`.

spring.cloud.stream.bindingRetryInterval::
The interval (in seconds) between retrying binding creation when, for example, the binder does not support late binding and the broker (for example, Apache Kafka) is down.
Set it to zero to treat such conditions as fatal, preventing the application from starting.
+
Default: `30`

[[binding-properties]]
=== Binding Properties

Binding properties are supplied by using the format of `spring.cloud.stream.bindings.<bindingName>.<property>=<value>`.
The `<bindingName>` represents the name of the binding being configured.

For example, for the following function
[source,java]

there are two bindings named `uppercase-in-0` for input and `uppercase-out-0` for output. See <<Binding and Binding names>> for more details.

To avoid repetition, Spring Cloud Stream supports setting values for all bindings, in the format of `spring.cloud.stream.default.<property>=<value>`
and `spring.cloud.stream.default.<producer|consumer>.<property>=<value>` for common binding properties.

When it comes to avoiding repetitions for extended binding properties, this format should be used - `spring.cloud.stream.<binder-type>.default.<producer|consumer>.<property>=<value>`.

==== Common Binding Properties

These properties are exposed via `org.springframework.cloud.stream.config.BindingProperties`

The following binding properties are available for both input and output bindings and must be prefixed with `spring.cloud.stream.bindings.<bindingName>.`
(for example, `spring.cloud.stream.bindings.uppercase-in-0.destination=ticktock`).

Default values can be set by using the `spring.cloud.stream.default` prefix (for example`spring.cloud.stream.default.contentType=application/json`).

destination::
The target destination of a binding on the bound middleware (for example, the RabbitMQ exchange or Kafka topic).
If binding represents a consumer binding (input), it could be bound to multiple destinations, and the destination names can be specified as comma-separated `String` values.
If not, he actual binding name is used instead.
The default value of this property cannot be overridden.
group::
The consumer group of the binding.
Applies only to inbound bindings.
See <<consumer-groups,Consumer Groups>>.
+
Default: `null` (indicating an anonymous consumer).
contentType::
The content type of this binding.
See `<<content-type-management>>`.
+
Default: `application/json`.
binder::
The binder used by this binding.
See `<<multiple-binders>>` for details.
+
Default: `null` (the default binder is used, if it exists).

==== Consumer Properties

These properties are exposed via `org.springframework.cloud.stream.binder.ConsumerProperties`

The following binding properties are available for input bindings only and must be prefixed with `spring.cloud.stream.bindings.<bindingName>.consumer.` (for example, `spring.cloud.stream.bindings.input.consumer.concurrency=3`).

Default values can be set by using the `spring.cloud.stream.default.consumer` prefix (for example, `spring.cloud.stream.default.consumer.headerMode=none`).

autoStartup::
Signals if this consumer needs to be started automatically
+
Default: `true`.
concurrency::
The concurrency of the inbound consumer.
+
Default: `1`.
partitioned::
Whether the consumer receives data from a partitioned producer.
+
Default: `false`.
headerMode::
When set to `none`, disables header parsing on input.
Effective only for messaging middleware that does not support message headers natively and requires header embedding.
This option is useful when consuming data from non-Spring Cloud Stream applications when native headers are not supported.
When set to `headers`, it uses the middleware's native header mechanism.
When set to `embeddedHeaders`, it embeds headers into the message payload.
+
Default: depends on the binder implementation.
maxAttempts::
If processing fails, the number of attempts to process the message (including the first).
Set to `1` to disable retry.
+
Default: `3`.
backOffInitialInterval::
The backoff initial interval on retry.
+
Default: `1000`.
backOffMaxInterval::
The maximum backoff interval.
+
Default: `10000`.
backOffMultiplier::
The backoff multiplier.
+
Default: `2.0`.
defaultRetryable::
Whether exceptions thrown by the listener that are not listed in the `retryableExceptions` are retryable.
+
Default: `true`.
instanceCount::
When set to a value greater than equal to zero, it allows customizing the instance count of this consumer (if different from `spring.cloud.stream.instanceCount`).
When set to a negative value, it defaults to `spring.cloud.stream.instanceCount`.
See `<<spring-cloud-stream-overview-instance-index-instance-count>>` for more information.
+
Default: `-1`.
instanceIndex::
When set to a value greater than equal to zero, it allows customizing the instance index of this consumer (if different from `spring.cloud.stream.instanceIndex`).
When set to a negative value, it defaults to `spring.cloud.stream.instanceIndex`.
Ignored if `instanceIndexList` is provided.
See `<<spring-cloud-stream-overview-instance-index-instance-count>>` for more information.
+
Default: `-1`.
instanceIndexList::
Used with binders that do not support native partitioning (such as RabbitMQ); allows an application instance to consume from more than one partition.
+
Default: empty.
retryableExceptions::
A map of Throwable class names in the key and a boolean in the value.
Specify those exceptions (and subclasses) that will or won't be retried.
Also see `defaultRetriable`.
Example: `spring.cloud.stream.bindings.input.consumer.retryable-exceptions.java.lang.IllegalStateException=false`.
+
Default: empty.
useNativeDecoding::
When set to `true`, the inbound message is deserialized directly by the client library, which must be configured correspondingly (for example, setting an appropriate Kafka producer value deserializer).
When this configuration is being used, the inbound message unmarshalling is not based on the `contentType` of the binding.
When native decoding is used, it is the responsibility of the producer to use an appropriate encoder (for example, the Kafka producer value serializer) to serialize the outbound message.
Also, when native encoding and decoding is used, the `headerMode=embeddedHeaders` property is ignored and headers are not embedded in the message.
See the producer property `useNativeEncoding`.
+
Default: `false`.
multiplex::
When set to true, the underlying binder will natively multiplex destinations on the same input binding.
+
Default: `false`.

==== Advanced Consumer Configuration

For advanced configuration of the underlying message listener container for message-driven consumers, add a single `ListenerContainerCustomizer` bean to the application context.
It will be invoked after the above properties have been applied and can be used to set additional properties.
Similarly, for polled consumers, add a `MessageSourceCustomizer` bean.

The following is an example for the RabbitMQ binder:

====
[source, java]

====

==== Producer Properties

These properties are exposed via `org.springframework.cloud.stream.binder.ProducerProperties`

The following binding properties are available for output bindings only and must be prefixed with `spring.cloud.stream.bindings.<bindingName>.producer.`
(for example, `spring.cloud.stream.bindings.func-out-0.producer.partitionKeyExpression=payload.id`).

Default values can be set by using the prefix `spring.cloud.stream.default.producer` (for example, `spring.cloud.stream.default.producer.partitionKeyExpression=payload.id`).

autoStartup::
Signals if this consumer needs to be started automatically
+
Default: `true`.
partitionKeyExpression::
A SpEL expression that determines how to partition outbound data.
If set, outbound data on this binding is partitioned. `partitionCount` must be set to a value greater than 1 to be effective.
See `<<partitioning>>`.
+
Default: null.

partitionKeyExtractorName::
The name of the bean that implements `PartitionKeyExtractorStrategy`. Used to extract a key used to compute
the partition id (see 'partitionSelector*'). Mutually exclusive with 'partitionKeyExpression'.
+
Default: null.

partitionSelectorName::
The name of the bean that implements `PartitionSelectorStrategy`. Used to determine partition id based
on partition key (see 'partitionKeyExtractor*'). Mutually exclusive with 'partitionSelectorExpression'.
+
Default: null.


partitionSelectorExpression::
A SpEL expression for customizing partition selection.
If neither is set, the partition is selected as the `hashCode(key) % partitionCount`, where `key` is computed through either `partitionKeyExpression`.
+
Default: `null`.
partitionCount::
The number of target partitions for the data, if partitioning is enabled.
Must be set to a value greater than 1 if the producer is partitioned.
On Kafka, it is interpreted as a hint. The larger of this and the partition count of the target topic is used instead.
+
Default: `1`.
requiredGroups::
A comma-separated list of groups to which the producer must ensure message delivery even if they start after it has been created (for example, by pre-creating durable queues in RabbitMQ).
headerMode::
When set to `none`, it disables header embedding on output.
It is effective only for messaging middleware that does not support message headers natively and requires header embedding.
This option is useful when producing data for non-Spring Cloud Stream applications when native headers are not supported.
When set to `headers`, it uses the middleware's native header mechanism.
When set to `embeddedHeaders`, it embeds headers into the message payload.
+
Default: Depends on the binder implementation.

useNativeEncoding::
When set to `true`, the outbound message is serialized directly by the client library, which must be configured correspondingly (for example, setting an appropriate Kafka producer value serializer).
When this configuration is being used, the outbound message marshalling is not based on the `contentType` of the binding.
When native encoding is used, it is the responsibility of the consumer to use an appropriate decoder (for example, the Kafka consumer value de-serializer) to deserialize the inbound message.
Also, when native encoding and decoding is used, the `headerMode=embeddedHeaders` property is ignored and headers are not embedded in the message.
See the consumer property `useNativeDecoding`.
+
Default: `false`.

errorChannelEnabled::
When set to true, if the binder supports asynchroous send results, send failures are sent to an error channel for the destination. See Error Handling for more information.
+
Default: false.

[[content-type-management]]
== Content Type Negotiation

Data transformation is one of the core features of any message-driven microservice architecture. Given that, in Spring Cloud Stream, such data
is represented as a Spring `Message`, a message may have to be transformed to a desired shape or size before reaching its destination. This is required for two reasons:

. To convert the contents of the incoming message to match the signature of the application-provided handler.

. To convert the contents of the outgoing message to the wire format.

The wire format is typically `byte[]` (that is true for the Kafka and Rabbit binders), but it is governed by the binder implementation.

In Spring Cloud Stream, message transformation is accomplished with an `org.springframework.messaging.converter.MessageConverter`.

NOTE: As a supplement to the details to follow, you may also want to read the following https://spring.io/blog/2018/02/26/spring-cloud-stream-2-0-content-type-negotiation-and-transformation[blog post].

=== Mechanics

To better understand the mechanics and the necessity behind content-type negotiation, we take a look at a very simple use case by using the following message handler as an example:

[source, java]

NOTE: For simplicity, we assume that this is the only handler function in the application (we assume there is no internal pipeline).

The handler shown in the preceding example expects a `Person` object as an argument and produces a `String` type as an output.
In order for the framework to succeed in passing the incoming `Message` as an argument to this handler, it has to somehow transform the payload of the `Message` type from the wire format to a `Person` type.
In other words, the framework must locate and apply the appropriate `MessageConverter`.
To accomplish that, the framework needs some instructions from the user.
One of these instructions is already provided by the signature of the handler method itself (`Person` type).
Consequently, in theory, that should be (and, in some cases, is) enough.
However, for the majority of use cases, in order to select the appropriate `MessageConverter`, the framework needs an additional piece of information.
That missing piece is `contentType`.

Spring Cloud Stream provides three mechanisms to define `contentType` (in order of precedence):

. *HEADER*: The `contentType` can be communicated through the Message itself. By providing a `contentType` header, you declare the content type to use to locate and apply the appropriate `MessageConverter`.

. *BINDING*: The `contentType` can be set per destination binding by setting the `spring.cloud.stream.bindings.input.content-type` property.
+
NOTE: The `input` segment in the property name corresponds to the actual name of the destination (which is “input” in our case). This approach lets you declare, on a per-binding basis, the content type to use to locate and apply the appropriate `MessageConverter`.

. *DEFAULT*: If `contentType` is not present in the `Message` header or the binding, the default `application/json` content type is used to
locate and apply the appropriate `MessageConverter`.

As mentioned earlier, the preceding list also demonstrates the order of precedence in case of a tie. For example, a header-provided content type takes precedence over any other content type.
The same applies for a content type set on a per-binding basis, which essentially lets you override the default content type.
However, it also provides a sensible default (which was determined from community feedback).

Another reason for making `application/json` the default stems from the interoperability requirements driven by distributed microservices architectures, where producer and consumer not only run in different JVMs but can also run on different non-JVM platforms.

When the non-void handler method returns, if the return value is already a `Message`, that `Message` becomes the payload. However, when the return value is not a `Message`, the new `Message` is constructed with the return value as the payload while inheriting
headers from the input `Message` minus the headers defined or filtered by `SpringIntegrationProperties.messageHandlerNotPropagatedHeaders`.
By default, there is only one header set there: `contentType`. This means that the new `Message` does not have `contentType` header set, thus ensuring that the `contentType` can evolve.
You can always opt out of returning a `Message` from the handler method where you can inject any header you wish.

If there is an internal pipeline, the `Message` is sent to the next handler by going through the same process of conversion. However, if there is no internal pipeline or you have reached the end of it, the `Message` is sent back to the output destination.

==== Content Type versus Argument Type

As mentioned earlier, for the framework to select the appropriate `MessageConverter`, it requires argument type and, optionally, content type information.
The logic for selecting the appropriate `MessageConverter` resides with the argument resolvers (`HandlerMethodArgumentResolvers`), which trigger right before the invocation of the user-defined handler method (which is when the actual argument type is known to the framework).
If the argument type does not match the type of the current payload, the framework delegates to the stack of the
pre-configured `MessageConverters` to see if any one of them can convert the payload.
As you can see, the  `Object fromMessage(Message<?> message, Class<?> targetClass);`
operation of the MessageConverter takes `targetClass` as one of its arguments.
The framework also ensures that the provided `Message` always contains a `contentType` header.
When no contentType header was already present, it injects either the per-binding `contentType` header or the default `contentType` header.
The combination of `contentType` argument type is the mechanism by which framework determines if message can be converted to a target type.
If no appropriate `MessageConverter` is found, an exception is thrown, which you can handle by adding a custom `MessageConverter` (see `<<spring-cloud-stream-overview-user-defined-message-converters>>`).

But what if the payload type matches the target type declared by the handler method? In this case, there is nothing to convert, and the
payload is passed unmodified. While this sounds pretty straightforward and logical, keep in mind handler methods that take a `Message<?>` or `Object` as an argument.
By declaring the target type to be `Object` (which is an `instanceof` everything in Java), you essentially forfeit the conversion process.

NOTE: Do not expect `Message` to be converted into some other type based only on the `contentType`.
Remember that the `contentType` is complementary to the target type.
If you wish, you can provide a hint, which `MessageConverter` may or may not take into consideration.

==== Message Converters

`MessageConverters` define two methods:

[source, java]

It is important to understand the contract of these methods and their usage, specifically in the context of Spring Cloud Stream.

The `fromMessage` method converts an incoming `Message` to an argument type.
The payload of the `Message` could be any type, and it is
up to the actual implementation of the `MessageConverter` to support multiple types.
For example, some JSON converter may support the payload type as `byte[]`, `String`, and others.
This is important when the application contains an internal pipeline (that is, input -> handler1 -> handler2 ->. . . -> output) and the output of the upstream handler results in a `Message` which may not be in the initial wire format.

However, the `toMessage` method has a more strict contract and must always convert `Message` to the wire format: `byte[]`.

So, for all intents and purposes (and especially when implementing your own converter) you regard the two methods as having the following signatures:

[source, java]

=== Provided MessageConverters

As mentioned earlier, the framework already provides a stack of `MessageConverters` to handle most common use cases.
The following list describes the provided `MessageConverters`, in order of precedence (the first `MessageConverter` that works is used):

. `ApplicationJsonMessageMarshallingConverter`: Variation of the `org.springframework.messaging.converter.MappingJackson2MessageConverter`. Supports conversion of the payload of the `Message` to/from POJO for cases when `contentType` is `application/json` (DEFAULT).
. `ByteArrayMessageConverter`: Supports conversion of the payload of the `Message` from `byte[]` to `byte[]` for cases when `contentType` is `application/octet-stream`. It is essentially a pass through and exists primarily for backward compatibility.
. `ObjectStringMessageConverter`: Supports conversion of any type to a `String` when `contentType` is `text/plain`.
It invokes Object’s `toString()` method or, if the payload is `byte[]`, a new `String(byte[])`.
. `JsonUnmarshallingConverter`: Similar to the `ApplicationJsonMessageMarshallingConverter`. It supports conversion of any type when `contentType` is `application/x-java-object`.
It expects the actual type information to be embedded in the `contentType` as an attribute (for example, `application/x-java-object;type=foo.bar.Cat`).

When no appropriate converter is found, the framework throws an exception. When that happens, you should check your code and configuration and ensure you did not miss anything (that is, ensure that you provided a `contentType` by using a binding or a header).
However, most likely, you found some uncommon case (such as a custom `contentType` perhaps) and the current stack of provided `MessageConverters`
does not know how to convert. If that is the case, you can add custom `MessageConverter`. See <<spring-cloud-stream-overview-user-defined-message-converters>>.

[[spring-cloud-stream-overview-user-defined-message-converters]]
=== User-defined Message Converters

Spring Cloud Stream exposes a mechanism to define and register additional `MessageConverters`.
To use it, implement `org.springframework.messaging.converter.MessageConverter`, configure it as a `@Bean`.
It is then appended to the existing stack of `MessageConverter`s.

NOTE: It is important to understand that custom `MessageConverter` implementations are added to the head of the existing stack.
Consequently, custom `MessageConverter` implementations take precedence over the existing ones, which lets you override as well as add to the existing converters.

The following example shows how to create a message converter bean to support a new content type called `application/bar`:

[source,java]

Spring Cloud Stream also provides support for Avro-based converters and schema evolution.
See `<<schema-evolution>>` for details.

[
== Inter-Application Communication

Spring Cloud Stream enables communication between applications. Inter-application communication is a complex issue spanning several concerns, as described in the following topics:

* `<<spring-cloud-stream-overview-connecting-multiple-application-instances>>`
* `<<spring-cloud-stream-overview-instance-index-instance-count>>`
* `<<spring-cloud-stream-overview-partitioning>>`

[[spring-cloud-stream-overview-connecting-multiple-application-instances]]
=== Connecting Multiple Application Instances

While Spring Cloud Stream makes it easy for individual Spring Boot applications to connect to messaging systems, the typical scenario for Spring Cloud Stream is the creation of multi-application pipelines, where microservice applications send data to each other.
You can achieve this scenario by correlating the input and output destinations of "`adjacent`" applications.

Suppose a design calls for the Time Source application to send data to the Log Sink application. You could use a common destination named `ticktock` for bindings within both applications.

Time Source (that has the binding named `output`) would set the following property:

Log Sink (that has the binding named `input`) would set the following property:

[[spring-cloud-stream-overview-instance-index-instance-count]]
=== Instance Index and Instance Count

When scaling up Spring Cloud Stream applications, each instance can receive information about how many other instances of the same application exist and what its own instance index is.
Spring Cloud Stream does this through the `spring.cloud.stream.instanceCount` and `spring.cloud.stream.instanceIndex` properties.
For example, if there are three instances of a HDFS sink application, all three instances have `spring.cloud.stream.instanceCount` set to `3`, and the individual applications have `spring.cloud.stream.instanceIndex` set to `0`, `1`, and `2`, respectively.

When Spring Cloud Stream applications are deployed through Spring Cloud Data Flow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly.
By default, `spring.cloud.stream.instanceCount` is `1`, and `spring.cloud.stream.instanceIndex` is `0`.

In a scaled-up scenario, correct configuration of these two properties is important for addressing partitioning behavior (see below) in general, and the two properties are always required by certain binders (for example, the Kafka binder) in order to ensure that data are split correctly across multiple consumer instances.

[[spring-cloud-stream-overview-partitioning]]
=== Partitioning

Partitioning in Spring Cloud Stream consists of two tasks:

* `<<spring-cloud-stream-overview-configuring-output-bindings-partitioning>>`
* `<<spring-cloud-stream-overview-configuring-input-bindings-partitioning>>`

[[spring-cloud-stream-overview-configuring-output-bindings-partitioning]]
==== Configuring Output Bindings for Partitioning

You can configure an output binding to send partitioned data by setting one and only one of its `partitionKeyExpression` or `partitionKeyExtractorName` properties, as well as its `partitionCount` property.

For example, the following is a valid and typical configuration:

Based on that example configuration, data is sent to the target partition by using the following logic.

A partition key's value is calculated for each message sent to a partitioned output binding based on the `partitionKeyExpression`.
The `partitionKeyExpression` is a SpEL expression that is evaluated against the outbound message for extracting the partitioning key.

If a SpEL expression is not sufficient for your needs, you can instead calculate the partition key value by providing an implementation of `org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy` and configuring it as a bean (by using the `@Bean` annotation).
If you have more then one bean of type `org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy` available in the Application Context, you can further filter it by specifying its name with the `partitionKeyExtractorName` property, as shown in the following example:

[source]

NOTE: In previous versions of Spring Cloud Stream, you could specify the implementation of `org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy` by setting the  `spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass` property.
Since version 3.0, this property is removed.

Once the message key is calculated, the partition selection process determines the target partition as a value between `0` and `partitionCount - 1`.
The default calculation, applicable in most scenarios, is based on the following formula: `key.hashCode() % partitionCount`.
This can be customized on the binding, either by setting a SpEL expression to be evaluated against the 'key' (through the `partitionSelectorExpression` property) or by configuring an implementation of `org.springframework.cloud.stream.binder.PartitionSelectorStrategy` as a bean (by using the @Bean annotation).
Similar to the `PartitionKeyExtractorStrategy`, you can further filter it by using the `spring.cloud.stream.bindings.output.producer.partitionSelectorName` property when more than one bean of this type is available in the Application Context, as shown in the following example:

[source]

NOTE: In previous versions of Spring Cloud Stream you could specify the implementation of `org.springframework.cloud.stream.binder.PartitionSelectorStrategy` by setting the `spring.cloud.stream.bindings.output.producer.partitionSelectorClass` property.
Since version 3.0, this property is removed.

[[spring-cloud-stream-overview-configuring-input-bindings-partitioning]]
==== Configuring Input Bindings for Partitioning

An input binding (with the binding name `uppercase-in-0`) is configured to receive partitioned data by setting its `partitioned`
property, as well as the `instanceIndex` and `instanceCount` properties on the application itself, as shown in the following example:

The `instanceCount` value represents the total number of application instances between which the data should be partitioned.
The `instanceIndex` must be a unique value across the multiple instances, with a value between `0` and `instanceCount - 1`.
The instance index helps each application instance to identify the unique partition(s) from which it receives data.
It is required by binders using technology that does not support partitioning natively.
For example, with RabbitMQ, there is a queue for each partition, with the queue name containing the instance index.
With Kafka, if `autoRebalanceEnabled` is `true` (default), Kafka takes care of distributing partitions across instances, and these properties are not required.
If `autoRebalanceEnabled` is set to false, the `instanceCount` and `instanceIndex` are used by the binder to determine which partition(s) the instance subscribes to (you must have at least as many partitions as there are instances).
The binder allocates the partitions instead of Kafka.
This might be useful if you want messages for a particular partition to always go to the same instance.
When a binder configuration requires them, it is important to set both values correctly in order to ensure that all of the data is consumed and that the application instances receive mutually exclusive datasets.

While a scenario in which using multiple instances for partitioned data processing may be complex to set up in a standalone case, Spring Cloud Dataflow can simplify the process significantly by populating both the input and output values correctly and by letting you rely on the runtime infrastructure to provide information about the instance index and instance count.

== Testing

Spring Cloud Stream provides support for testing your microservice applications without connecting to a messaging system.


[[spring_integration_test_binder]]
=== Spring Integration Test Binder
The old test binder defined in `spring-cloud-stream-test-support` module was specifically designed to facilitate _unit testing_ of the actual messaging components and thus bypasses some of the core functionality of the binder API.

While such light-weight approach is sufficient for a lot of cases, it usually requires additional _integration testing_ with real binders (e.g., Rabbit, Kafka etc). So we are effectively deprecating it.

To begin bridging the gap between _unit_ and _integration_ testing we've developed a new test binder which uses https://spring.io/projects/spring-integration[Spring Integration] framework
as an in-JVM Message Broker essentially giving you the best of both worlds - a real binder without the networking.

==== Test Binder configuration
To enable Spring Integration Test Binder all you need is:

- Add required dependencies
- Remove the dependency for `spring-cloud-stream-test-support`

***Add required dependencies***

Below is the example of the required Maven POM entries.

[source,xml]

Or for build.gradle.kts

[source,kotlin]
---
testImplementation("org.springframework.cloud:spring-cloud-stream") {
	artifact {
		name = "spring-cloud-stream"
		extension = "jar"
		type ="test-jar"
		classifier = "test-binder"
	}
}
---



==== Test Binder usage

Now you can test your microservice as a simple unit test

[source,java]

And if you need more control or want to test several configurations in the same test suite
you can also do the following:

[source,java]

For cases where you have multiple bindings and/or multiple inputs and outputs, or simply want to be explicit about names of
the destination you are sending to or receiving from, the `send()` and `receive()`
methods of `InputDestination` and `OutputDestination` are overridden to allow you to provide the name of the input and output destination.

Consider the following sample:
[source,java]

and the actual test

[source,java]

For cases where you have additional mapping properties such as `destination` you should use those names. For example, consider a different version of the
preceding test where we explicitly map inputs and outputs of the `uppercase` function to `myInput` and `myOutput` binding names:
[source,java]

You can also use this binder with legacy annotation-based configuration:

[source,java]

In the above you simply create an ApplicationContext with your configuration (your application) while additionally supplying `TestChannelBinderConfiguration`
provided by the framework. Then you access `InputDestination` and `OutputDestination` beans to send/receive messages. In the context of this binder
`InputDestination` and `OutputDestination` emulate remote destinations such as Rabbit _exchange/queue_ or Kafka _topic_.

In the future we plan to simplify the API.

NOTE: In its current state Spring Integration Test Binder only supports the three bindings provided by the framework (Source, Processor, Sink) specifically to promote
light-weight microservices architectures rather then general purpose messaging applications.

==== Test Binder and PollableMessageSource
Spring Integration Test Binder also allows you to write tests when working with `PollableMessageSource` (see <<Using Polled Consumers>> for more details).

The important thing that needs to be understood though is that polling is not event-driven, and that `PollableMessageSource` is a strategy which exposes operation to produce (poll for) a Message (singular).
How often you poll or how many threads you use or where you're polling from (message queue or file system) is entirely up to you;
In other words it is your responsibility to configure Poller or Threads or the actual source of Message. Luckily Spring has plenty of abstractions to configure exactly that.

Let's look at the example:

[source, java]

The above (very rudimentary) example will produce 3 messages in 2 second intervals sending them to the output destination of `Source`
which this binder sends to `OutputDestination` where we retrieve them (for any assertions).
Currently it prints the following:
[source, text]

As you can see the data is the same. That is because this binder defines a default implementation of the actual `MessageSource` - the source
from which the Messages are polled using `poll()` operation. While sufficient for most testing scenarios, there are cases where you may want
to define your own `MessageSource`. To do so simply configure a bean of type `MessageSource` in your test configuration providing your own
implementation of Message sourcing.

Here is the example:

[source, java]

rendering the following output;
[source, text]

NOTE: DO NOT name this bean `messageSource` as it is going to be in conflict with the bean of the same name (different type)
provided by Spring Boot for unrelated reasons.


== Health Indicator

Spring Cloud Stream provides a health indicator for binders.
It is registered under the name `binders` and can be enabled or disabled by setting the `management.health.binders.enabled` property.

To enable health check you first need to enable both "web" and "actuator" by including its dependencies (see <<binding_visualization_control>>)

If `management.health.binders.enabled` is not set explicitly by the application, then `management.health.defaults.enabled` is matched as `true` and the binder health indicators are enabled.
If you want to disable health indicator completely, then you have to set `management.health.binders.enabled` to `false`.

You can use Spring Boot actuator health endpoint to access the health indicator - `/actuator/health`.
By default, you will only receive the top level application status when you hit the above endpoint.
In order to receive the full details from the binder specific health indicators, you need to include the property `management.endpoint.health.show-details` with the value `ALWAYS` in your application.

Health indicators are binder-specific and certain binder implementations may not necessarily provide a health indicator.

If you want to completely disable all health indicators available out of the box and instead provide your own health indicators,
you can do so by setting property `management.health.binders.enabled` to `false` and then provide your own `HealthIndicator` beans in your application.
In this case, the health indicator infrastructure from Spring Boot will still pick up these custom beans.
Even if you are not disabling the binder health indicators, you can still enhance the health checks by providing your own `HealthIndicator` beans in addition to the out of the box health checks.

When you have multiple binders in the same application, health indicators are enabled by default unless the application turns them off by setting `management.health.binders.enabled` to `false`.
In this case, if the user wants to disable health check for a subset of the binders, then that should be done by setting `management.health.binders.enabled` to `false` in the multi binder configurations's environment.
See <<multiple-systems,Connecting to Multiple Systems>> for details on how environment specific properties can be provided.

If there are multiple binders present in the classpath but not all of them are used in the application, this may cause some issues in the context of health indicators.
There may be implementation specific details as to how the health checks are performed. For example, a Kafka binder may decide the status as `DOWN` if there are no destinations registered by the binder.
For this reason, if you include a binder in the classpath, it is advised to use that binder by providing at least one binding (for E.g. through `EnableBinding`).
If you don't have any bindings to provide for this binder, then that is an indication that you don't need to include that binder in the classpath.

Lets take a concrete situation. Imagine you have both Kafka and Kafka Streams binders present in the classpath, but only use the Kafka Streams binder in the application code, i.e. only provide bindings using the Kafka Streams binder.
Since Kafka binder is not used and it has specific checks to see if any destinations are registered, the binder health check will fail.
The top level application health check status will be reported as `DOWN`.
In this situation, you can simply remove the dependency for kafka binder from your application since you are not using it.

== Samples

For Spring Cloud Stream samples, see the https://github.com/spring-cloud/spring-cloud-stream-samples[spring-cloud-stream-samples] repository on GitHub.

=== Deploying Stream Applications on CloudFoundry

On CloudFoundry, services are usually exposed through a special environment variable called https://docs.cloudfoundry.org/devguide/deploy-apps/environment-variable.html#VCAP-SERVICES[VCAP_SERVICES].

When configuring your binder connections, you can use the values from an environment variable as explained on the http://docs.spring.io/spring-cloud-dataflow-server-cloudfoundry/docs/current-SNAPSHOT/reference/htmlsingle/#getting-started-ups[dataflow Cloud Foundry Server] docs.

== Binder Implementations

The following is the list of available binder implementations

* https://cloud.spring.io/spring-cloud-stream-binder-rabbit/[RabbitMQ]
* https://cloud.spring.io/spring-cloud-stream-binder-kafka/[Apache Kafka]
* https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis[Amazon Kinesis]
* https://github.com/spring-cloud/spring-cloud-gcp/tree/master/spring-cloud-gcp-pubsub-stream-binder[Google PubSub _(partner maintained)_]
* https://github.com/SolaceProducts/spring-cloud-stream-binder-solace[Solace PubSub+ _(partner maintained)_]
* https://github.com/Microsoft/spring-cloud-azure/tree/master/spring-cloud-azure-stream-binder/spring-cloud-azure-eventhubs-stream-binder[Azure Event Hubs _(partner maintained)_]
* https://github.com/alibaba/spring-cloud-alibaba/wiki/RocketMQ-en[Apache RocketMQ _(partner maintained)_]

As it was mentioned earlier Binder abstraction is also one of the extension points of the framework. So if you can't find a suitable binder in the preceding list you can implement your own binder on top of Spring Cloud Stream.
In the https://medium.com/@domenicosibilio/how-to-create-a-spring-cloud-stream-binder-from-scratch-ab8b29ee931b[How to create a Spring Cloud Stream Binder from scratch] post a community member documents
in details, with an example, a set of steps necessary to implement a custom binder.
The steps are also highlighted in the `<<spring-cloud-stream-overview-custom-binder-impl>>` section.