SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Avro serialization and schema management support

Browse Source 
- 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
This commit is contained in:
Vinicius Carvalho
2016-07-28 18:33:53 -04:00
committed by Marius Bogoevici
parent 8dd22ebca0
commit 4422b21438
50 changed files with 3261 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pom.xml
View File

@@ -44,6 +44,8 @@
<module>spring-cloud-stream-integration-tests</module>
<module>spring-cloud-stream-docs</module>
<module>spring-cloud-stream-reactive</module>
<module>spring-cloud-stream-schema</module>
<module>spring-cloud-stream-schema-server</module>
</modules>
<build>
<pluginManagement>

199
spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc
View File

@@ -1333,6 +1333,205 @@ This is because the payload at the module's output channel is already a String s
While conversion is supported for both input and output channels, it is especially recommended to be used for the conversion of outbound messages.
For the conversion of inbound messages, especially when the target is a POJO, the `@StreamListener` support will perform the conversion automatically.
=== Customizing message conversion
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`
=== Schema-based message converters
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.
==== Apache Avro Message Converters
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.
===== Converters with schema support
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:
[source,java]
----
@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:
[source,java]
----
@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.
=== Schema Registry Support
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.
==== Schema Registry Server
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 http://docs.spring.io/spring-boot/docs/current-SNAPSHOT/reference/htmlsingle/#boot-features-sql[Spring Boot SQL database and JDBC configuration options].
A Spring Boot application enabling the schema registry looks as follows:
[source,java]
----
@SpringBootApplication
@EnableSchemaRegistryServer
public class SchemaRegistryServerApplication {
public static void main(String[] args) {
SpringApplication.run(SchemaRegistryServerApplication.class, args);
}
}
----
===== Schema Registry Server API
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.
==== Schema Registry Client
The client-side abstraction for interacting with schema registry servers is the `SchemaRegistryClient` interface, with the following structure:
[source,java]
----
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:
[source,java]
----
@EnableBinding(Sink.class)
@SpringBootApplication
@EnableSchemaRegistryClient
public static class AvroSinkApplication {
...
}
----
==== Avro Schema Registry Client Message Converters
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.:
[source,properties]
----
spring.cloud.stream.bindings.output.contentType=application/*+avro
----
During 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 Conversion
The `@StreamListener` annotation provides a convenient way for converting incoming messages without the need to specify the content type of an input channel.

21
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java
View File

@@ -16,7 +16,6 @@
package org.springframework.cloud.stream.config;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -38,7 +37,6 @@ import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.tuple.Tuple;
@@ -83,16 +81,15 @@ public class MessageChannelConfigurerTests {
@Test
public void testObjectMapperConfig() throws Exception {
CompositeMessageConverter compositeMessageConverter = messageConverterFactory.getMessageConverterForType(MimeTypeUtils.APPLICATION_JSON);
List<MessageConverter> converters = compositeMessageConverter.getConverters();
for (MessageConverter converter : converters) {
DirectFieldAccessor converterAccessor = new DirectFieldAccessor(converter);
ObjectMapper objectMapper = (ObjectMapper) converterAccessor.getPropertyValue("objectMapper");
// assert that the ObjectMapper used by the converters is compliant with the Boot configuration
assertThat(!objectMapper.getSerializationConfig().isEnabled(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)).withFailMessage("SerializationFeature 'WRITE_DATES_AS_TIMESTAMPS' should be disabled");
// assert that the globally set bean is used by the converters
assertThat(objectMapper).isSameAs(this.objectMapper);
}
MessageConverter converter = messageConverterFactory.getMessageConverterForType(MimeTypeUtils
.APPLICATION_JSON);
DirectFieldAccessor converterAccessor = new DirectFieldAccessor(converter);
ObjectMapper objectMapper = (ObjectMapper) converterAccessor.getPropertyValue("objectMapper");
// assert that the ObjectMapper used by the converters is compliant with the Boot configuration
assertThat(!objectMapper.getSerializationConfig().isEnabled(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS))
.withFailMessage("SerializationFeature 'WRITE_DATES_AS_TIMESTAMPS' should be disabled");
// assert that the globally set bean is used by the converters
assertThat(objectMapper).isSameAs(this.objectMapper);
}
@EnableBinding(Sink.class)

67
spring-cloud-stream-schema-server/pom.xml Normal file
View File

@@ -0,0 +1,67 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2016 the original author or authors.
~
~ Licensed under the Apache License, Version 2.0 (the "License");
~ you may not use this file except in compliance with the License.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>spring-cloud-stream-parent</artifactId>
<groupId>org.springframework.cloud</groupId>
<version>1.1.0.BUILD-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-schema-server</artifactId>
<version>1.1.0.BUILD-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<version>1.4.192</version>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>1.8.1</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

38
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/EnableSchemaRegistryServer.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.cloud.stream.schema.server.config.SchemaServerConfiguration;
import org.springframework.context.annotation.Import;
/**
* Enables the schema registry server enpoints.
*
* @author Vinicius Carvalho
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(SchemaServerConfiguration.class)
public @interface EnableSchemaRegistryServer {
}

31
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerApplication.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author Vinicius Carvalho
*/
@SpringBootApplication
@EnableSchemaRegistryServer
public class SchemaRegistryServerApplication {
public static void main(String[] args) {
SpringApplication.run(SchemaRegistryServerApplication.class, args);
}
}

50
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerConfiguration.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.config;
import java.util.HashMap;
import java.util.Map;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.schema.server.controllers.ServerController;
import org.springframework.cloud.stream.schema.server.repository.SchemaRepository;
import org.springframework.cloud.stream.schema.server.support.AvroSchemaValidator;
import org.springframework.cloud.stream.schema.server.support.SchemaValidator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
/**
* @author Vinicius Carvalho
*/
@Configuration
@EnableJpaRepositories(basePackageClasses = SchemaRepository.class)
@EnableConfigurationProperties(SchemaServerProperties.class)
public class SchemaServerConfiguration {
@Bean
public ServerController serverController(SchemaRepository repository) {
return new ServerController(repository, schemaValidators());
}
@Bean
public Map<String, SchemaValidator> schemaValidators() {
Map<String, SchemaValidator> validatorMap = new HashMap<>();
validatorMap.put("avro", new AvroSchemaValidator());
return validatorMap;
}
}

42
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerProperties.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author Vinicius Carvalho
*/
@ConfigurationProperties("spring.cloud.stream.schema.server")
public class SchemaServerProperties {
/**
* Prefix for configuration resource paths (default is empty). Useful when embedding
* in another application when you don't want to change the context path or servlet
* path.
*/
private String path;
public String getPath() {
return this.path;
}
public void setPath(String path) {
this.path = path;
}
}

144
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/controllers/ServerController.java Normal file
View File

@@ -0,0 +1,144 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.controllers;
import java.util.List;
import java.util.Map;
import org.springframework.cloud.stream.schema.server.model.Schema;
import org.springframework.cloud.stream.schema.server.repository.SchemaRepository;
import org.springframework.cloud.stream.schema.server.support.InvalidSchemaException;
import org.springframework.cloud.stream.schema.server.support.SchemaNotFoundException;
import org.springframework.cloud.stream.schema.server.support.SchemaValidator;
import org.springframework.cloud.stream.schema.server.support.UnsupportedFormatException;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.util.UriComponentsBuilder;
/**
* @author Vinicius Carvalho
*/
@RestController
@RequestMapping(path = "${spring.cloud.stream.schema.server.prefix:}")
public class ServerController {
private final SchemaRepository repository;
private final Map<String, SchemaValidator> validators;
public ServerController(SchemaRepository repository,
Map<String, SchemaValidator> validators) {
Assert.notNull(repository, "cannot be null");
Assert.notEmpty(validators, "cannot be empty");
this.repository = repository;
this.validators = validators;
}
@RequestMapping(method = RequestMethod.POST, path = "/", consumes = "application/json", produces = "application/json")
public synchronized ResponseEntity<Schema> register(@RequestBody Schema schema,
UriComponentsBuilder builder) {
SchemaValidator validator = this.validators.get(schema.getFormat());
if (validator == null) {
throw new UnsupportedFormatException(String.format(
"Invalid format, supported types are: %s",
StringUtils.collectionToCommaDelimitedString(this.validators.keySet())));
}
if (!validator.isValid(schema.getDefinition())) {
throw new InvalidSchemaException("Invalid schema");
}
Schema result;
List<Schema> registeredEntities = this.repository.findBySubjectAndFormatOrderByVersion(
schema.getSubject(), schema.getFormat());
if (registeredEntities == null || registeredEntities.size() == 0) {
schema.setVersion(1);
result = this.repository.save(schema);
}
else {
result = validator.match(registeredEntities, schema.getDefinition());
if (result == null) {
schema.setVersion(
registeredEntities.get(registeredEntities.size() - 1).getVersion()
+ 1);
result = this.repository.save(schema);
}
}
HttpHeaders headers = new HttpHeaders();
headers.add(HttpHeaders.LOCATION,
builder.path("/{subject}/{format}/v{version}")
.buildAndExpand(result.getSubject(), result.getFormat(),
result.getVersion())
.toString());
ResponseEntity<Schema> response = new ResponseEntity<>(result, headers,
HttpStatus.CREATED);
return response;
}
@RequestMapping(method = RequestMethod.GET, produces = "application/json", path = "/{subject}/{format}/v{version}")
public ResponseEntity<Schema> findOne(@PathVariable("subject") String subject,
@PathVariable("format") String format,
@PathVariable("version") Integer version) {
Schema schema = this.repository.findOneBySubjectAndFormatAndVersion(subject, format,
version);
if (schema == null) {
throw new SchemaNotFoundException("Could not find Schema");
}
return new ResponseEntity<>(schema, HttpStatus.OK);
}
@RequestMapping(method = RequestMethod.GET, produces = "application/json", path = "/schemas/{id}")
public ResponseEntity<Schema> findOne(@PathVariable("id") Integer id) {
Schema schema = this.repository.findOne(id);
if (schema == null) {
throw new SchemaNotFoundException("Could not find Schema");
}
return new ResponseEntity<>(schema, HttpStatus.OK);
}
@ExceptionHandler(UnsupportedFormatException.class)
@ResponseStatus(value = HttpStatus.BAD_REQUEST, reason = "Format not supported")
public void unsupportedFormat(UnsupportedFormatException ex) {
}
@ExceptionHandler(InvalidSchemaException.class)
@ResponseStatus(value = HttpStatus.BAD_REQUEST, reason = "Invalid schema")
public void invalidSchema(InvalidSchemaException ex) {
}
@ExceptionHandler(SchemaNotFoundException.class)
@ResponseStatus(value = HttpStatus.NOT_FOUND, reason = "Schema not found")
public void schemaNotFound(SchemaNotFoundException ex) {
}
}

25
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/model/Compatibility.java Normal file
View File

@@ -0,0 +1,25 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.model;
/**
* @author Vinicius Carvalho
*/
public enum Compatibility {
BACKWARD, FORWARD, FULL, INCOMPATIBLE;
}

90
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/model/Schema.java Normal file
View File

@@ -0,0 +1,90 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.model;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.Id;
import javax.persistence.Lob;
/**
* @author Vinicius Carvalho
*
* Represents a persisted schema entity.
*/
@Entity
public class Schema {
@Id
@GeneratedValue
@Column(name = "ID")
private Integer id;
@Column(name = "VERSION", nullable = false)
private Integer version;
@Column(name = "SUBJECT", nullable = false)
private String subject;
@Column(name = "FORMAT", nullable = false)
private String format;
@Lob
@Column(name = "DEFINITION", nullable = false, length = 8192)
private String definition;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public Integer getVersion() {
return version;
}
public void setVersion(Integer version) {
this.version = version;
}
public String getSubject() {
return subject;
}
public void setSubject(String subject) {
this.subject = subject;
}
public String getFormat() {
return format;
}
public void setFormat(String format) {
this.format = format;
}
public String getDefinition() {
return definition;
}
public void setDefinition(String definition) {
this.definition = definition;
}
}

34
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/repository/SchemaRepository.java Normal file
View File

@@ -0,0 +1,34 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.repository;
import java.util.List;
import org.springframework.cloud.stream.schema.server.model.Schema;
import org.springframework.data.repository.PagingAndSortingRepository;
/**
* @author Vinicius Carvalho
*/
public interface SchemaRepository extends PagingAndSortingRepository<Schema, Integer> {
List<Schema> findBySubjectAndFormatOrderByVersion(String subject,
String format);
Schema findOneBySubjectAndFormatAndVersion(String subject, String format,
Integer version);
}

68
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/AvroSchemaValidator.java Normal file
View File

@@ -0,0 +1,68 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.support;
import java.util.List;
import org.apache.avro.SchemaParseException;
import org.springframework.cloud.stream.schema.server.model.Compatibility;
import org.springframework.cloud.stream.schema.server.model.Schema;
/**
* @author Vinicius Carvalho
*/
public class AvroSchemaValidator implements SchemaValidator {
@Override
public boolean isValid(String definition) {
boolean result = true;
try {
new org.apache.avro.Schema.Parser().parse(definition);
}
catch (SchemaParseException ex) {
result = false;
}
return result;
}
@Override
public Compatibility compatibilityCheck(String source, String other) {
return null;
}
@Override
public Schema match(List<Schema> schemas, String definition) {
Schema result = null;
org.apache.avro.Schema source = new org.apache.avro.Schema.Parser()
.parse(definition);
for (Schema s : schemas) {
org.apache.avro.Schema target = new org.apache.avro.Schema.Parser()
.parse(s.getDefinition());
if (target.equals(source)) {
result = s;
break;
}
}
return result;
}
@Override
public String getFormat() {
return "avro";
}
}

26
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/InvalidSchemaException.java Normal file
View File

@@ -0,0 +1,26 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.support;
/**
* @author Vinicius Carvalho
*/
public class InvalidSchemaException extends RuntimeException {
public InvalidSchemaException(String message) {
super(message);
}
}

26
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/SchemaNotFoundException.java Normal file
View File

@@ -0,0 +1,26 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.support;
/**
* @author Vinicius Carvalho
*/
public class SchemaNotFoundException extends RuntimeException {
public SchemaNotFoundException(String message) {
super(message);
}
}

58
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/SchemaValidator.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.support;
import java.util.List;
import org.springframework.cloud.stream.schema.server.model.Compatibility;
import org.springframework.cloud.stream.schema.server.model.Schema;
/**
* @author Vinicius Carvalho
*
* Provides utility methods to validate, check compatibility and match schemas of
* different implementations
*/
public interface SchemaValidator {
/**
* Verifies if a definition is a valid schema
* @param definition - The textual representation of the schema file
* @return
*/
boolean isValid(String definition);
/**
* Checks for compatibility between two schemas @see Compatibility class for types
* This method may not be supported for certain formats
* @param source - The textual representation of the schema to tested
* @param other - The textual representation of the other schema to tested
* @return
*/
Compatibility compatibilityCheck(String source, String other);
/**
* Return the Schema that is represented by the definition.
* @param schemas List of schemas to be tested
* @param definition Textual representation of the schema
* @return A full Schema object with identifier and subject properties
*/
Schema match(List<Schema> schemas, String definition);
String getFormat();
}

27
spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/UnsupportedFormatException.java Normal file
View File

@@ -0,0 +1,27 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server.support;
/**
* @author Vinicius Carvalho
*/
public class UnsupportedFormatException extends RuntimeException {
public UnsupportedFormatException(String message) {
super(message);
}
}

0
spring-cloud-stream-schema-server/src/main/resources/META-INF/spring.factories Normal file
View File

5
spring-cloud-stream-schema-server/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,5 @@
spring:
application:
name: SchemaRegistryServer
server:
port: 8990

160
spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerAvroTests.java Normal file
View File

@@ -0,0 +1,160 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.server;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.cloud.stream.schema.server.model.Schema;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @author Vinicius Carvalho
*/
@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)
public class SchemaRegistryServerAvroTests {
final String USER_SCHEMA_V1 = "{\"namespace\": \"example.avro\",\n"
+ " \"type\": \"record\",\n" + " \"name\": \"User\",\n" + " \"fields\": [\n"
+ " {\"name\": \"name\", \"type\": \"string\"},\n"
+ " {\"name\": \"favorite_number\", \"type\": [\"int\", \"null\"]}\n"
+ " ]\n" + "}";
final String USER_SCHEMA_V2 = "{\"namespace\": \"example.avro\",\n"
+ " \"type\": \"record\",\n" + " \"name\": \"User\",\n" + " \"fields\": [\n"
+ " {\"name\": \"name\", \"type\": \"string\"},\n"
+ " {\"name\": \"favorite_number\", \"type\": [\"int\", \"null\"]},\n"
+ " {\"name\": \"favorite_color\", \"type\": [\"string\", \"null\"]}\n"
+ " ]\n" + "}";
@Autowired
private TestRestTemplate client;
@Test
public void testUnsupportedFormat() throws Exception {
Schema schema = new Schema();
schema.setFormat("spring");
schema.setSubject("boot");
ResponseEntity<Schema> response = client.postForEntity("http://localhost:8990/",
schema, Schema.class);
Assert.assertEquals(HttpStatus.BAD_REQUEST, response.getStatusCode());
}
@Test
public void testInvalidSchema() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("boot");
schema.setDefinition("{}");
ResponseEntity<Schema> response = client.postForEntity("http://localhost:8990/",
schema, Schema.class);
Assert.assertEquals(HttpStatus.BAD_REQUEST, response.getStatusCode());
}
@Test
public void testUserSchemaV1() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("org.springframework.cloud.stream.schema.User");
schema.setDefinition(USER_SCHEMA_V1);
ResponseEntity<Schema> response = client.postForEntity("http://localhost:8990/",
schema, Schema.class);
Assert.assertTrue(response.getStatusCode().is2xxSuccessful());
Assert.assertEquals(new Integer(1), response.getBody().getVersion());
List<String> location = response.getHeaders().get(HttpHeaders.LOCATION);
Assert.assertNotNull(location);
ResponseEntity<Schema> persistedSchema = client.getForEntity(location.get(0),
Schema.class);
Assert.assertEquals(response.getBody().getId(),
persistedSchema.getBody().getId());
}
@Test
public void testUserSchemaV2() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("org.springframework.cloud.stream.schema.User");
schema.setDefinition(USER_SCHEMA_V1);
Schema schema2 = new Schema();
schema2.setFormat("avro");
schema2.setSubject("org.springframework.cloud.stream.schema.User");
schema2.setDefinition(USER_SCHEMA_V2);
ResponseEntity<Schema> response = client.postForEntity("http://localhost:8990/",
schema, Schema.class);
Assert.assertTrue(response.getStatusCode().is2xxSuccessful());
Assert.assertEquals(new Integer(1), response.getBody().getVersion());
List<String> location = response.getHeaders().get(HttpHeaders.LOCATION);
Assert.assertNotNull(location);
ResponseEntity<Schema> response2 = client.postForEntity("http://localhost:8990/",
schema2, Schema.class);
Assert.assertTrue(response.getStatusCode().is2xxSuccessful());
Assert.assertEquals(new Integer(2), response2.getBody().getVersion());
List<String> location2 = response2.getHeaders().get(HttpHeaders.LOCATION);
Assert.assertNotNull(location2);
}
@Test
public void testIdempotentRegistration() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("org.springframework.cloud.stream.schema.User");
schema.setDefinition(USER_SCHEMA_V1);
ResponseEntity<Schema> response = client.postForEntity("http://localhost:8990/",
schema, Schema.class);
Assert.assertTrue(response.getStatusCode().is2xxSuccessful());
Assert.assertEquals(new Integer(1), response.getBody().getVersion());
List<String> location = response.getHeaders().get(HttpHeaders.LOCATION);
Assert.assertNotNull(location);
ResponseEntity<Schema> response2 = client.postForEntity("http://localhost:8990/",
schema, Schema.class);
Assert.assertEquals(response.getBody().getId(), response2.getBody().getId());
}
@Test
public void testSchemaNotfound() throws Exception {
ResponseEntity<Schema> response = client
.getForEntity("http://localhost:8990/foo/avro/v42", Schema.class);
Assert.assertEquals(HttpStatus.NOT_FOUND, response.getStatusCode());
}
@TestConfiguration
static class Config {
@Bean
public TestRestTemplate testRestTemplate() {
return new TestRestTemplate();
}
}
}

48
spring-cloud-stream-schema/pom.xml Normal file
View File

@@ -0,0 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>spring-cloud-stream-parent</artifactId>
<groupId>org.springframework.cloud</groupId>
<version>1.1.0.BUILD-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-schema</artifactId>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>1.8.1</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-schema-server</artifactId>
<version>1.1.0.BUILD-SNAPSHOT</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

27
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaNotFoundException.java Normal file
View File

@@ -0,0 +1,27 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema;
/**
* @author Vinicius Carvalho
*/
public class SchemaNotFoundException extends RuntimeException {
public SchemaNotFoundException(String message) {
super(message);
}
}

106
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaReference.java Normal file
View File

@@ -0,0 +1,106 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema;
import org.springframework.util.Assert;
/**
* References a schema through its subject and version.
* @author Marius Bogoevici
*/
public class SchemaReference {
private String subject;
private int version;
private String format;
public SchemaReference(String subject, int version, String format) {
Assert.hasText(subject, "cannot be empty");
Assert.isTrue(version > 0, "must be a positive integer");
Assert.hasText(format, "cannot be empty");
this.subject = subject;
this.version = version;
this.format = format;
}
public String getSubject() {
return this.subject;
}
public void setSubject(String subject) {
Assert.hasText(subject, "cannot be empty");
this.subject = subject;
}
public int getVersion() {
return this.version;
}
public void setVersion(int version) {
Assert.isTrue(version > 0, "must be a positive integer");
this.version = version;
}
public String getFormat() {
return this.format;
}
public void setFormat(String format) {
Assert.hasText(format, "cannot be empty");
this.format = format;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
SchemaReference that = (SchemaReference) o;
if (this.version != that.version) {
return false;
}
if (!this.subject.equals(that.subject)) {
return false;
}
return this.format.equals(that.format);
}
@Override
public int hashCode() {
int result = this.subject.hashCode();
result = 31 * result + this.version;
result = 31 * result + this.format.hashCode();
return result;
}
@Override
public String toString() {
return "SchemaReference{" +
"subject='" + this.subject + '\'' +
", version=" + this.version +
", format='" + this.format + '\'' +
'}';
}
}

43
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaRegistrationResponse.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema;
/**
* @author Marius Bogoevici
*/
public class SchemaRegistrationResponse {
private long id;
private SchemaReference schemaReference;
public long getId() {
return this.id;
}
public void setId(long id) {
this.id = id;
}
public SchemaReference getSchemaReference() {
return this.schemaReference;
}
public void setSchemaReference(SchemaReference schemaReference) {
this.schemaReference = schemaReference;
}
}

192
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java Normal file
View File

@@ -0,0 +1,192 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.avro;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecord;
import org.springframework.core.io.Resource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.util.MimeType;
/**
* Base class for Apache Avro {@link org.springframework.messaging.converter.MessageConverter} implementations.
* @author Marius Bogoevici
*/
public abstract class AbstractAvroMessageConverter extends AbstractMessageConverter {
protected AbstractAvroMessageConverter(MimeType supportedMimeType) {
super(supportedMimeType);
}
protected AbstractAvroMessageConverter(Collection<MimeType> supportedMimeTypes) {
super(supportedMimeTypes);
}
protected static Schema parseSchema(Resource r) throws IOException {
return new Schema.Parser().parse(r.getInputStream());
}
@Override
protected boolean canConvertFrom(Message<?> message, Class<?> targetClass) {
return super.canConvertFrom(message, targetClass) && (message.getPayload() instanceof byte[]);
}
@Override
protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
Object result = null;
try {
byte[] payload = (byte[]) message.getPayload();
ByteBuffer buf = ByteBuffer.wrap(payload);
MimeType mimeType = getContentTypeResolver().resolve(message.getHeaders());
if (mimeType == null) {
if (conversionHint instanceof MimeType) {
mimeType = (MimeType) conversionHint;
}
else {
return null;
}
}
buf.get(payload);
Schema writerSchema = resolveWriterSchemaForDeserialization(mimeType);
Schema readerSchema = resolveReaderSchemaForDeserialization(targetClass);
DatumReader<Object> reader = getDatumReader((Class<Object>) targetClass, readerSchema, writerSchema);
Decoder decoder = DecoderFactory.get().binaryDecoder(payload, null);
result = reader.read(null, decoder);
}
catch (IOException e) {
throw new MessageConversionException(message, "Failed to read payload", e);
}
return result;
}
private DatumWriter<Object> getDatumWriter(Class<Object> type, Schema schema) {
DatumWriter<Object> writer;
this.logger.debug("Finding correct DatumWriter for type " + type.getName());
if (SpecificRecord.class.isAssignableFrom(type)) {
if (schema != null) {
writer = new SpecificDatumWriter<>(schema);
}
else {
writer = new SpecificDatumWriter<>(type);
}
}
else if (GenericRecord.class.isAssignableFrom(type)) {
writer = new GenericDatumWriter<>(schema);
}
else {
if (schema != null) {
writer = new ReflectDatumWriter<>(schema);
}
else {
writer = new ReflectDatumWriter<>(type);
}
}
return writer;
}
protected DatumReader<Object> getDatumReader(Class<Object> type, Schema schema, Schema writerSchema) {
DatumReader<Object> reader = null;
if (SpecificRecord.class.isAssignableFrom(type)) {
if (schema != null) {
if (writerSchema != null) {
reader = new SpecificDatumReader<>(writerSchema, schema);
}
else {
reader = new SpecificDatumReader<>(schema);
}
}
else {
reader = new SpecificDatumReader<>(type);
if (writerSchema != null) {
reader.setSchema(writerSchema);
}
}
}
else if (GenericRecord.class.isAssignableFrom(type)) {
if (schema != null) {
if (writerSchema != null) {
reader = new GenericDatumReader<>(writerSchema, schema);
}
else {
reader = new GenericDatumReader<>(schema);
}
}
}
else {
reader = new ReflectDatumReader(type);
if (writerSchema != null) {
reader.setSchema(writerSchema);
}
}
if (reader == null) {
throw new MessageConversionException(
"No schema can be inferred from type " + type
.getName() + " and no schema has been explicitly configured.");
}
return reader;
}
@Override
protected Object convertToInternal(Object payload, MessageHeaders headers, Object conversionHint) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try {
MimeType hintedContentType = null;
if (conversionHint instanceof MimeType) {
hintedContentType = (MimeType) conversionHint;
}
Schema schema = resolveSchemaForWriting(payload, headers, hintedContentType);
DatumWriter<Object> writer = getDatumWriter((Class<Object>) payload.getClass(), schema);
Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null);
writer.write(payload, encoder);
encoder.flush();
}
catch (IOException e) {
throw new MessageConversionException("Failed to write payload", e);
}
return baos.toByteArray();
}
protected abstract Schema resolveSchemaForWriting(Object payload, MessageHeaders headers,
MimeType hintedContentType);
protected abstract Schema resolveWriterSchemaForDeserialization(MimeType mimeType);
protected abstract Schema resolveReaderSchemaForDeserialization(Class<?> targetClass);
}

64
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java Normal file
View File

@@ -0,0 +1,64 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.avro;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.StringConvertingContentTypeResolver;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.ObjectUtils;
/**
* @author Marius Bogoevici
* @author Vinicius Carvalho
*/
@Configuration
@ConditionalOnClass(name = "org.apache.avro.Schema")
@ConditionalOnProperty(value = "spring.cloud.stream.schemaRegistryClient.enabled", matchIfMissing = true)
@ConditionalOnBean(type = "org.springframework.cloud.stream.schema.client.SchemaRegistryClient")
@EnableConfigurationProperties(AvroMessageConverterProperties.class)
public class AvroMessageConverterAutoConfiguration {
@Autowired
private AvroMessageConverterProperties avroMessageConverterProperties;
@Bean
public AvroSchemaRegistryClientMessageConverter avroSchemaMessageConverter(
SchemaRegistryClient schemaRegistryClient) {
AvroSchemaRegistryClientMessageConverter
avroSchemaRegistryClientMessageConverter = new AvroSchemaRegistryClientMessageConverter(
schemaRegistryClient);
avroSchemaRegistryClientMessageConverter.setDynamicSchemaGenerationEnabled(
this.avroMessageConverterProperties.isDynamicSchemaGenerationEnabled());
avroSchemaRegistryClientMessageConverter.setContentTypeResolver(new StringConvertingContentTypeResolver());
if (this.avroMessageConverterProperties.getReaderSchema() != null) {
avroSchemaRegistryClientMessageConverter.setReaderSchema(
this.avroMessageConverterProperties.getReaderSchema());
}
if (!ObjectUtils.isEmpty(this.avroMessageConverterProperties.getSchemaLocations())) {
avroSchemaRegistryClientMessageConverter.setSchemaLocations(
this.avroMessageConverterProperties.getSchemaLocations());
}
avroSchemaRegistryClientMessageConverter.setPrefix(this.avroMessageConverterProperties.getPrefix());
return avroSchemaRegistryClientMessageConverter;
}
}

70
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterProperties.java Normal file
View File

@@ -0,0 +1,70 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.avro;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;
/**
* @author Vinicius Carvalho
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.schema.avro")
public class AvroMessageConverterProperties {
private boolean dynamicSchemaGenerationEnabled;
private Resource readerSchema;
private Resource[] schemaLocations;
private String prefix = "vnd";
public Resource getReaderSchema() {
return this.readerSchema;
}
public void setReaderSchema(Resource readerSchema) {
Assert.notNull(readerSchema, "cannot be null");
this.readerSchema = readerSchema;
}
public Resource[] getSchemaLocations() {
return this.schemaLocations;
}
public void setSchemaLocations(Resource[] schemaLocations) {
Assert.notEmpty(schemaLocations, "cannot be null");
this.schemaLocations = schemaLocations;
}
public boolean isDynamicSchemaGenerationEnabled() {
return this.dynamicSchemaGenerationEnabled;
}
public void setDynamicSchemaGenerationEnabled(boolean dynamicSchemaGenerationEnabled) {
this.dynamicSchemaGenerationEnabled = dynamicSchemaGenerationEnabled;
}
public String getPrefix() {
return this.prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
}

117
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.avro;
import java.io.IOException;
import java.util.Collection;
import org.apache.avro.Schema;
import org.springframework.core.io.Resource;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
/**
* A {@link org.springframework.messaging.converter.MessageConverter}
* using Apache Avro.
* The schema for serializing and deserializing will be automatically inferred
* from the class for {@link org.apache.avro.specific.SpecificRecord} and regular
* classes, unless a specific schema is set, case in which that schema will be used
* instead.
* For converting to {@link org.apache.avro.generic.GenericRecord} targets,
* a schema must be set.s
* @author Marius Bogoevici
*/
public class AvroSchemaMessageConverter extends AbstractAvroMessageConverter {
private Schema schema;
/**
* Create a {@link AvroSchemaMessageConverter}.
* Uses the default {@link MimeType} of {@code "application/avro"}.
*/
public AvroSchemaMessageConverter() {
super(new MimeType("application", "avro"));
}
/**
* Create a {@link AvroSchemaMessageConverter}.
* The converter will be used for the provided {@link MimeType}.
*/
public AvroSchemaMessageConverter(MimeType supportedMimeType) {
super(supportedMimeType);
}
/**
* Create a {@link AvroSchemaMessageConverter}.
* The converter will be used for the provided {@link MimeType}s.
* @param supportedMimeTypes the mime types supported by this converter
*/
public AvroSchemaMessageConverter(Collection<MimeType> supportedMimeTypes) {
super(supportedMimeTypes);
}
public Schema getSchema() {
return this.schema;
}
/**
* Sets the Apache Avro schema to be used by this converter.
* @param schema schema to be used by this converter
*/
public void setSchema(Schema schema) {
Assert.notNull(schema, "schema cannot be null");
this.schema = schema;
}
/**
* The location of the Apache Avro schema to be used by this converter.
* @param schemaLocation the location of the schema used by this converter.
*/
public void setSchemaLocation(Resource schemaLocation) {
Assert.notNull(schemaLocation, "schema cannot be null");
try {
this.schema = parseSchema(schemaLocation);
}
catch (IOException e) {
throw new IllegalStateException("Schema cannot be parsed:", e);
}
}
@Override
protected boolean supports(Class<?> clazz) {
return true;
}
@Override
protected Schema resolveWriterSchemaForDeserialization(MimeType mimeType) {
return this.schema;
}
@Override
protected Schema resolveReaderSchemaForDeserialization(Class<?> targetClass) {
return this.schema;
}
@Override
protected Schema resolveSchemaForWriting(Object payload, MessageHeaders headers,
MimeType hintedContentType) {
return this.schema;
}
}

274
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java Normal file
View File

@@ -0,0 +1,274 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.avro;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericContainer;
import org.apache.avro.reflect.ReflectData;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cloud.stream.schema.SchemaNotFoundException;
import org.springframework.cloud.stream.schema.SchemaReference;
import org.springframework.cloud.stream.schema.SchemaRegistrationResponse;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
import org.springframework.core.io.Resource;
import org.springframework.integration.support.MutableMessageHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.ObjectUtils;
/**
* A {@link org.springframework.messaging.converter.MessageConverter}
* for Apache Avro, with the ability to publish and retrieve schemas
* stored in a schema server, allowing for schema evolution in applications.
* The supported content types are in the form `application/*+avro`.
*
* During the conversion to a message, the converter will set the 'contentType'
* header to 'application/[prefix].[subject].v[version]+avro', where:
*
* <li>
* <ul><i>prefix</i> is a configurable prefix (default 'vnd');</ul>
* <ul><i>subject</i> is a subject derived from the type of the outgoing object - typically the class name;</ul>
* <ul><i>version</i> is the schema version for the given subject;</ul>
* </li>
*
* When converting from a message, the converter will parse the content-type
* and use it to fetch and cache the writer schema using the provided
* {@link SchemaRegistryClient}.
* @author Marius Bogoevici
* @author Vinicius Carvalho
*/
public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessageConverter implements InitializingBean {
public static final String AVRO_FORMAT = "avro";
public static final Pattern PREFIX_VALIDATION_PATTERN = Pattern.compile("[\\p{Alnum}]");
private Pattern versionedSchema;
private boolean dynamicSchemaGenerationEnabled;
private Map<String, Schema> localSchemaMap = new HashMap<>();
private Schema readerSchema;
private Resource[] schemaLocations;
private SchemaRegistryClient schemaRegistryClient;
private String prefix = "vnd";
/**
* Creates a new instance, configuring it with a {@link SchemaRegistryClient}.
* @param schemaRegistryClient the {@link SchemaRegistryClient} used to interact with the schema registry server.
*/
public AvroSchemaRegistryClientMessageConverter(SchemaRegistryClient schemaRegistryClient) {
super(Arrays.asList(new MimeType("application", "*+avro")));
Assert.notNull(schemaRegistryClient, "cannot be null");
this.schemaRegistryClient = schemaRegistryClient;
}
/**
* Allows the converter to generate and register schemas automatically.
* If set to false, it only allows the converter to use pre-registered schemas.
* Default 'true'.
* @param dynamicSchemaGenerationEnabled true if dynamic schema generation is enabled
*/
public void setDynamicSchemaGenerationEnabled(boolean dynamicSchemaGenerationEnabled) {
this.dynamicSchemaGenerationEnabled = dynamicSchemaGenerationEnabled;
}
public boolean isDynamicSchemaGenerationEnabled() {
return this.dynamicSchemaGenerationEnabled;
}
/**
* A set of locations where the converter can load schemas from.
* Schemas provided at these locations will be registered automatically.
*
* @param schemaLocations
*/
public void setSchemaLocations(Resource[] schemaLocations) {
Assert.notEmpty(schemaLocations, "cannot be empty");
this.schemaLocations = schemaLocations;
}
/**
* Set the prefix to be used in the publised subtype. Default 'vnd'.
* @param prefix
*/
public void setPrefix(String prefix) {
Assert.hasText(prefix, "Prefix cannot be empty");
Assert.isTrue(!PREFIX_VALIDATION_PATTERN.matcher(this.prefix).matches(), "Invalid prefix:" + this.prefix);
this.prefix = prefix;
}
@Override
public void afterPropertiesSet() throws Exception {
this.versionedSchema = Pattern.compile(
"application/" + this.prefix + "\\.([\\p{Alnum}\\$\\.]+)\\.v(\\p{Digit}+)\\+avro");
if (!ObjectUtils.isEmpty(this.schemaLocations)) {
this.logger.info("Scanning avro schema resources on classpath");
if (this.logger.isInfoEnabled()) {
this.logger.info("Parsing" + this.schemaLocations.length);
}
for (Resource schemaLocation : this.schemaLocations) {
try {
Schema schema = parseSchema(schemaLocation);
if (this.logger.isInfoEnabled()) {
this.logger.info("Resource " + schemaLocation.getFilename() + " parsed into schema " + schema
.getNamespace() + "." + schema.getName());
}
this.schemaRegistryClient.register(toSubject(schema), AVRO_FORMAT, schema.toString(true));
if (this.logger.isInfoEnabled()) {
this.logger.info("Schema " + schema.getName() + " registered with id " + schema);
}
this.localSchemaMap.put(schema.getNamespace() + "." + schema.getName(), schema);
}
catch (IOException e) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("Failed to parse schema at " + schemaLocation.getFilename(), e);
}
}
}
}
}
protected String toSubject(Schema schema) {
return schema.getName().toLowerCase();
}
@Override
protected boolean supports(Class<?> clazz) {
// we support all types
return true;
}
@Override
protected boolean supportsMimeType(MessageHeaders headers) {
if (super.supportsMimeType(headers)) {
return true;
}
MimeType mimeType = getContentTypeResolver().resolve(headers);
return MimeType.valueOf("application/*+avro").includes(mimeType);
}
@Override
protected Schema resolveSchemaForWriting(Object payload, MessageHeaders headers,
MimeType hintedContentType) {
Schema schema;
SchemaReference schemaReference = extractSchemaReference(hintedContentType);
// the mimeType does not contain a schema reference
if (schemaReference == null) {
schema = extractSchemaForWriting(payload);
SchemaRegistrationResponse schemaRegistrationResponse = this.schemaRegistryClient.register(
toSubject(schema), AVRO_FORMAT, schema.toString(true));
schemaReference = schemaRegistrationResponse.getSchemaReference();
}
else {
Schema.Parser parser = new Schema.Parser();
String schemaContents = this.schemaRegistryClient.fetch(schemaReference);
schema = parser.parse(schemaContents);
}
if (headers instanceof MutableMessageHeaders) {
headers.put(MessageHeaders.CONTENT_TYPE,
"application/vnd." + schemaReference.getSubject() + ".v" + schemaReference
.getVersion() + "+avro");
}
return schema;
}
private SchemaReference extractSchemaReference(MimeType mimeType) {
SchemaReference schemaReference = null;
Matcher schemaMatcher = this.versionedSchema.matcher(mimeType.toString());
if (schemaMatcher.find()) {
String subject = schemaMatcher.group(1);
Integer version = Integer.parseInt(schemaMatcher.group(2));
schemaReference = new SchemaReference(subject, version, AVRO_FORMAT);
}
return schemaReference;
}
@Override
protected Schema resolveWriterSchemaForDeserialization(MimeType mimeType) {
if (this.readerSchema == null) {
Schema schema = null;
SchemaReference schemaReference = extractSchemaReference(mimeType);
if (schemaReference != null) {
String schemaContent = this.schemaRegistryClient.fetch(schemaReference);
schema = new Schema.Parser().parse(schemaContent);
}
return schema;
}
else {
return this.readerSchema;
}
}
@Override
protected Schema resolveReaderSchemaForDeserialization(Class<?> targetClass) {
return this.readerSchema;
}
public void setReaderSchema(Resource readerSchema) {
Assert.notNull(readerSchema, "cannot be null");
try {
this.readerSchema = parseSchema(readerSchema);
}
catch (IOException e) {
throw new BeanInitializationException("Cannot initialize reader schema", e);
}
}
private Schema extractSchemaForWriting(Object payload) {
Schema schema = null;
if (this.logger.isDebugEnabled()) {
this.logger.debug("Obtaining schema for class " + payload.getClass());
}
if (GenericContainer.class.isAssignableFrom(payload.getClass())) {
schema = ((GenericContainer) payload).getSchema();
if (this.logger.isDebugEnabled()) {
this.logger.debug("Avro type detected, using schema from object");
}
}
else {
schema = this.localSchemaMap.get(payload.getClass().getName());
if (schema == null) {
if (!isDynamicSchemaGenerationEnabled()) {
throw new SchemaNotFoundException(
String.format("No schema found in the local cache for %s, and dynamic schema generation " +
"is not enabled", payload.getClass()));
}
else {
schema = ReflectData.get().getSchema(payload.getClass());
this.schemaRegistryClient.register(toSubject(schema), AVRO_FORMAT, schema.toString(true));
}
this.localSchemaMap.put(payload.getClass().getName(), schema);
}
}
return schema;
}
}

