@StreamListener and Message ConversionVersion: Brixton.SR3
+Version: Brixton.BUILD-SNAPSHOT
If you don’t use HTTPS and user credentials, SSH should also work out
of the box when you store keys in the default directories (~/.ssh)
and the uri points to an SSH location,
-e.g. "git@github.com:configuration/cloud-configuration". The
-repository is accessed using JGit, so any documentation you find on
+e.g. "git@github.com:configuration/cloud-configuration". It is important that all
+keys in ~/.ssh/known_hosts are in "ssh-rsa" format. The new "ecdsa-sha2-nistp256" format is NOT supported.
+The repository is accessed using JGit, so any documentation you find on
that should be applicable. HTTPS proxy settings can be set in
~/.git/config or in the same way as for any other JVM process via
system properties (-Dhttps.proxyHost and -Dhttps.proxyPort).
file:/tmp/config,file:/tmp/config/{label}
Spring Cloud Config Server also supports Vault as a backend.
+For more information on Vault see the Vault quickstart guide.
+To enable the config server to use a Vault backend you must run your config server
+with the vault profile. For example in your config server’s application.properties
+you can add spring.profiles.active=vault.
By default the config server will assume your Vault server is running at
+http://127.0.0.1:8200. It also will assume that the name of backend
+is secret and the key is application. All of these defaults can be
+configured in your config server’s application.properties. Below is a
+table of configurable Vault properties. All properties are prefixed with
+spring.cloud.config.server.vault.
| Name | +Default Value | +
|---|---|
host |
+127.0.0.1 |
+
port |
+8200 |
+
scheme |
+http |
+
backend |
+secret |
+
defaultKey |
+application |
+
profileSeparator |
+, |
+
All configurable properties can be found in
+org.springframework.cloud.config.server.environment.VaultEnvironmentRepository.
With your config server running you can make HTTP requests to the server to retrieve +values from the Vault backend. To do this you will need a token for your Vault server.
+First place some data in you Vault. For example
+$ vault write secret/application foo=bar baz=bam
+$ vault write secret/myapp foo=myappsbarNow make the HTTP request to your config server to retrieve the values.
+$ curl -X "GET" "http://localhost:8888/myapp/default" -H "X-Config-Token: yourtoken"
You should see a response similar to this after making the above request.
+{
+ "name":"myapp",
+ "profiles":[
+ "default"
+ ],
+ "label":null,
+ "version":null,
+ "state":null,
+ "propertySources":[
+ {
+ "name":"vault:myapp",
+ "source":{
+ "foo":"myappsbar"
+ }
+ },
+ {
+ "name":"vault:application",
+ "source":{
+ "baz":"bam",
+ "foo":"bar"
+ }
+ }
+ ]
+}When using Vault you can provide your applications with multiple properties sources. +For example, assume you have written data to the following paths in Vault.
+secret/myApp,dev
+secret/myApp
+secret/application,dev
+secret/applicationProperties written to secret/application are available to
+all applications using the Config Server. An
+application with the name myApp would have any properties
+written to secret/myApp and secret/application available to it.
+When myApp has the dev profile enabled than properties written to
+all of the above paths would be available to it, with properties in
+the first path in the list taking priority over the others.
With file-based (i.e. git, svn and native) repositories, resources
with file names in application* are shared between all client
@@ -2180,6 +2337,24 @@ part of the server.
When using Vault as a backend you can share configuration with
+all applications by placing configuration in
+html5/application. For example, if you run this Vault command
$ vault write secret/application foo=bar baz=bamAll applications using the config server will have the properties
+foo and baz available to them.
RestTemplate to the ConfigServicePropertySourceLocator (e.g. by
grabbing it in the bootstrap context and injecting one).
When using Vault as a backend to your config server the client will need to
+supply a token for the server to retrieve values from Vault. This token
+can be provided within the client by setting spring.cloud.config.token
+in bootstrap.yml.
spring:
+ cloud:
+ config:
+ token: YourVaultTokenVault supports the ability to nest keys in a value stored in Vault. For example
+echo -n '{"appA": {"secret": "appAsecret"}, "bar": "baz"}' | vault write secret/myapp -
This command will write a JSON object to your Vault. To access these values in Spring +you would use the traditional dot(.) annotation. For example
+@Value("${appA.secret}")
+String name = "World";The above code would set the name variable to appAsecret.
turbine.combineHostPort=true
+by default Spring Cloud allows Turbine to use the host and port to allow multiple processes per host, per cluster. If you want the native Netflix behaviour built into Turbine that does not allow multiple processes per host, per cluster (the key to the instance id is the hostname), then set the property turbine.combineHostPort=false.
Spring Cloud Stream models this behavior through the concept of a consumer group.
(Spring Cloud Stream consumer groups are similar to and inspired by Kafka consumer groups.)
-Each consumer binding can use the spring.cloud.stream.bindings.input.group property to specify a group name.
-For the consumers shown in the following figure, this property would be set as spring.cloud.stream.bindings.input.group=hdfsWrite or spring.cloud.stream.bindings.input.group=average.
spring.cloud.stream.bindings.<channelName>.group property to specify a group name.
+For the consumers shown in the following figure, this property would be set as spring.cloud.stream.bindings.<channelName>.group=hdfsWrite or spring.cloud.stream.bindings.<channelName>.group=average.
Spring Cloud Stream also supports the use of reactive APIs where incoming and outgoing data is handled as continuous data flows.
+Support for reactive APIs is available via the spring-cloud-stream-reactive, which needs to be added explicitly to your project.
The programming model with reactive APIs is declarative, where instead of specifying how each individual message should be handled, you can use operators that describe functional transformations from inbound to outbound data flows.
+Spring Cloud Stream supports the following reactive APIs:
+Reactor
+RxJava 1.x
+In the future, it is intended to support a more generic model based on Reactive Streams.
+The reactive programming model is also using the @StreamListener annotation for setting up reactive handlers. The differences are that:
the @StreamListener annotation must not specify an input or output, as they are provided as arguments and return values from the method;
the arguments of the method must be annotated with @Input and @Output indicating which input or output will the incoming and respectively outgoing data flows connect to;
the return value of the method, if any, will be annotated with @Output, indicating the input where data shall be sent.
|
+ Note
+ |
+
+
+
+Reactive programming support requires Java 1.8. + |
+
|
+ Note
+ |
+
+
+
+Reactive programming support requires the use of Reactor 3.0.0 and higher. |
+
|
+ Note
+ |
+
+
+
+The use of term |
+
A Reactor based handler can have the following argument types:
+For arguments annotated with @Input, it supports the Reactor type Flux.
+The parameterization of the inbound Flux follows the same rules as in the case of individual message handling: it can be the entire Message, a POJO which can be the Message payload, or a POJO which is the result of a transformation based on the Message content-type header. Multiple inputs are provided;
For arguments annotated with Output, it supports the type FluxSender which connects a Flux produced by the method with an output. Generally speaking, specifying outputs as arguments is only recommended when the method can have multiple outputs;
A Reactor based handler supports a return type of Flux, case in which it must be annotated with @Output. We recommend using the return value of the method when a single output flux is available.
Here is an example of a simple Reactor-based Processor.
+@EnableBinding(Processor.class)
+@EnableAutoConfiguration
+public static class UppercaseTransformer {
+
+ @StreamListener
+ @Output(Processor.OUTPUT)
+ public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
+ return input.map(s -> s.toUpperCase());
+ }
+}The same processor using output arguments looks like this:
+@EnableBinding(Processor.class)
+@EnableAutoConfiguration
+public static class UppercaseTransformer {
+
+ @StreamListener
+ public void receive(@Input(Processor.INPUT) Flux<String> input,
+ @Output(Processor.OUTPUT) FluxSender output) {
+ output.send(input.map(s -> s.toUpperCase()));
+ }
+}RxJava 1.x handlers follow the same rules as Reactor-based one, but will use Observable and ObservableSender arguments and return types.
So the first example above will become:
+@EnableBinding(Processor.class)
+@EnableAutoConfiguration
+public static class UppercaseTransformer {
+
+ @StreamListener
+ @Output(Processor.OUTPUT)
+ public Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
+ return input.map(s -> s.toUpperCase());
+ }
+}The second example above will become:
+@EnableBinding(Processor.class)
+@EnableAutoConfiguration
+public static class UppercaseTransformer {
+
+ @StreamListener
+ public void receive(@Input(Processor.INPUT) Observable<String> input,
+ @Output(Processor.OUTPUT) ObservableSender output) {
+ output.send(input.map(s -> s.toUpperCase()));
+ }
+}Spring Cloud Stream provides support for aggregating multiple applications together, connecting their input and output channels directly and avoiding the additional cost of exchanging messages via a broker. @@ -6148,72 +6544,6 @@ Multiple processors of the same type can be chained together (e.g. for pipelinin For each component, the builder can provide runtime arguments for Spring Boot configuration.
Spring Cloud Stream provides support for RxJava-based processors through the RxJavaProcessor available in spring-cloud-stream-rxjava.
public interface RxJavaProcessor<I, O> {
- Observable<O> process(Observable<I> input);
-}An implementation of RxJavaProcessor will receive Observable as an input that represents the flow of inbound message payloads.
-The process method is invoked once at startup for setting up the data flow.
You can enable the use of RxJava-based processors and use them in your processor application by using the @EnableRxJavaProcessor annotation.
-@EnableRxJavaProcessor is meta-annotated with @EnableBinding(Processor.class) and will create the Processor binding.
-Here is an example of an RxJava-based processor:
@EnableRxJavaProcessor
-public class RxJavaTransformer {
-
- private static Logger logger = LoggerFactory.getLogger(RxJavaTransformer.class);
-
- @Bean
- public RxJavaProcessor<String,String> processor() {
- return inputStream -> inputStream.map(data -> {
- logger.info("Got data = " + data);
- return data;
- })
- .buffer(5)
- .map(data -> String.valueOf(avg(data)));
- }
-
- private static Double avg(List<String> data) {
- double sum = 0;
- double count = 0;
- for(String d : data) {
- count++;
- sum += Double.valueOf(d);
- }
- return sum/count;
- }
-}|
- Note
- |
-
-
-
-When implementing an RxJava processor, it is important to handle exceptions as part of your processing flow.
-Uncaught exceptions will be treated as errors by RxJava and will cause the |
-
Binder selection can either be performed globally, using the spring.cloud.stream.defaultBinder property (e.g., spring.cloud.stream.defaultBinder=rabbit) or individually, by configuring the binder on each channel binding.
-For instance, a processor application which reads from Kafka and writes to RabbitMQ can specify the following configuration:
input and output for read/write respectively) which reads from Kafka and writes to RabbitMQ can specify the following configuration:
Besides the conversions that it supports out of the box, Spring Cloud Stream also supports registering your own message conversion implementations.
+This allows you to send and receive data in a variety of custom formats, including binary, and associate them with specific contentTypes.
+In order to do so, you can create a class that extends AbstractMessageConverter
Spring Cloud Stream provides support for schema-based message converters through its spring-cloud-stream-schema module.
+Currently, the only serialization format supported out of the box is Apache Avro, with more formats to be added in future versions.
The spring-cloud-stream-schema module contains two types of message converters that can be used for Apache Avro serialization:
converters using the class information of the serialized/deserialized objects, or a schema with a location known at startup;
+converters using a schema registry - they locate the schemas at runtime, as well as dynamically registering new schemas as domain objects evolve.
+The AvroSchemaMessageConverter supports serializing and deserializing messages either using a predefined schema or by using the schema information available in the class (either reflectively, or contained in the SpecificRecord).
+If the target type of the conversion is a GenericRecord, then a schema must be set.
For using it, you can simply add it to the application context, optionally specifying one ore more MimeTypes to associate it with.
+The default MimeType is application/avro.
+Here is an example of configuring it in a processor application registering the Apache Avro, without a predefined schema:
@EnableBinding(Sink.class)
+@SpringBootApplication
+public static class SinkApplication {
+
+ ...
+
+ @Bean
+ public MessageConverter userMessageConverter() throws IOException {
+ AvroSchemaMessageConverter avroSchemaMessageConverter {
+ return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes");
+ }
+}Conversely, here is an application that registers a converter with a predefined schema, to be found on the classpath:
+@EnableBinding(Sink.class)
+@SpringBootApplication
+public static class SinkApplication {
+
+ ...
+
+ @Bean
+ public MessageConverter userMessageConverter() throws IOException {
+ AvroSchemaMessageConverter avroSchemaMessageConverter {
+ MessageConverter converter = new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes");
+ converter.setSchemaLocation("classpath:schemas/User.avro");
+ return converter;
+ }
+}In order to understand the schema registry client converter, we will describe the schema registry support first.
+Most serialization models, especially the ones that aim for portability across different platforms and languages, rely on a schema that describes how the data is serialized in the binary payload. +In order to serialize the data and then to interpret it, both the sending and receiving sides must have access to a schema that describes the binary format. +In certain cases, the schema can be inferred from the payload type on serialization, or from the target type on deserialization, but in a lot of cases applications benefit from having access to an explicit schema that describes the binary data format. +A schema registry allows you to store schema information in a textual format (typically JSON) and makes that information accessible to various applications that need it to receive and send data in binary format. +A schema is referenceable as a tuple consisting of:
+a subject that is the logical name of the schema;
+the schema version;
+the schema format which describes the binary format of the data.
+Spring Cloud Stream provides a schema registry server implementation.
+In order to use it, you can simply add the spring-cloud-stream-server artifact to your project and use the @EnableSchemaRegistryServer annotation, adding the schema registry server REST controller to your application.
+This annotation is intended to be used with Spring Boot web applications, and the listening port of the server is controlled by the server.port setting.
+The spring.cloud.stream.schema.server.path setting can be used to control the root path of the schema server (especially when it is embedded in other applications).
The schema registry server uses a relational database to store the schemas. + By default, it uses an embedded database. +You can customize the schema storage using the Spring Boot SQL database and JDBC configuration options.
+A Spring Boot application enabling the schema registry looks as follows:
+@SpringBootApplication
+@EnableSchemaRegistryServer
+public class SchemaRegistryServerApplication {
+ public static void main(String[] args) {
+ SpringApplication.run(SchemaRegistryServerApplication.class, args);
+ }
+}The Schema Registry Server API consists of the following operations:
+POST /Register a new schema.
+Accepts JSON payload with the following fields:
+subject the schema subject;
format the schema format;
definition the schema definition.
Response is a schema object in JSON format, with the following fields:
+id the schema id;
subject the schema subject;
format the schema format;
version the schema version;
definition the schema definition.
GET /{subject}/{format}/{version}Retrieve an existing schema by its subject, format and version.
+Response is a schema object in JSON format, with the following fields:
+id the schema id;
subject the schema subject;
format the schema format;
version the schema version;
definition the schema definition.
GET /schemas/{id}Retrieve an existing schema by its id.
+Response is a schema object in JSON format, with the following fields:
+id the schema id;
subject the schema subject;
format the schema format;
version the schema version;
definition the schema definition.
The client-side abstraction for interacting with schema registry servers is the SchemaRegistryClient interface, with the following structure:
public interface SchemaRegistryClient {
+
+ SchemaRegistrationResponse register(String subject, String format, String schema);
+
+ String fetch(SchemaReference schemaReference);
+
+ String fetch(Integer id);
+
+}Spring Cloud Stream provides out of the box implementations for interacting with its own schema server, as well as for interacting with the Confluent Schema Registry.
+A client for the Spring Cloud Stream schema registry can be configured using the @EnableSchemaRegistryClient as follows:
@EnableBinding(Sink.class)
+ @SpringBootApplication
+ @EnableSchemaRegistryClient
+ public static class AvroSinkApplication {
+ ...
+ }For Spring Boot applications that have a SchemaRegistryClient bean registered with the application context, Spring Cloud Stream will auto-configure an Apache Avro message converter that uses the schema registry client for schema management.
+This eases schema evolution, as applications that receive messages can get easy access to a writer schema that can be reconciled with their own reader schema.
For outbound messages, the MessageConverter will be activated if the content type of the channel is set to application/*+avro, e.g.:
spring.cloud.stream.bindings.output.contentType=application/*+avroDuring the outbound conversion, the message converter will try to infer the schemas of the outbound messages based on their type and register them to a subject based on the payload type using the SchemaRegistryClient.
+If an identical schema is already found, then a reference to it will be retrieved.
+If not, the schema will be registered and a new version number will be provided.
+The message will be sent with a contentType header using the scheme application/[prefix].[subject].v[version]+avro, where prefix is configurable and subject is deduced from the payload type.
For example, a message of the type User may be sent as a binary payload with a content type of application/vnd.user.v2+avro, where user is the subject and 2 is the version number.
When receiving messages, the converter will infer the schema reference from the header of the incoming message and will try to retrieve it. The schema will be used as the writer schema in the deserialization process.
+@StreamListener and Message ConversionThe @StreamListener annotation provides a convenient way for converting incoming messages without the need to specify the content type of an input channel.
@@ -7372,7 +8004,7 @@ You can achieve this scenario by correlating the input and output destinations o
Supposing that a design calls for the Time Source application to send data to the Log Sink application, you can use a common destination named ticktock for bindings within both applications.
Time Source will set the following property:
+Time Source (that has the channel name output) will set the following property:
Log Sink will set the following property:
+Log Sink (that has the channel name input) will set the following property:
An input binding is configured to receive partitioned data by setting its partitioned property, as well as the instanceIndex and instanceCount properties on the application itself, as in the following example:
An input binding (with the channel name input) is configured to receive partitioned data by setting its partitioned property, as well as the instanceIndex and instanceCount properties on the application itself, as in the following example:
@RunWith(SpringJUnit4ClassRunner.class)
+@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = ExampleTest.MyProcessor.class)
@IntegrationTest({"server.port=-1"})
@DirtiesContext
@@ -8485,6 +9117,22 @@ the Spring BOM
Marcin Grzejszczak talking about Spring Cloud Sleuth and Zipkin
+In order to customize the polling mechanism you can create a bean of PollerMetadata type
+with name equal to StreamSpanReporter.POLLER. Here you can find an example of such a configuration.
@Configuration
+public static class CustomPollerConfiguration {
+
+ @Bean(name = StreamSpanReporter.POLLER)
+ PollerMetadata customPoller() {
+ PollerMetadata poller = new PollerMetadata();
+ poller.setMaxMessagesPerPoll(500);
+ poller.setTrigger(new PeriodicTrigger(5000L));
+ return poller;
+ }
+}spring.sleuth.rxjava.schedulers.ignoredthrea
call is made a new Span is created. It gets closed upon receiving the response. In order to block the synchronous RestTemplate features
just set spring.sleuth.web.client.enabled to false.
|
+ Important
+ |
+
+You have to register RestTemplate as a bean so that the interceptors will get injected.
+If you create a RestTemplate instance with a new keyword then the instrumentation WILL NOT work.
+ |
+
Custom instrumentation is set to create and close Spans upon sending and receiving requests. You can customize the ClientHttpRequestFactory
and the AsyncClientHttpRequestFactory by registering your beans. Remember to use tracing compatible implementations (e.g. don’t forget to
-wrap ThreadPoolTaskScheduler in a TraceAsyncListenableTaskExecutor).
ThreadPoolTaskScheduler in a TraceAsyncListenableTaskExecutor). Example of custom request factories:
+@EnableAutoConfiguration
+@Configuration
+public static class TestConfiguration {
+
+ @Bean
+ ClientHttpRequestFactory mySyncClientFactory() {
+ return new MySyncClientHttpRequestFactory();
+ }
+
+ @Bean
+ AsyncClientHttpRequestFactory myAsyncClientFactory() {
+ return new MyAsyncClientHttpRequestFactory();
+ }
+}To block the AsyncRestTemplate features set spring.sleuth.web.async.client.enabled to false.
@@ -9496,6 +10194,30 @@ are creating Spans each time a new task is submitted, invoked or scheduled.
Spring Cloud Sleuth integrates with Spring Integration. It creates spans for publish and
subscribe events. To disable Spring Integration instrumentation, set spring.sleuth.integration.enabled to false.
Spring Cloud Sleuth up till version 1.0.4 is sending invalid tracing headers when using messaging. Those headers are actually
+the same as the ones sent in HTTP (they contain a -) in its name. For the sake of
+backwards compatibility in 1.0.4 we’ve started sending both valid and invalid headers. Please upgrade to 1.0.4 because
+in Spring Cloud Sleuth 1.1 we will remove the support for the deprecated headers.
You can provide the spring.sleuth.integration.patterns pattern to explicitly
+provide the names of channels that you want to include for tracing. By default all channels
+are included.
|
+ Important
+ |
+
+When using the Executor to build a Spring Integration IntegrationFlow remember to use the untraced version of the Executor.
+Decorating Spring Integration Executor Channel with TraceableExecutorService will cause the spans to be improperly closed.
+ |
+
With this metadata, and multiple service instances deployed on localhost, the random value will kick in there to make the instance unique. In Cloudfoundry the spring.application.instance_id will be populated automatically in a Spring Boot Actuator application, so the random value will not be needed.
With this metadata, and multiple service instances deployed on localhost, the random value will kick in there to make the instance unique. In Cloudfoundry the vcap.application.instance_id will be populated automatically in a Spring Boot application, so the random value will not be needed.
@EnableEurekaServer). You can also easily do things like encryption and decryption to support Spring Cloud
-Config clients with secret configuration values.
+Spring Boot CLI provides Spring
+Boot command line features for Spring
+Cloud. You can write Groovy scripts to run Spring Cloud component
+applications (e.g. @EnableEurekaServer). You can also easily do
+things like encryption and decryption to support Spring Cloud Config
+clients with secret configuration values. With the Launcher CLI you
+can also launch services like Eureka, Zipkin, Config Server
+conveniently all at once from the command line (very useful at
+development time).
and install the Spring Cloud plugin:
+and install the Spring Cloud plugins (they are independent, so you can install one or the other or both):
$ mvn install
-$ spring install org.springframework.cloud:spring-cloud-cli:1.1.2.RELEASEThe Launcher CLI can be used to run common services like Eureka,
+Config Server etc. from the command line. To list the available
+services you can do spring cloud --list, and to launch a default set
+of services just spring cloud. To choose the services to deploy,
+just list them on the command line, e.g.
$ spring cloud eureka configserver h2 kafka zipkin+
Summary of supported deployables:
+| Service | +Name | +Address | +Description | +
|---|---|---|---|
eureka |
+Eureka Server |
++ | Eureka server for service registration and discovery. All the other services show up in its catalog by default. |
+
configserver |
+Config Server |
++ | Spring Cloud Config Server running in the "native" profile and serving configuration from the local directory ./launcher |
+
h2 |
+H2 Database |
+http://localhost:9095 (console), jdbc:h2:tcp://localhost:9096/{data} |
+Relation database service. Use a file path for |
+
kafka |
+Kafka Broker |
+http://localhost:9091 (actuator endpoints), localhost:9092 |
++ |
hystrixdashboard |
+Hystrix Dashboard |
++ | Any Spring Cloud app that declares Hystrix circuit breakers publishes metrics on |
+
dataflow |
+Dataflow Server |
++ | Spring Cloud Dataflow server with UI at /admin-ui. Connect the Dataflow shell to target at root path. |
+
zipkin |
+Zipkin Server |
++ | Zipkin Server with UI for visualizing traces. Stores span data in memory and accepts them via HTTP POST of JSON data. |
+