diff --git a/pom.xml b/pom.xml index c8bceeefe..af3bdcfd5 100644 --- a/pom.xml +++ b/pom.xml @@ -44,6 +44,8 @@ spring-cloud-stream-integration-tests spring-cloud-stream-docs spring-cloud-stream-reactive + spring-cloud-stream-schema + spring-cloud-stream-schema-server diff --git a/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 1407b2d90..0dbdc4281 100644 --- a/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -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. diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java index 2c48e7d82..a9c62ca5d 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java @@ -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 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) diff --git a/spring-cloud-stream-schema-server/pom.xml b/spring-cloud-stream-schema-server/pom.xml new file mode 100644 index 000000000..08f9f2787 --- /dev/null +++ b/spring-cloud-stream-schema-server/pom.xml @@ -0,0 +1,67 @@ + + + + + + spring-cloud-stream-parent + org.springframework.cloud + 1.1.0.BUILD-SNAPSHOT + + 4.0.0 + + spring-cloud-stream-schema-server + 1.1.0.BUILD-SNAPSHOT + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.boot + spring-boot-starter-data-jpa + + + com.h2database + h2 + 1.4.192 + + + org.apache.avro + avro + 1.8.1 + + + org.springframework.cloud + spring-cloud-stream-test-support + test + + + org.springframework.cloud + spring-cloud-stream-test-support-internal + test + + + + \ No newline at end of file diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/EnableSchemaRegistryServer.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/EnableSchemaRegistryServer.java new file mode 100644 index 000000000..f46fa5e92 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/EnableSchemaRegistryServer.java @@ -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 { +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerApplication.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerApplication.java new file mode 100644 index 000000000..76b084281 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerApplication.java @@ -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); + } +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerConfiguration.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerConfiguration.java new file mode 100644 index 000000000..78912590c --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerConfiguration.java @@ -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 schemaValidators() { + Map validatorMap = new HashMap<>(); + validatorMap.put("avro", new AvroSchemaValidator()); + return validatorMap; + } +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerProperties.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerProperties.java new file mode 100644 index 000000000..347663dca --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/config/SchemaServerProperties.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/controllers/ServerController.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/controllers/ServerController.java new file mode 100644 index 000000000..1cfb0989a --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/controllers/ServerController.java @@ -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 validators; + + public ServerController(SchemaRepository repository, + Map 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 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 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 response = new ResponseEntity<>(result, headers, + HttpStatus.CREATED); + + return response; + + } + + @RequestMapping(method = RequestMethod.GET, produces = "application/json", path = "/{subject}/{format}/v{version}") + public ResponseEntity 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 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) { + } + +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/model/Compatibility.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/model/Compatibility.java new file mode 100644 index 000000000..3a3bb9fe2 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/model/Compatibility.java @@ -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; +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/model/Schema.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/model/Schema.java new file mode 100644 index 000000000..28ee7db1c --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/model/Schema.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/repository/SchemaRepository.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/repository/SchemaRepository.java new file mode 100644 index 000000000..638e1e299 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/repository/SchemaRepository.java @@ -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 { + + List findBySubjectAndFormatOrderByVersion(String subject, + String format); + + Schema findOneBySubjectAndFormatAndVersion(String subject, String format, + Integer version); +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/AvroSchemaValidator.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/AvroSchemaValidator.java new file mode 100644 index 000000000..9da3ed3a9 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/AvroSchemaValidator.java @@ -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 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"; + } +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/InvalidSchemaException.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/InvalidSchemaException.java new file mode 100644 index 000000000..098f9f8d0 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/InvalidSchemaException.java @@ -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); + } +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/SchemaNotFoundException.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/SchemaNotFoundException.java new file mode 100644 index 000000000..b62620915 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/SchemaNotFoundException.java @@ -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); + } +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/SchemaValidator.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/SchemaValidator.java new file mode 100644 index 000000000..52d85d1c4 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/SchemaValidator.java @@ -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 schemas, String definition); + + String getFormat(); + +} diff --git a/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/UnsupportedFormatException.java b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/UnsupportedFormatException.java new file mode 100644 index 000000000..af0c12887 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/java/org/springframework/cloud/stream/schema/server/support/UnsupportedFormatException.java @@ -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); + } +} diff --git a/spring-cloud-stream-schema-server/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-schema-server/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..e69de29bb diff --git a/spring-cloud-stream-schema-server/src/main/resources/application.yml b/spring-cloud-stream-schema-server/src/main/resources/application.yml new file mode 100644 index 000000000..7961552db --- /dev/null +++ b/spring-cloud-stream-schema-server/src/main/resources/application.yml @@ -0,0 +1,5 @@ +spring: + application: + name: SchemaRegistryServer +server: + port: 8990 \ No newline at end of file diff --git a/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerAvroTests.java b/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerAvroTests.java new file mode 100644 index 000000000..b8f886d49 --- /dev/null +++ b/spring-cloud-stream-schema-server/src/test/java/org/springframework/cloud/stream/schema/server/SchemaRegistryServerAvroTests.java @@ -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 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 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 response = client.postForEntity("http://localhost:8990/", + schema, Schema.class); + Assert.assertTrue(response.getStatusCode().is2xxSuccessful()); + Assert.assertEquals(new Integer(1), response.getBody().getVersion()); + List location = response.getHeaders().get(HttpHeaders.LOCATION); + Assert.assertNotNull(location); + ResponseEntity 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 response = client.postForEntity("http://localhost:8990/", + schema, Schema.class); + Assert.assertTrue(response.getStatusCode().is2xxSuccessful()); + Assert.assertEquals(new Integer(1), response.getBody().getVersion()); + List location = response.getHeaders().get(HttpHeaders.LOCATION); + Assert.assertNotNull(location); + + ResponseEntity response2 = client.postForEntity("http://localhost:8990/", + schema2, Schema.class); + Assert.assertTrue(response.getStatusCode().is2xxSuccessful()); + Assert.assertEquals(new Integer(2), response2.getBody().getVersion()); + List 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 response = client.postForEntity("http://localhost:8990/", + schema, Schema.class); + Assert.assertTrue(response.getStatusCode().is2xxSuccessful()); + Assert.assertEquals(new Integer(1), response.getBody().getVersion()); + List location = response.getHeaders().get(HttpHeaders.LOCATION); + Assert.assertNotNull(location); + ResponseEntity response2 = client.postForEntity("http://localhost:8990/", + schema, Schema.class); + Assert.assertEquals(response.getBody().getId(), response2.getBody().getId()); + + } + + @Test + public void testSchemaNotfound() throws Exception { + ResponseEntity 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(); + } + } + +} diff --git a/spring-cloud-stream-schema/pom.xml b/spring-cloud-stream-schema/pom.xml new file mode 100644 index 000000000..62170434d --- /dev/null +++ b/spring-cloud-stream-schema/pom.xml @@ -0,0 +1,48 @@ + + + + spring-cloud-stream-parent + org.springframework.cloud + 1.1.0.BUILD-SNAPSHOT + + 4.0.0 + + spring-cloud-stream-schema + + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.boot + spring-boot-starter-test + test + + + org.apache.avro + avro + 1.8.1 + true + + + org.springframework.cloud + spring-cloud-stream-test-support + test + + + org.springframework.cloud + spring-cloud-stream-test-support-internal + test + + + org.springframework.cloud + spring-cloud-stream-schema-server + 1.1.0.BUILD-SNAPSHOT + test + + + diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaNotFoundException.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaNotFoundException.java new file mode 100644 index 000000000..d02f957d5 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaNotFoundException.java @@ -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); + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaReference.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaReference.java new file mode 100644 index 000000000..9ef5a628c --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaReference.java @@ -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 + '\'' + + '}'; + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaRegistrationResponse.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaRegistrationResponse.java new file mode 100644 index 000000000..5b98bbd1d --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/SchemaRegistrationResponse.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java new file mode 100644 index 000000000..7199c6c46 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AbstractAvroMessageConverter.java @@ -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 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 reader = getDatumReader((Class) 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 getDatumWriter(Class type, Schema schema) { + DatumWriter 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 getDatumReader(Class type, Schema schema, Schema writerSchema) { + DatumReader 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 writer = getDatumWriter((Class) 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); +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java new file mode 100644 index 000000000..d4aeed817 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterProperties.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterProperties.java new file mode 100644 index 000000000..c2d82a78c --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterProperties.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java new file mode 100644 index 000000000..779f7ae49 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaMessageConverter.java @@ -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 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; + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java new file mode 100644 index 000000000..56ad431ce --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java @@ -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: + * + *
  • + *
      prefix is a configurable prefix (default 'vnd');
    + *
      subject is a subject derived from the type of the outgoing object - typically the class name;
    + *
      version is the schema version for the given subject;
    + *
    • + * + * 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 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; + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java new file mode 100644 index 000000000..dbdc62ca5 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java @@ -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 request = new HttpEntity<>(payload, headers); + ResponseEntity 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 request = new HttpEntity<>("", headers); + ResponseEntity 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 request = new HttpEntity<>("", headers); + ResponseEntity response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request, Map + .class); + return (String) response.getBody().get("schema"); + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java new file mode 100644 index 000000000..02a6a325c --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java @@ -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 requestBody = new HashMap<>(); + requestBody.put("subject", subject); + requestBody.put("format", format); + requestBody.put("definition", schema); + ResponseEntity responseEntity = this.template.postForEntity(this.endpoint, requestBody, Map.class); + if (responseEntity.getStatusCode().is2xxSuccessful()) { + SchemaRegistrationResponse registrationResponse = new SchemaRegistrationResponse(); + Map responseBody = (Map) 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 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 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"); + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/EnableSchemaRegistryClient.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/EnableSchemaRegistryClient.java new file mode 100644 index 000000000..f7767cd64 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/EnableSchemaRegistryClient.java @@ -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 { + +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/SchemaRegistryClient.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/SchemaRegistryClient.java new file mode 100644 index 000000000..b01ec8a95 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/SchemaRegistryClient.java @@ -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); + +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/config/SchemaRegistryClientConfiguration.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/config/SchemaRegistryClientConfiguration.java new file mode 100644 index 000000000..f0fa71393 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/config/SchemaRegistryClientConfiguration.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/config/SchemaRegistryClientProperties.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/config/SchemaRegistryClientProperties.java new file mode 100644 index 000000000..b1aae14fd --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/config/SchemaRegistryClientProperties.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-schema/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..0530b64e9 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +org.springframework.cloud.stream.schema.avro.AvroMessageConverterAutoConfiguration diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java new file mode 100644 index 000000000..fae1f8d6f --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaMessageConverterTests.java @@ -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 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 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 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; + } + + } +} diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java new file mode 100644 index 000000000..016ddef2f --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java @@ -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 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 receivedPojos = new ArrayList<>(); + + @StreamListener(Sink.INPUT) + public void listen(User2 fooPojo) { + receivedPojos.add(fooPojo); + } + + } +} diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroStubSchemaRegistryClientMessageConverterTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroStubSchemaRegistryClientMessageConverterTests.java new file mode 100644 index 000000000..c21e50c9d --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroStubSchemaRegistryClientMessageConverterTests.java @@ -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 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 receivedPojos = new ArrayList<>(); + + @StreamListener(Sink.INPUT) + public void listen(User2 fooPojo) { + receivedPojos.add(fooPojo); + } + + @Bean + public SchemaRegistryClient schemaRegistryClient() { + return stubSchemaRegistryClient; + } + + } +} diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/StubSchemaRegistryClient.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/StubSchemaRegistryClient.java new file mode 100644 index 000000000..3a09db983 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/StubSchemaRegistryClient.java @@ -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 schemasById = new HashMap<>(); + + private final Map> storedSchemas = new HashMap<>(); + + @Override + public SchemaRegistrationResponse register(String subject, String format, String schema) { + if (!this.storedSchemas.containsKey(subject)) { + this.storedSchemas.put(subject, new TreeMap()); + } + Map schemaVersions = this.storedSchemas.get(subject); + for (Map.Entry 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; + } + } + +} diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/User1.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/User1.java new file mode 100644 index 000000000..6a2fa1cbd --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/User1.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/User2.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/User2.java new file mode 100644 index 000000000..422c4035e --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/User2.java @@ -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; + } +} diff --git a/spring-cloud-stream-schema/src/test/resources/schemas/status.avsc b/spring-cloud-stream-schema/src/test/resources/schemas/status.avsc new file mode 100644 index 000000000..5b0383229 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/resources/schemas/status.avsc @@ -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"} + ] +} \ No newline at end of file diff --git a/spring-cloud-stream-schema/src/test/resources/schemas/users_v1.schema b/spring-cloud-stream-schema/src/test/resources/schemas/users_v1.schema new file mode 100644 index 000000000..f5ef8c98d --- /dev/null +++ b/spring-cloud-stream-schema/src/test/resources/schemas/users_v1.schema @@ -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"]} + + ] +} \ No newline at end of file diff --git a/spring-cloud-stream-schema/src/test/resources/schemas/users_v2.schema b/spring-cloud-stream-schema/src/test/resources/schemas/users_v2.schema new file mode 100644 index 000000000..586a1ff47 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/resources/schemas/users_v2.schema @@ -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"} + ] +} \ No newline at end of file diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java index 2d6d93eb3..4646f6d7e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java @@ -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]); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index 323080f35..ca0ed6940 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -45,7 +45,6 @@ import org.springframework.util.StringUtils; /** * Default {@link BinderFactory} implementation. - * * @author Marius Bogoevici */ public class DefaultBinderFactory implements BinderFactory, DisposableBean, ApplicationContextAware { @@ -97,9 +96,9 @@ public class DefaultBinderFactory implements BinderFactory, 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 defaultCandidateConfigurations = new HashSet<>(); - for (Map.Entry binderConfigurationEntry : binderConfigurations + for (Map.Entry binderConfigurationEntry : this.binderConfigurations .entrySet()) { if (binderConfigurationEntry.getValue().isDefaultCandidate()) { defaultCandidateConfigurations.add(binderConfigurationEntry.getKey()); @@ -149,7 +148,7 @@ public class DefaultBinderFactory implements BinderFactory, 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 implements BinderFactory, 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 implements BinderFactory, DisposableBean springApplicationBuilder.run(args.toArray(new String[args.size()])); @SuppressWarnings("unchecked") Binder binder = binderProducingContext.getBean(Binder.class); - if (bindersHealthIndicator != null) { + if (this.bindersHealthIndicator != null) { OrderedHealthAggregator healthAggregator = new OrderedHealthAggregator(); Map 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 */ private static final class BinderInstanceHolder { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index 1bc954e05..e49eb8614 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -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; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java index fad90e011..4c12b5fca 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java @@ -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 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() { diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml index b1642ed35..c4baccbb4 100644 --- a/src/checkstyle/checkstyle.xml +++ b/src/checkstyle/checkstyle.xml @@ -46,7 +46,6 @@ -