109
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java Normal file
View File

@@ -0,0 +1,109 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.client;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cloud.stream.schema.SchemaReference;
import org.springframework.cloud.stream.schema.SchemaRegistrationResponse;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.util.Assert;
import org.springframework.web.client.RestTemplate;
/**
* @author Vinicius Carvalho
* @author Marius Bogoevici
*/
public class ConfluentSchemaRegistryClient implements SchemaRegistryClient {
private RestTemplate template;
private String endpoint = "http://localhost:8081";
private ObjectMapper mapper;
public ConfluentSchemaRegistryClient() {
this.template = new RestTemplate();
this.mapper = new ObjectMapper();
}
public void setEndpoint(String endpoint) {
this.endpoint = endpoint;
}
@Override
public SchemaRegistrationResponse register(String subject, String format, String schema) {
Assert.isTrue("avro".equals(format), "Only Avro is supported");
String path = String.format("/subjects/%s/versions", subject);
HttpHeaders headers = new HttpHeaders();
headers.put("Accept",
Arrays.asList("application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json",
"application/json"));
headers.add("Content-Type", "application/json");
Integer id = null;
try {
String payload = this.mapper.writeValueAsString(Collections.singletonMap("schema", schema));
HttpEntity<String> request = new HttpEntity<>(payload, headers);
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.POST, request,
Map.class);
id = (Integer) response.getBody().get("id");
}
catch (JsonProcessingException e) {
e.printStackTrace();
}
SchemaRegistrationResponse schemaRegistrationResponse = new SchemaRegistrationResponse();
schemaRegistrationResponse.setId(id);
schemaRegistrationResponse.setSchemaReference(new SchemaReference(subject, id, "avro"));
return schemaRegistrationResponse;
}
@Override
public String fetch(SchemaReference schemaReference) {
String path = String.format("/schemas/ids/%d", schemaReference.getVersion());
HttpHeaders headers = new HttpHeaders();
headers.put("Accept",
Arrays.asList("application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json",
"application/json"));
headers.add("Content-Type", "application/vnd.schemaregistry.v1+json");
HttpEntity<String> request = new HttpEntity<>("", headers);
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request, Map
.class);
return (String) response.getBody().get("schema");
}
@Override
public String fetch(Integer id) {
String path = String.format("/schemas/ids/%d", id);
HttpHeaders headers = new HttpHeaders();
headers.put("Accept",
Arrays.asList("application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json",
"application/json"));
headers.add("Content-Type", "application/vnd.schemaregistry.v1+json");
HttpEntity<String> request = new HttpEntity<>("", headers);
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request, Map
.class);
return (String) response.getBody().get("schema");
}
}

87
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java Normal file
View File

@@ -0,0 +1,87 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.client;
import java.util.HashMap;
import java.util.Map;
import org.springframework.cloud.stream.schema.SchemaReference;
import org.springframework.cloud.stream.schema.SchemaRegistrationResponse;
import org.springframework.http.ResponseEntity;
import org.springframework.util.Assert;
import org.springframework.web.client.RestTemplate;
/**
* @author Marius Bogoevici
*/
public class DefaultSchemaRegistryClient implements SchemaRegistryClient {
private RestTemplate template;
private String endpoint = "http://localhost:8990";
public DefaultSchemaRegistryClient() {
this.template = new RestTemplate();
}
public void setEndpoint(String endpoint) {
Assert.hasText(endpoint, "cannot be empty");
this.endpoint = endpoint;
}
@Override
public SchemaRegistrationResponse register(String subject, String format, String schema) {
Map<String, String> requestBody = new HashMap<>();
requestBody.put("subject", subject);
requestBody.put("format", format);
requestBody.put("definition", schema);
ResponseEntity<Map> responseEntity = this.template.postForEntity(this.endpoint, requestBody, Map.class);
if (responseEntity.getStatusCode().is2xxSuccessful()) {
SchemaRegistrationResponse registrationResponse = new SchemaRegistrationResponse();
Map<String, Object> responseBody = (Map<String, Object>) responseEntity.getBody();
registrationResponse.setId((Integer) responseBody.get("id"));
registrationResponse.setSchemaReference(
new SchemaReference(subject, (Integer) responseBody.get("version"),
responseBody.get("format").toString()));
return registrationResponse;
}
throw new RuntimeException("Failed to register schema: " + responseEntity.toString());
}
@Override
public String fetch(SchemaReference schemaReference) {
ResponseEntity<Map> responseEntity = this.template.getForEntity(
this.endpoint + "/" + schemaReference.getSubject() + "/" + schemaReference
.getFormat() + "/v" + schemaReference
.getVersion(), Map.class);
if (!responseEntity.getStatusCode().is2xxSuccessful()) {
throw new RuntimeException("Failed to fetch schema: " + responseEntity.toString());
}
return (String) responseEntity.getBody().get("definition");
}
@Override
public String fetch(Integer id) {
ResponseEntity<Map> responseEntity = this.template.getForEntity(
this.endpoint + "/schemas/" + id, Map.class);
if (!responseEntity.getStatusCode().is2xxSuccessful()) {
throw new RuntimeException("Failed to fetch schema: " + responseEntity.toString());
}
return (String) responseEntity.getBody().get("definition");
}
}

41
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/EnableSchemaRegistryClient.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.client;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.cloud.stream.schema.client.config.SchemaRegistryClientConfiguration;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
/**
* @author Marius Bogoevici
*/
@Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Configuration
@Import(SchemaRegistryClientConfiguration.class)
public @interface EnableSchemaRegistryClient {
}

50
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/SchemaRegistryClient.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.client;
import org.springframework.cloud.stream.schema.SchemaReference;
import org.springframework.cloud.stream.schema.SchemaRegistrationResponse;
/**
* @author Vinicius Carvalho
* @author Marius Bogoevici
*/
public interface SchemaRegistryClient {
/**
* Registers a schema with the remote repository returning the unique identifier associated with this schema.
* @param subject the full name of the schema
* @param schema
* @return a {@link SchemaRegistrationResponse} representing the result of the operation
*/
SchemaRegistrationResponse register(String subject, String format, String schema);
/**
* Retrieves a schema by its reference (subject and version).
* @param schemaReference a {@link SchemaReference} used to identify the target schema.
* @return
*/
String fetch(SchemaReference schemaReference);
/**
* Retrieves a schema by its identifier.
* @param id the id of the target schema.
* @return
*/
String fetch(Integer id);
}

41
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/config/SchemaRegistryClientConfiguration.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.client.config;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.schema.client.DefaultSchemaRegistryClient;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;
/**
* @author Marius Bogoevici
*/
@Configuration
@EnableConfigurationProperties(SchemaRegistryClientProperties.class)
public class SchemaRegistryClientConfiguration {
@Bean
public SchemaRegistryClient schemaRegistryClient(SchemaRegistryClientProperties schemaRegistryClientProperties) {
DefaultSchemaRegistryClient defaultSchemaRegistryClient = new DefaultSchemaRegistryClient();
if (StringUtils.hasText(schemaRegistryClientProperties.getEndpoint())) {
defaultSchemaRegistryClient.setEndpoint(schemaRegistryClientProperties.getEndpoint());
}
return defaultSchemaRegistryClient;
}
}

36
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/config/SchemaRegistryClientProperties.java Normal file
View File

@@ -0,0 +1,36 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.schema.client.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author Marius Bogoevici
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.schemaRegistryClient")
public class SchemaRegistryClientProperties {
private String endpoint;
public String getEndpoint() {
return this.endpoint;
}
public void setEndpoint(String endpoint) {
this.endpoint = endpoint;
}
}

2
spring-cloud-stream-schema/src/main/resources/META-INF/spring.factories Normal file
View File

@@ -0,0 +1,2 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.stream.schema.avro.AvroMessageConverterAutoConfiguration

252
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java Normal file
View File

@@ -0,0 +1,252 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.schema.avro;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.Resource;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeType;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
*/
public class AvroSchemaMessageConverterTests {
static StubSchemaRegistryClient stubSchemaRegistryClient = new StubSchemaRegistryClient();
@Test
public void testSendMessageWithLocation() throws Exception {
ConfigurableApplicationContext sourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--schemaLocation=classpath:schemas/users_v1.schema",
"--spring.cloud.stream.schemaRegistryClient.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=avro/bytes");
Source source = sourceContext.getBean(Source.class);
User1 firstOutboundFoo = new User1();
firstOutboundFoo.setName("foo" + UUID.randomUUID().toString());
firstOutboundFoo.setFavoriteColor("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(firstOutboundFoo).build());
MessageCollector sourceMessageCollector = sourceContext.getBean(MessageCollector.class);
Message<?> outboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext barSourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--schemaLocation=classpath:schemas/users_v1.schema",
"--spring.cloud.stream.schemaRegistryClient.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=avro/bytes");
Source barSource = barSourceContext.getBean(Source.class);
User2 firstOutboundUser2 = new User2();
firstOutboundUser2.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundUser2.setFavoritePlace("foo" + UUID.randomUUID().toString());
firstOutboundUser2.setName("foo" + UUID.randomUUID().toString());
barSource.output().send(MessageBuilder.withPayload(firstOutboundUser2).build());
MessageCollector barSourceMessageCollector = barSourceContext.getBean(MessageCollector.class);
Message<?> barOutboundMessage = barSourceMessageCollector.forChannel(barSource.output()).poll(1000,
TimeUnit.MILLISECONDS);
assertThat(barOutboundMessage).isNotNull();
User2 secondUser2OutboundPojo = new User2();
secondUser2OutboundPojo.setFavoriteColor("foo" + UUID.randomUUID().toString());
secondUser2OutboundPojo.setFavoritePlace("foo" + UUID.randomUUID().toString());
secondUser2OutboundPojo.setName("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(secondUser2OutboundPojo).build());
Message<?> secondBarOutboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext sinkContext = SpringApplication.run(AvroSinkApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.schemaRegistryClient.enabled=false",
"--schemaLocation=classpath:schemas/users_v1.schema");
Sink sink = sinkContext.getBean(Sink.class);
sink.input().send(outboundMessage);
sink.input().send(barOutboundMessage);
sink.input().send(secondBarOutboundMessage);
List<User1> receivedUsers = sinkContext.getBean(AvroSinkApplication.class).receivedUsers;
assertThat(receivedUsers).hasSize(3);
assertThat(receivedUsers.get(0)).isNotSameAs(firstOutboundFoo);
assertThat(receivedUsers.get(0).getFavoriteColor()).isEqualTo(firstOutboundFoo.getFavoriteColor());
assertThat(receivedUsers.get(0).getName()).isEqualTo(firstOutboundFoo.getName());
assertThat(receivedUsers.get(1)).isNotSameAs(firstOutboundUser2);
assertThat(receivedUsers.get(1).getFavoriteColor()).isEqualTo(firstOutboundUser2.getFavoriteColor());
assertThat(receivedUsers.get(1).getName()).isEqualTo(firstOutboundUser2.getName());
assertThat(receivedUsers.get(2)).isNotSameAs(secondUser2OutboundPojo);
assertThat(receivedUsers.get(2).getFavoriteColor()).isEqualTo(secondUser2OutboundPojo.getFavoriteColor());
assertThat(receivedUsers.get(2).getName()).isEqualTo(secondUser2OutboundPojo.getName());
sourceContext.close();
}
@Test
public void testSendMessageWithoutLocation() throws Exception {
ConfigurableApplicationContext sourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.schemaRegistryClient.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=avro/bytes");
Source source = sourceContext.getBean(Source.class);
User1 firstOutboundFoo = new User1();
firstOutboundFoo.setName("foo" + UUID.randomUUID().toString());
firstOutboundFoo.setFavoriteColor("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(firstOutboundFoo).build());
MessageCollector sourceMessageCollector = sourceContext.getBean(MessageCollector.class);
Message<?> outboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext barSourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.schemaRegistryClient.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=avro/bytes");
Source barSource = barSourceContext.getBean(Source.class);
User2 firstOutboundUser2 = new User2();
firstOutboundUser2.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundUser2.setFavoritePlace("foo" + UUID.randomUUID().toString());
firstOutboundUser2.setName("foo" + UUID.randomUUID().toString());
barSource.output().send(MessageBuilder.withPayload(firstOutboundUser2).build());
MessageCollector barSourceMessageCollector = barSourceContext.getBean(MessageCollector.class);
Message<?> barOutboundMessage = barSourceMessageCollector.forChannel(barSource.output()).poll(1000,
TimeUnit.MILLISECONDS);
assertThat(barOutboundMessage).isNotNull();
User2 secondUser2OutboundPojo = new User2();
secondUser2OutboundPojo.setFavoriteColor("foo" + UUID.randomUUID().toString());
secondUser2OutboundPojo.setFavoritePlace("foo" + UUID.randomUUID().toString());
secondUser2OutboundPojo.setName("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(secondUser2OutboundPojo).build());
Message<?> secondBarOutboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext sinkContext = SpringApplication.run(AvroSinkApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.schemaRegistryClient.enabled=false");
Sink sink = sinkContext.getBean(Sink.class);
sink.input().send(outboundMessage);
sink.input().send(barOutboundMessage);
sink.input().send(secondBarOutboundMessage);
List<User1> receivedUsers = sinkContext.getBean(AvroSinkApplication.class).receivedUsers;
assertThat(receivedUsers).hasSize(3);
assertThat(receivedUsers.get(0)).isNotSameAs(firstOutboundFoo);
assertThat(receivedUsers.get(0).getFavoriteColor()).isEqualTo(firstOutboundFoo.getFavoriteColor());
assertThat(receivedUsers.get(0).getName()).isEqualTo(firstOutboundFoo.getName());
assertThat(receivedUsers.get(1)).isNotSameAs(firstOutboundUser2);
assertThat(receivedUsers.get(1).getFavoriteColor()).isEqualTo(firstOutboundUser2.getFavoriteColor());
assertThat(receivedUsers.get(1).getName()).isEqualTo(firstOutboundUser2.getName());
assertThat(receivedUsers.get(2)).isNotSameAs(secondUser2OutboundPojo);
assertThat(receivedUsers.get(2).getFavoriteColor()).isEqualTo(secondUser2OutboundPojo.getFavoriteColor());
assertThat(receivedUsers.get(2).getName()).isEqualTo(secondUser2OutboundPojo.getName());
sourceContext.close();
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@ConfigurationProperties
public static class AvroSourceApplication {
@Bean
public SchemaRegistryClient schemaRegistryClient() {
return stubSchemaRegistryClient;
}
private Resource schemaLocation;
public void setSchemaLocation(Resource schemaLocation) {
this.schemaLocation = schemaLocation;
}
@Bean
public MessageConverter userMessageConverter() throws IOException {
AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(
MimeType.valueOf("avro/bytes"));
if (schemaLocation != null) {
avroSchemaMessageConverter.setSchemaLocation(schemaLocation);
}
return avroSchemaMessageConverter;
}
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@ConfigurationProperties
public static class AvroSinkApplication {
public List<User1> receivedUsers = new ArrayList<>();
@StreamListener(Sink.INPUT)
public void listen(User1 user) {
receivedUsers.add(user);
}
private Resource schemaLocation;
public void setSchemaLocation(Resource schemaLocation) {
this.schemaLocation = schemaLocation;
}
@Bean
public MessageConverter userMessageConverter() throws IOException {
AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(
MimeType.valueOf("avro/bytes"));
if (schemaLocation != null) {
avroSchemaMessageConverter.setSchemaLocation(schemaLocation);
}
return avroSchemaMessageConverter;
}
}
}

145
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java Normal file
View File

@@ -0,0 +1,145 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.schema.avro;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.schema.client.EnableSchemaRegistryClient;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
import org.springframework.cloud.stream.schema.server.SchemaRegistryServerApplication;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
*/
public class AvroSchemaRegistryClientMessageConverterTests {
static SchemaRegistryClient stubSchemaRegistryClient = new StubSchemaRegistryClient();
@Test
public void testSendMessage() throws Exception {
ConfigurableApplicationContext schemaRegistryServerContext = SpringApplication.run(
SchemaRegistryServerApplication.class);
ConfigurableApplicationContext sourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=application/*+avro",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true");
Source source = sourceContext.getBean(Source.class);
User1 firstOutboundFoo = new User1();
firstOutboundFoo.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundFoo.setName("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(firstOutboundFoo).build());
MessageCollector sourceMessageCollector = sourceContext.getBean(MessageCollector.class);
Message<?> outboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext barSourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=application/vnd.user1.v1+avro",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true");
Source barSource = barSourceContext.getBean(Source.class);
User2 firstOutboundUser2 = new User2();
firstOutboundUser2.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundUser2.setName("foo" + UUID.randomUUID().toString());
barSource.output().send(MessageBuilder.withPayload(firstOutboundUser2).build());
MessageCollector barSourceMessageCollector = barSourceContext.getBean(MessageCollector.class);
Message<?> barOutboundMessage = barSourceMessageCollector.forChannel(barSource.output()).poll(1000,
TimeUnit.MILLISECONDS);
assertThat(barOutboundMessage).isNotNull();
User2 secondBarOutboundPojo = new User2();
secondBarOutboundPojo.setFavoriteColor("foo" + UUID.randomUUID().toString());
secondBarOutboundPojo.setName("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(secondBarOutboundPojo).build());
Message<?> secondBarOutboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext sinkContext = SpringApplication.run(AvroSinkApplication.class,
"--server.port=0", "--spring.jmx.enabled=false");
Sink sink = sinkContext.getBean(Sink.class);
sink.input().send(outboundMessage);
sink.input().send(barOutboundMessage);
sink.input().send(secondBarOutboundMessage);
List<User2> receivedPojos = sinkContext.getBean(AvroSinkApplication.class).receivedPojos;
assertThat(receivedPojos).hasSize(3);
assertThat(receivedPojos.get(0)).isNotSameAs(firstOutboundFoo);
assertThat(receivedPojos.get(0).getFavoriteColor()).isEqualTo(firstOutboundFoo.getFavoriteColor());
assertThat(receivedPojos.get(0).getName()).isEqualTo(firstOutboundFoo.getName());
assertThat(receivedPojos.get(0).getFavoritePlace()).isEqualTo("NYC");
assertThat(receivedPojos.get(1)).isNotSameAs(firstOutboundUser2);
assertThat(receivedPojos.get(1).getFavoriteColor()).isEqualTo(firstOutboundUser2.getFavoriteColor());
assertThat(receivedPojos.get(1).getName()).isEqualTo(firstOutboundUser2.getName());
assertThat(receivedPojos.get(1).getFavoritePlace()).isEqualTo("NYC");
assertThat(receivedPojos.get(2)).isNotSameAs(secondBarOutboundPojo);
assertThat(receivedPojos.get(2).getFavoriteColor()).isEqualTo(secondBarOutboundPojo.getFavoriteColor());
assertThat(receivedPojos.get(2).getName()).isEqualTo(secondBarOutboundPojo.getName());
assertThat(receivedPojos.get(2).getFavoritePlace()).isEqualTo(secondBarOutboundPojo.getFavoritePlace());
sinkContext.close();
barSourceContext.close();
sourceContext.close();
schemaRegistryServerContext.close();
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@EnableSchemaRegistryClient
public static class AvroSourceApplication {
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@EnableSchemaRegistryClient
public static class AvroSinkApplication {
public List<User2> receivedPojos = new ArrayList<>();
@StreamListener(Sink.INPUT)
public void listen(User2 fooPojo) {
receivedPojos.add(fooPojo);
}
}
}

144
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroStubSchemaRegistryClientMessageConverterTests.java Normal file
View File

@@ -0,0 +1,144 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.schema.avro;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
*/
public class AvroStubSchemaRegistryClientMessageConverterTests {
static SchemaRegistryClient stubSchemaRegistryClient = new StubSchemaRegistryClient();
@Test
public void testSendMessage() throws Exception {
ConfigurableApplicationContext sourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=application/*+avro",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true");
Source source = sourceContext.getBean(Source.class);
User1 firstOutboundFoo = new User1();
firstOutboundFoo.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundFoo.setName("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(firstOutboundFoo).build());
MessageCollector sourceMessageCollector = sourceContext.getBean(MessageCollector.class);
Message<?> outboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext barSourceContext = SpringApplication.run(AvroSourceApplication.class,
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.output.contentType=application/vnd.user1.v1+avro",
"--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true");
Source barSource = barSourceContext.getBean(Source.class);
User2 firstOutboundUser2 = new User2();
firstOutboundUser2.setFavoriteColor("foo" + UUID.randomUUID().toString());
firstOutboundUser2.setName("foo" + UUID.randomUUID().toString());
barSource.output().send(MessageBuilder.withPayload(firstOutboundUser2).build());
MessageCollector barSourceMessageCollector = barSourceContext.getBean(MessageCollector.class);
Message<?> barOutboundMessage = barSourceMessageCollector.forChannel(barSource.output()).poll(1000,
TimeUnit.MILLISECONDS);
assertThat(barOutboundMessage).isNotNull();
User2 secondBarOutboundPojo = new User2();
secondBarOutboundPojo.setFavoriteColor("foo" + UUID.randomUUID().toString());
secondBarOutboundPojo.setName("foo" + UUID.randomUUID().toString());
source.output().send(MessageBuilder.withPayload(secondBarOutboundPojo).build());
Message<?> secondBarOutboundMessage = sourceMessageCollector.forChannel(source.output()).poll(1000,
TimeUnit.MILLISECONDS);
ConfigurableApplicationContext sinkContext = SpringApplication.run(AvroSinkApplication.class,
"--server.port=0", "--spring.jmx.enabled=false");
Sink sink = sinkContext.getBean(Sink.class);
sink.input().send(outboundMessage);
sink.input().send(barOutboundMessage);
sink.input().send(secondBarOutboundMessage);
List<User2> receivedPojos = sinkContext.getBean(AvroSinkApplication.class).receivedPojos;
assertThat(receivedPojos).hasSize(3);
assertThat(receivedPojos.get(0)).isNotSameAs(firstOutboundFoo);
assertThat(receivedPojos.get(0).getFavoriteColor()).isEqualTo(firstOutboundFoo.getFavoriteColor());
assertThat(receivedPojos.get(0).getName()).isEqualTo(firstOutboundFoo.getName());
assertThat(receivedPojos.get(0).getFavoritePlace()).isEqualTo("NYC");
assertThat(receivedPojos.get(1)).isNotSameAs(firstOutboundUser2);
assertThat(receivedPojos.get(1).getFavoriteColor()).isEqualTo(firstOutboundUser2.getFavoriteColor());
assertThat(receivedPojos.get(1).getName()).isEqualTo(firstOutboundUser2.getName());
assertThat(receivedPojos.get(1).getFavoritePlace()).isEqualTo("NYC");
assertThat(receivedPojos.get(2)).isNotSameAs(secondBarOutboundPojo);
assertThat(receivedPojos.get(2).getFavoriteColor()).isEqualTo(secondBarOutboundPojo.getFavoriteColor());
assertThat(receivedPojos.get(2).getName()).isEqualTo(secondBarOutboundPojo.getName());
assertThat(receivedPojos.get(2).getFavoritePlace()).isEqualTo(secondBarOutboundPojo.getFavoritePlace());
sourceContext.close();
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class AvroSourceApplication {
@Bean
public SchemaRegistryClient schemaRegistryClient() {
return stubSchemaRegistryClient;
}
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class AvroSinkApplication {
public List<User2> receivedPojos = new ArrayList<>();
@StreamListener(Sink.INPUT)
public void listen(User2 fooPojo) {
receivedPojos.add(fooPojo);
}
@Bean
public SchemaRegistryClient schemaRegistryClient() {
return stubSchemaRegistryClient;
}
}
}

108
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/StubSchemaRegistryClient.java Normal file
View File

@@ -0,0 +1,108 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.schema.avro;
import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.cloud.stream.schema.SchemaNotFoundException;
import org.springframework.cloud.stream.schema.SchemaReference;
import org.springframework.cloud.stream.schema.SchemaRegistrationResponse;
import org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter;
import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
/**
* @author Marius Bogoevici
*/
public class StubSchemaRegistryClient implements SchemaRegistryClient {
private final AtomicInteger index = new AtomicInteger(0);
private final Map<Integer, String> schemasById = new HashMap<>();
private final Map<String, Map<Integer, SchemaWithId>> storedSchemas = new HashMap<>();
@Override
public SchemaRegistrationResponse register(String subject, String format, String schema) {
if (!this.storedSchemas.containsKey(subject)) {
this.storedSchemas.put(subject, new TreeMap<Integer, SchemaWithId>());
}
Map<Integer, SchemaWithId> schemaVersions = this.storedSchemas.get(subject);
for (Map.Entry<Integer, SchemaWithId> integerSchemaEntry : schemaVersions.entrySet()) {
if (integerSchemaEntry.getValue().getSchema().equals(schema)) {
SchemaRegistrationResponse schemaRegistrationResponse = new SchemaRegistrationResponse();
schemaRegistrationResponse.setId(integerSchemaEntry.getValue().getId());
schemaRegistrationResponse.setSchemaReference(
new SchemaReference(subject, integerSchemaEntry.getKey(),
AvroSchemaRegistryClientMessageConverter.AVRO_FORMAT));
return schemaRegistrationResponse;
}
}
int nextVersion = schemaVersions.size() + 1;
int id = this.index.incrementAndGet();
schemaVersions.put(nextVersion, new SchemaWithId(id, schema));
SchemaRegistrationResponse schemaRegistrationResponse = new SchemaRegistrationResponse();
schemaRegistrationResponse.setId(this.index.getAndIncrement());
schemaRegistrationResponse.setSchemaReference(
new SchemaReference(subject, nextVersion, AvroSchemaRegistryClientMessageConverter.AVRO_FORMAT));
this.schemasById.put(id, schema);
return schemaRegistrationResponse;
}
@Override
public String fetch(SchemaReference schemaReference) {
if (!AvroSchemaRegistryClientMessageConverter.AVRO_FORMAT.equals(schemaReference.getFormat())) {
throw new IllegalArgumentException("Only 'avro' is supported by this client");
}
if (!this.storedSchemas.containsKey(schemaReference.getSubject())) {
throw new SchemaNotFoundException("Not found: " + schemaReference);
}
if (!this.storedSchemas.get(schemaReference.getSubject()).containsKey(schemaReference.getVersion())) {
throw new SchemaNotFoundException("Not found: " + schemaReference);
}
return this.storedSchemas.get(schemaReference.getSubject()).get(schemaReference.getVersion()).getSchema();
}
@Override
public String fetch(Integer id) {
return this.schemasById.get(id);
}
static class SchemaWithId {
int id;
String schema;
SchemaWithId(int id, String schema) {
this.id = id;
this.schema = schema;
}
public int getId() {
return this.id;
}
public String getSchema() {
return this.schema;
}
}
}

57
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/User1.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.schema.avro;
import org.apache.avro.reflect.Nullable;
/**
* @author Marius Bogoevici
*/
public class User1 {
@Nullable
private String name;
private int favoriteNumber;
@Nullable
private String favoriteColor;
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public int getFavoriteNumber() {
return this.favoriteNumber;
}
public void setFavoriteNumber(int favoriteNumber) {
this.favoriteNumber = favoriteNumber;
}
public String getFavoriteColor() {
return this.favoriteColor;
}
public void setFavoriteColor(String favoriteColor) {
this.favoriteColor = favoriteColor;
}
}

69
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/User2.java Normal file
View File

@@ -0,0 +1,69 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.schema.avro;
import org.apache.avro.reflect.AvroDefault;
import org.apache.avro.reflect.Nullable;
/**
* @author Marius Bogoevici
*/
public class User2 {
@Nullable
private String name;
private int favoriteNumber;
@Nullable
private String favoriteColor;
@AvroDefault("\"NYC\"")
private String favoritePlace = "Boston";
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public int getFavoriteNumber() {
return this.favoriteNumber;
}
public void setFavoriteNumber(int favoriteNumber) {
this.favoriteNumber = favoriteNumber;
}
public String getFavoriteColor() {
return this.favoriteColor;
}
public void setFavoriteColor(String favoriteColor) {
this.favoriteColor = favoriteColor;
}
public String getFavoritePlace() {
return this.favoritePlace;
}
public void setFavoritePlace(String favoritePlace) {
this.favoritePlace = favoritePlace;
}
}

10
spring-cloud-stream-schema/src/test/resources/schemas/status.avsc Normal file
View File

@@ -0,0 +1,10 @@
{
"namespace":"org.springframework.cloud.stream.samples",
"name": "Status",
"type" : "record",
"fields": [
{"name": "id", "type": "string"},
{"name": "text", "type": "string"},
{"name": "timestamp", "type": "long"}
]
}

10
spring-cloud-stream-schema/src/test/resources/schemas/users_v1.schema Normal file
View File

@@ -0,0 +1,10 @@
{"namespace": "example.avro",
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type": "string"},
{"name": "favoriteNumber", "type": ["int", "null"]},
{"name": "favoriteColor", "type": ["string", "null"]}
]
}

10
spring-cloud-stream-schema/src/test/resources/schemas/users_v2.schema Normal file
View File

@@ -0,0 +1,10 @@
{"namespace": "example.avro",
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type": "string"},
{"name": "favoriteNumber", "type": ["int", "null"]},
{"name": "favoriteColor", "type": ["string", "null"]},
{"name": "favoritePlace", "type": ["string","null"], "default" : "NYC"}
]
}

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java
View File

@@ -30,7 +30,6 @@ import org.springframework.messaging.SubscribableChannel;
/**
* Class that is responsible for embedding apps using shared channel registry.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Venil Noronha
@@ -39,7 +38,8 @@ abstract class AggregateApplication {
private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
public static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
public static final String CHANNEL_NAMESPACE_PROPERTY_NAME =
SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
public static final String INPUT_CHANNEL_NAME = "input";
@@ -95,7 +95,8 @@ abstract class AggregateApplication {
static void createChildContexts(ConfigurableApplicationContext parentContext, Class<?>[] apps, String[][] args) {
for (int i = apps.length - 1; i >= 0; i--) {
String appClassName = apps[i].getName();
embedApp(parentContext, getNamespace(appClassName, i), apps[i]).run(args != null ? args[i] : new String[0]);
embedApp(parentContext, getNamespace(appClassName, i), apps[i]).run(args != null ? args[i] : new
String[0]);
}
}

24
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -45,7 +45,6 @@ import org.springframework.util.StringUtils;
/**
* Default {@link BinderFactory} implementation.
*
* @author Marius Bogoevici
*/
public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean, ApplicationContextAware {
@@ -97,9 +96,9 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
throw new IllegalStateException(
"A default binder has been requested, but there there is no binder available");
}
else if (!StringUtils.hasText(defaultBinder)) {
else if (!StringUtils.hasText(this.defaultBinder)) {
Set<String> defaultCandidateConfigurations = new HashSet<>();
for (Map.Entry<String, BinderConfiguration> binderConfigurationEntry : binderConfigurations
for (Map.Entry<String, BinderConfiguration> binderConfigurationEntry : this.binderConfigurations
.entrySet()) {
if (binderConfigurationEntry.getValue().isDefaultCandidate()) {
defaultCandidateConfigurations.add(binderConfigurationEntry.getKey());
@@ -149,7 +148,7 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
args.add(String.format("--%s=%s", property.getKey(), property.getValue()));
}
// Initialize the domain with a unique name based on the bootstrapping context setting
ConfigurableEnvironment environment = context != null ? context.getEnvironment() : null;
ConfigurableEnvironment environment = this.context != null ? this.context.getEnvironment() : null;
String defaultDomain = environment != null ? environment.getProperty("spring.jmx.default-domain") : null;
if (defaultDomain == null) {
defaultDomain = "";
@@ -163,16 +162,16 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
Arrays.asList(binderConfiguration.getBinderType().getConfigurationClasses()));
SpringApplicationBuilder springApplicationBuilder =
new SpringApplicationBuilder()
.sources(configurationClasses.toArray(new Class<?>[]{}))
.sources(configurationClasses.toArray(new Class<?>[] {}))
.bannerMode(Mode.OFF)
.web(false);
// If the environment is not customized and a main context is available, we will set the latter as parent.
// This ensures that the defaults and user-defined customizations (e.g. custom connection factory beans)
// are propagated to the binder context. If the environment is customized, then the binder context should
// not inherit any beans from the parent
boolean useApplicationContextAsParent = binderProperties.isEmpty() && context != null;
boolean useApplicationContextAsParent = binderProperties.isEmpty() && this.context != null;
if (useApplicationContextAsParent) {
springApplicationBuilder.parent(context);
springApplicationBuilder.parent(this.context);
}
if (useApplicationContextAsParent || (environment != null && binderConfiguration.isInheritEnvironment())) {
if (environment != null) {
@@ -185,23 +184,24 @@ public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean
springApplicationBuilder.run(args.toArray(new String[args.size()]));
@SuppressWarnings("unchecked")
Binder<T, ?, ?> binder = binderProducingContext.getBean(Binder.class);
if (bindersHealthIndicator != null) {
if (this.bindersHealthIndicator != null) {
OrderedHealthAggregator healthAggregator = new OrderedHealthAggregator();
Map<String, HealthIndicator> indicators = binderProducingContext.getBeansOfType(HealthIndicator.class);
// if there are no health indicators in the child context, we just mark the binder's health as unknown
// this can happen due to the fact that configuration is inherited
HealthIndicator binderHealthIndicator =
indicators.isEmpty() ? new DefaultHealthIndicator() : new CompositeHealthIndicator(healthAggregator, indicators);
bindersHealthIndicator.addHealthIndicator(configurationName, binderHealthIndicator);
indicators.isEmpty() ? new DefaultHealthIndicator() : new CompositeHealthIndicator(
healthAggregator, indicators);
this.bindersHealthIndicator.addHealthIndicator(configurationName, binderHealthIndicator);
}
this.binderInstanceCache.put(configurationName, new BinderInstanceHolder<>(binder, binderProducingContext));
this.binderInstanceCache.put(configurationName, new BinderInstanceHolder<>(binder,
binderProducingContext));
}
return this.binderInstanceCache.get(configurationName).getBinderInstance();
}
/**
* Utility class for storing {@link Binder} instances, along with their associated contexts.
*
* @param <T>
*/
private static final class BinderInstanceHolder<T> {

27
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -33,9 +33,11 @@ import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.MutableMessageBuilderFactory;
import org.springframework.integration.support.MutableMessageHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
@@ -122,6 +124,8 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea
private final MessageConverter messageConverter;
private final boolean provideHint;
private ContentTypeConvertingInterceptor(String contentType, boolean input) {
this.contentType = contentType;
this.mimeType = MessageConverterUtils.getMimeType(contentType);
@@ -144,6 +148,7 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea
this.messageConverter =
MessageConverterConfigurer.this.compositeMessageConverterFactory
.getMessageConverterForType(this.mimeType);
this.provideHint = this.messageConverter instanceof AbstractMessageConverter;
}
@Override
@@ -162,8 +167,26 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea
}
}
else {
Object converted = this.input ? this.messageConverter.fromMessage(message, this.klazz)
: this.messageConverter.toMessage(message.getPayload(), message.getHeaders());
Object converted;
if (this.input) {
if (this.provideHint) {
converted = ((AbstractMessageConverter) this.messageConverter).fromMessage(message, this.klazz,
this.mimeType);
}
else {
converted = this.messageConverter.fromMessage(message, this.klazz);
}
}
else {
if (this.provideHint) {
converted = ((AbstractMessageConverter) this.messageConverter).toMessage(message.getPayload(),
new MutableMessageHeaders(message.getHeaders()), this.mimeType);
}
else {
converted = this.messageConverter.toMessage(message.getPayload(),
new MutableMessageHeaders(message.getHeaders()));
}
}
if (converted instanceof Message) {
sentMessage = (Message<?>) converted;
}

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java
View File

@@ -89,7 +89,7 @@ public class CompositeMessageConverterFactory {
* @param mimeType the target MIME type
* @return a converter for the target MIME type
*/
public CompositeMessageConverter getMessageConverterForType(MimeType mimeType) {
public MessageConverter getMessageConverterForType(MimeType mimeType) {
List<MessageConverter> converters = new ArrayList<>();
for (MessageConverter converter : this.converters) {
if (converter instanceof AbstractMessageConverter) {
@@ -110,7 +110,12 @@ public class CompositeMessageConverterFactory {
throw new ConversionException("No message converter is registered for "
+ mimeType.toString());
}
return new CompositeMessageConverter(converters);
if (converters.size() > 1) {
return new CompositeMessageConverter(converters);
}
else {
return converters.get(0);
}
}
public CompositeMessageConverter getMessageConverterForAllRegistered() {

1
src/checkstyle/checkstyle.xml
View File

@@ -46,7 +46,6 @@
<!-- Class Design -->
<module name="FinalClass" />
<module name="InterfaceIsType" />
<module name="HideUtilityClassConstructor" />
<module name="MutableException" />
<module name="InnerTypeLast" />
<module name="OneTopLevelClass" />