diff --git a/spring-cloud-stream-schema/pom.xml b/spring-cloud-stream-schema/pom.xml index b0e554503..f28718ccb 100644 --- a/spring-cloud-stream-schema/pom.xml +++ b/spring-cloud-stream-schema/pom.xml @@ -68,8 +68,17 @@ + ${project.basedir}/target/generated-test-sources ${project.basedir}/target/generated-test-sources ${project.basedir}/src/test/resources/schemas + + **/*.avsc + + + ${project.basedir}/src/test/resources/schemas/imports/Email.avsc + ${project.basedir}/src/test/resources/schemas/imports/Sms.avsc + ${project.basedir}/src/test/resources/schemas/imports/PushNotification.avsc + 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 index 296f116fa..2e22ec936 100644 --- 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 @@ -43,15 +43,24 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.util.MimeType; +import org.springframework.util.ObjectUtils; /** * Base class for Apache Avro * {@link org.springframework.messaging.converter.MessageConverter} implementations. * @author Marius Bogoevici * @author Vinicius Carvalho + * @author Sercan Karaoglu */ public abstract class AbstractAvroMessageConverter extends AbstractMessageConverter { + /** + * common parser will let user to import external schemas. + */ + private Schema.Parser schemaParser = new Schema.Parser(); + + protected Resource[] schemaImports = new Resource[]{}; + protected AbstractAvroMessageConverter(MimeType supportedMimeType) { this(Collections.singletonList(supportedMimeType)); } @@ -61,8 +70,17 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver setContentTypeResolver(new OriginalContentTypeResolver()); } - protected static Schema parseSchema(Resource r) throws IOException { - return new Schema.Parser().parse(r.getInputStream()); + protected Schema parseSchema(Resource r) throws IOException { + if (ObjectUtils.isEmpty(schemaImports)) { + return new Schema.Parser().parse(r.getInputStream()); + } + else { + return schemaParser.parse(r.getInputStream()); + } + } + + protected void setSchemaImports(Resource[] imports) { + this.schemaImports = imports; } @Override 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 index 81de82679..17c1f283d 100644 --- 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 @@ -36,6 +36,7 @@ import org.springframework.util.ReflectionUtils; /** * @author Marius Bogoevici * @author Vinicius Carvalho + * @author Sercan Karaoglu */ @Configuration @ConditionalOnClass(name = "org.apache.avro.Schema") @@ -64,6 +65,10 @@ public class AvroMessageConverterAutoConfiguration { avroSchemaRegistryClientMessageConverter.setSchemaLocations( this.avroMessageConverterProperties.getSchemaLocations()); } + if (!ObjectUtils.isEmpty(this.avroMessageConverterProperties.getSchemaImports())) { + avroSchemaRegistryClientMessageConverter.setSchemaImports( + this.avroMessageConverterProperties.getSchemaImports()); + } avroSchemaRegistryClientMessageConverter.setPrefix(this.avroMessageConverterProperties.getPrefix()); try { 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 index 2090d6e2e..da1c71cef 100644 --- 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 @@ -22,6 +22,7 @@ import org.springframework.util.Assert; /** * @author Vinicius Carvalho + * @author Sercan Karaoglu */ @ConfigurationProperties(prefix = "spring.cloud.stream.schema.avro") public class AvroMessageConverterProperties { @@ -30,8 +31,22 @@ public class AvroMessageConverterProperties { private Resource readerSchema; + /** + * The source directory of Apache Avro schema. This schema is used by this + * converter. If this schema depends on other schemas consider defining those + * those dependent ones in the {@link #schemaImports} + * @parameter + */ private Resource[] schemaLocations; + /** + * A list of files or directories that should be loaded first thus making + * them importable by subsequent schemas. Note that imported files + * should not reference each other. + * @parameter + */ + private Resource[] schemaImports; + private String prefix = "vnd"; private Class subjectNamingStrategy = DefaultSubjectNamingStrategy.class; @@ -78,4 +93,13 @@ public class AvroMessageConverterProperties { Assert.notNull(subjectNamingStrategy, "cannot be null"); this.subjectNamingStrategy = subjectNamingStrategy; } + + public Resource[] getSchemaImports() { + return schemaImports; + } + + public void setSchemaImports(Resource[] schemaImports) { + this.schemaImports = schemaImports; + } + } 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 index dbc9358a7..3853405f1 100644 --- 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 @@ -18,10 +18,12 @@ package org.springframework.cloud.stream.schema.avro; import java.io.IOException; +import java.util.Arrays; import java.util.Collections; import java.util.Map; import java.util.regex.Matcher; import java.util.regex.Pattern; +import java.util.stream.Stream; import org.apache.avro.Schema; import org.apache.avro.generic.GenericContainer; @@ -71,6 +73,7 @@ import org.springframework.util.ObjectUtils; * @author Marius Bogoevici * @author Vinicius Carvalho * @author Oleg Zhurakousky + * @author Sercan Karaoglu */ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessageConverter implements InitializingBean { @@ -147,6 +150,16 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag this.schemaLocations = schemaLocations; } + /** + * A set of schema locations where should be imported first. Schemas provided at these + * locations will be reference, thus they should not reference each other. + * + * @param schemaImports + */ + public void setSchemaImports(Resource[] schemaImports) { + this.schemaImports = schemaImports; + } + /** * Set the prefix to be used in the publised subtype. Default 'vnd'. * @param prefix @@ -162,28 +175,35 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag public void afterPropertiesSet() throws Exception { this.versionedSchema = Pattern.compile("application/" + this.prefix + "\\.([\\p{Alnum}\\$\\.]+)\\.v(\\p{Digit}+)\\+"+AVRO_FORMAT); - 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 (schema.getType().equals(Schema.Type.UNION)) { - schema.getTypes().forEach(innerSchema -> registerSchema(schemaLocation, innerSchema)); - } else { - registerSchema(schemaLocation, schema); + + Stream.of(this.schemaImports, this.schemaLocations) + .filter(arr -> !ObjectUtils.isEmpty(arr)) + .distinct() + .peek(resources -> { + this.logger.info("Scanning avro schema resources on classpath"); + if (this.logger.isInfoEnabled()) { + this.logger.info("Parsing" + this.schemaImports.length); } + }).flatMap(Arrays::stream).forEach(resource -> { + try { + Schema schema = parseSchema(resource); + if (schema.getType().equals(Schema.Type.UNION)) { + schema.getTypes().forEach( + innerSchema -> registerSchema(resource, innerSchema)); } - catch (IOException e) { - if (this.logger.isWarnEnabled()) { - this.logger.warn("Failed to parse schema at " - + schemaLocation.getFilename(), e); - } + else { + registerSchema(resource, schema); } } - } + catch (IOException e) { + if (this.logger.isWarnEnabled()) { + this.logger.warn( + "Failed to parse schema at " + resource.getFilename(), + e); + } + } + }); + if (this.cacheManager instanceof NoOpCacheManager) { logger.warn("Schema caching is effectively disabled " + "since configured cache manager is a NoOpCacheManager. If this was not " diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java index 1d2cc915c..3a4c6b431 100644 --- a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java @@ -21,6 +21,10 @@ import java.util.Collections; import java.util.regex.Matcher; import java.util.regex.Pattern; +import example.avro.Command; +import example.avro.Email; +import example.avro.PushNotification; +import example.avro.Sms; import example.avro.User; import org.apache.avro.Schema; import org.apache.avro.generic.GenericData; @@ -55,101 +59,168 @@ import org.springframework.util.MimeTypeUtils; /** * @author Vinicius Carvalho + * @author Sercan Karaoglu */ public class AvroMessageConverterSerializationTests { - Pattern versionedSchema = Pattern.compile( - "application/" + "vnd" + "\\.([\\p{Alnum}\\$\\.]+)\\.v(\\p{Digit}+)\\+avro"); + Pattern versionedSchema = Pattern.compile("application/" + "vnd" + + "\\.([\\p{Alnum}\\$\\.]+)\\.v(\\p{Digit}+)\\+avro"); - Log logger = LogFactory.getLog(getClass()); + Log logger = LogFactory.getLog(getClass()); - private ConfigurableApplicationContext schemaRegistryServerContext; + private ConfigurableApplicationContext schemaRegistryServerContext; - @Before - public void setup() { - schemaRegistryServerContext = SpringApplication - .run(SchemaRegistryServerApplication.class, "--spring.main.allow-bean-definition-overriding=true"); - } - - @After - public void tearDown() { - schemaRegistryServerContext.close(); - } - - @Test - public void sourceWriteSameVersion() throws Exception { - User specificRecord = new User(); - specificRecord.setName("joe"); - Schema v1 = new Schema.Parser().parse(AvroMessageConverterSerializationTests.class - .getClassLoader().getResourceAsStream("schemas/user.avsc")); - GenericRecord genericRecord = new GenericData.Record(v1); - genericRecord.put("name", "joe"); - SchemaRegistryClient client = new DefaultSchemaRegistryClient(); - AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter( - client, new NoOpCacheManager()); - - converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy()); - converter.setDynamicSchemaGenerationEnabled(false); - converter.afterPropertiesSet(); - - Message specificMessage = converter.toMessage(specificRecord, - new MutableMessageHeaders(Collections. emptyMap()), - MimeTypeUtils.parseMimeType("application/*+avro")); - SchemaReference specificRef = extractSchemaReference(MimeTypeUtils.parseMimeType( - specificMessage.getHeaders().get("contentType").toString())); - - Message genericMessage = converter.toMessage(genericRecord, - new MutableMessageHeaders(Collections. emptyMap()), - MimeTypeUtils.parseMimeType("application/*+avro")); - SchemaReference genericRef = extractSchemaReference(MimeTypeUtils.parseMimeType( - genericMessage.getHeaders().get("contentType").toString())); - - Assert.assertEquals(genericRef, specificRef); - Assert.assertEquals(1, genericRef.getVersion()); - } - - @Test - public void testOriginalContentTypeHeaderOnly() throws Exception { - User specificRecord = new User(); - specificRecord.setName("joe"); - Schema v1 = new Schema.Parser().parse(AvroMessageConverterSerializationTests.class - .getClassLoader().getResourceAsStream("schemas/user.avsc")); - GenericRecord genericRecord = new GenericData.Record(v1); - genericRecord.put("name", "joe"); - SchemaRegistryClient client = new DefaultSchemaRegistryClient(); - client.register("user", "avro", v1.toString()); - AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter( - client, new NoOpCacheManager()); - converter.setDynamicSchemaGenerationEnabled(false); - converter.afterPropertiesSet(); - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - DatumWriter writer = new SpecificDatumWriter<>(User.class); - Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null); - writer.write(specificRecord, encoder); - encoder.flush(); - Message source = MessageBuilder.withPayload(baos.toByteArray()) - .setHeader(MessageHeaders.CONTENT_TYPE, - MimeTypeUtils.APPLICATION_OCTET_STREAM) - .setHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, - "application/vnd.user.v1+avro") - .build(); - Object converted = converter.fromMessage(source, User.class); - Assert.assertNotNull(converted); - Assert.assertEquals(specificRecord.getName().toString(), - ((User) converted).getName().toString()); - - } - - 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, - AvroSchemaRegistryClientMessageConverter.AVRO_FORMAT); + @Before + public void setup() { + schemaRegistryServerContext = SpringApplication + .run(SchemaRegistryServerApplication.class, + "--spring.main.allow-bean-definition-overriding=true"); + } + + @After + public void tearDown() { + schemaRegistryServerContext.close(); + } + + @Test + public void testSchemaImport() throws Exception { + SchemaRegistryClient client = new DefaultSchemaRegistryClient(); + AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter( + client, new NoOpCacheManager()); + converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy()); + converter.setDynamicSchemaGenerationEnabled(false); + converter.setSchemaLocations(schemaRegistryServerContext + .getResources("classpath:schemas/Command.avsc")); + converter.setSchemaImports(schemaRegistryServerContext + .getResources("classpath:schemas/imports/*.avsc")); + converter.afterPropertiesSet(); + Command notification = notification(); + Message specificMessage = converter.toMessage(notification, + new MutableMessageHeaders( + Collections. emptyMap())); + Object o = converter.fromMessage(specificMessage, Command.class); + + Assert.assertEquals("Serialization issue when use schema-imports", o, notification); + } + + @Test + public void sourceWriteSameVersion() throws Exception { + User specificRecord = new User(); + specificRecord.setName("joe"); + Schema v1 = new Schema.Parser() + .parse(AvroMessageConverterSerializationTests.class + .getClassLoader() + .getResourceAsStream("schemas/user.avsc")); + GenericRecord genericRecord = new GenericData.Record(v1); + genericRecord.put("name", "joe"); + SchemaRegistryClient client = new DefaultSchemaRegistryClient(); + AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter( + client, new NoOpCacheManager()); + + converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy()); + converter.setDynamicSchemaGenerationEnabled(false); + converter.afterPropertiesSet(); + + Message specificMessage = converter.toMessage(specificRecord, + new MutableMessageHeaders( + Collections. emptyMap()), + MimeTypeUtils.parseMimeType("application/*+avro")); + SchemaReference specificRef = extractSchemaReference(MimeTypeUtils + .parseMimeType(specificMessage.getHeaders().get("contentType") + .toString())); + + Message genericMessage = converter.toMessage(genericRecord, + new MutableMessageHeaders( + Collections. emptyMap()), + MimeTypeUtils.parseMimeType("application/*+avro")); + SchemaReference genericRef = extractSchemaReference(MimeTypeUtils + .parseMimeType(genericMessage.getHeaders().get("contentType") + .toString())); + + Assert.assertEquals(genericRef, specificRef); + Assert.assertEquals(1, genericRef.getVersion()); + } + + @Test + public void testOriginalContentTypeHeaderOnly() throws Exception { + User specificRecord = new User(); + specificRecord.setName("joe"); + Schema v1 = new Schema.Parser() + .parse(AvroMessageConverterSerializationTests.class + .getClassLoader() + .getResourceAsStream("schemas/user.avsc")); + GenericRecord genericRecord = new GenericData.Record(v1); + genericRecord.put("name", "joe"); + SchemaRegistryClient client = new DefaultSchemaRegistryClient(); + client.register("user", "avro", v1.toString()); + AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter( + client, new NoOpCacheManager()); + converter.setDynamicSchemaGenerationEnabled(false); + converter.afterPropertiesSet(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DatumWriter writer = new SpecificDatumWriter<>(User.class); + Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null); + writer.write(specificRecord, encoder); + encoder.flush(); + Message source = MessageBuilder.withPayload(baos.toByteArray()) + .setHeader(MessageHeaders.CONTENT_TYPE, + MimeTypeUtils.APPLICATION_OCTET_STREAM) + .setHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, + "application/vnd.user.v1+avro").build(); + Object converted = converter.fromMessage(source, User.class); + Assert.assertNotNull(converted); + Assert.assertEquals(specificRecord.getName().toString(), + ((User) converted).getName().toString()); + + } + + 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, + AvroSchemaRegistryClientMessageConverter.AVRO_FORMAT); + } + return schemaReference; + } + + public static Command notification() { + Command messageToSend = getCommandToSend(); + messageToSend.setType("notification"); + PushNotification pushNotification = new PushNotification(); + pushNotification.setArn("google"); + pushNotification.setText("hello"); + messageToSend.setPayload(pushNotification); + return messageToSend; + } + + public static Command sms() { + Command messageToSend = getCommandToSend(); + messageToSend.setType("sms"); + Sms sms = new Sms(); + sms.setPhoneNumber("6141231212"); + sms.setText("hello"); + messageToSend.setPayload(sms); + return messageToSend; + } + + public static Command email() { + Command messageToSend = getCommandToSend(); + messageToSend.setType("email"); + Email email = new Email(); + email.setAddressTo("sercan"); + email.setText("hello"); + email.setTitle("hi"); + messageToSend.setPayload(email); + return messageToSend; + } + + public static Command getCommandToSend() { + Command messageToSend = new Command(); + messageToSend.setCorrelationId("abc"); + return messageToSend; } - return schemaReference; - } } 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 index a56387fed..f40cfb2e1 100644 --- 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 @@ -21,6 +21,9 @@ import java.util.List; import java.util.UUID; import java.util.concurrent.TimeUnit; +import example.avro.Command; +import org.junit.After; +import org.junit.Before; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; @@ -47,130 +50,209 @@ import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.cloud.schema.avro.AvroMessageConverterSerializationTests.notification; /** * @author Marius Bogoevici * @author Oleg Zhurakousky + * @author Sercan Karaoglu */ public class AvroSchemaRegistryClientMessageConverterTests { - static SchemaRegistryClient stubSchemaRegistryClient = new StubSchemaRegistryClient(); + static SchemaRegistryClient stubSchemaRegistryClient = new StubSchemaRegistryClient(); - @Test - public void testSendMessage() throws Exception { + private ConfigurableApplicationContext schemaRegistryServerContext; - ConfigurableApplicationContext schemaRegistryServerContext = SpringApplication.run( - SchemaRegistryServerApplication.class, "--spring.main.allow-bean-definition-overriding=true"); - - 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("Boston"); - - 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(); - } - - @Test - public void testNoCacheConfiguration() { - ConfigurableApplicationContext sourceContext = SpringApplication.run(NoCacheConfiguration.class, - "--spring.main.web-environment=false"); - AvroSchemaRegistryClientMessageConverter converter = sourceContext - .getBean(AvroSchemaRegistryClientMessageConverter.class); - DirectFieldAccessor accessor = new DirectFieldAccessor(converter); - assertThat(accessor.getPropertyValue("cacheManager")).isInstanceOf(NoOpCacheManager.class); - } - - @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); + @Before + public void setup() { + schemaRegistryServerContext = SpringApplication + .run(SchemaRegistryServerApplication.class, + "--spring.main.allow-bean-definition-overriding=true"); } - } - - @Configuration - public static class NoCacheConfiguration { - - @Bean - @StreamMessageConverter - AvroSchemaRegistryClientMessageConverter avroSchemaRegistryClientMessageConverter() { - return new AvroSchemaRegistryClientMessageConverter(new DefaultSchemaRegistryClient(), new NoOpCacheManager()); + @After + public void tearDown() { + schemaRegistryServerContext.close(); } - @Bean - ServletWebServerFactory servletWebServerFactory(){ - return new TomcatServletWebServerFactory(); + @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("Boston"); + + 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(); + } + + @Test + public void testSchemaImportConfiguration() throws Exception{ + final String[] args = { "--server.port=0", "--spring.jmx.enabled=false", + "--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true", + "--spring.cloud.stream.bindings.output.contentType=application/*+avro", + "--spring.cloud.stream.bindings.output.destination=test", + "--spring.cloud.stream.bindings.schema-registry-client.endpoint=http://localhost:8990", + "--spring.cloud.stream.schema.avro.schema-locations=classpath:schemas/Command.avsc", + "--spring.cloud.stream.schema.avro.schema-imports=classpath:schemas/imports/Sms.avsc, classpath:schemas/imports/Email.avsc, classpath:schemas/imports/PushNotification.avsc" }; + + final ConfigurableApplicationContext sourceContext = SpringApplication + .run(AvroSourceApplication.class, args); + final ConfigurableApplicationContext sinkContext = SpringApplication + .run(CommandSinkApplication.class, args); + final Source barSource = sourceContext.getBean(Source.class); + final Command notification = notification(); + barSource.output() + .send(MessageBuilder.withPayload(notification).build()); + final MessageCollector barSourceMessageCollector = sourceContext + .getBean(MessageCollector.class); + final Message outboundMessage = barSourceMessageCollector + .forChannel(barSource.output()).poll(1000, TimeUnit.MILLISECONDS); + assertThat(outboundMessage).isNotNull(); + Sink sink = sinkContext.getBean(Sink.class); + sink.input().send(outboundMessage); + List receivedPojos = sinkContext + .getBean(CommandSinkApplication.class).receivedPojos; + + assertThat(receivedPojos).hasSize(1); + assertThat(receivedPojos.get(0)).isEqualTo(notification); + + } + + @Test + public void testNoCacheConfiguration() { + ConfigurableApplicationContext sourceContext = SpringApplication + .run(NoCacheConfiguration.class, + "--spring.main.web-environment=false"); + AvroSchemaRegistryClientMessageConverter converter = sourceContext + .getBean(AvroSchemaRegistryClientMessageConverter.class); + DirectFieldAccessor accessor = new DirectFieldAccessor(converter); + assertThat(accessor.getPropertyValue("cacheManager")) + .isInstanceOf(NoOpCacheManager.class); + } + + @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); + } + + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + @EnableSchemaRegistryClient + public static class CommandSinkApplication { + + public List receivedPojos = new ArrayList<>(); + + @StreamListener(Sink.INPUT) + public void listen(Command fooPojo) { + receivedPojos.add(fooPojo); + } + + } + + @Configuration + public static class NoCacheConfiguration { + + @Bean + @StreamMessageConverter + AvroSchemaRegistryClientMessageConverter avroSchemaRegistryClientMessageConverter() { + return new AvroSchemaRegistryClientMessageConverter( + new DefaultSchemaRegistryClient(), + new NoOpCacheManager()); + } + + @Bean + ServletWebServerFactory servletWebServerFactory() { + return new TomcatServletWebServerFactory(); + } } - } } diff --git a/spring-cloud-stream-schema/src/test/resources/schemas/Command.avsc b/spring-cloud-stream-schema/src/test/resources/schemas/Command.avsc new file mode 100644 index 000000000..754067146 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/resources/schemas/Command.avsc @@ -0,0 +1,19 @@ +{ + "namespace":"example.avro", + "name":"Command", + "type":"record", + "fields":[ + { + "name":"type", + "type":"string" + }, + { + "name":"correlationId", + "type":"string" + }, + { + "name":"payload", + "type":["Sms", "Email", "PushNotification"] + } + ] +} \ No newline at end of file diff --git a/spring-cloud-stream-schema/src/test/resources/schemas/imports/Email.avsc b/spring-cloud-stream-schema/src/test/resources/schemas/imports/Email.avsc new file mode 100644 index 000000000..c7c68e736 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/resources/schemas/imports/Email.avsc @@ -0,0 +1,19 @@ +{ + "namespace":"example.avro", + "name": "Email", + "type": "record", + "fields":[ + { + "name":"addressTo", + "type":"string" + }, + { + "name":"title", + "type":"string" + }, + { + "name":"text", + "type":"string" + } + ] +} \ No newline at end of file diff --git a/spring-cloud-stream-schema/src/test/resources/schemas/imports/PushNotification.avsc b/spring-cloud-stream-schema/src/test/resources/schemas/imports/PushNotification.avsc new file mode 100644 index 000000000..66286b536 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/resources/schemas/imports/PushNotification.avsc @@ -0,0 +1,15 @@ +{ + "namespace":"example.avro", + "name": "PushNotification", + "type": "record", + "fields":[ + { + "name":"arn", + "type":"string" + }, + { + "name":"text", + "type":"string" + } + ] +} \ No newline at end of file diff --git a/spring-cloud-stream-schema/src/test/resources/schemas/imports/Sms.avsc b/spring-cloud-stream-schema/src/test/resources/schemas/imports/Sms.avsc new file mode 100644 index 000000000..e306afcc0 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/resources/schemas/imports/Sms.avsc @@ -0,0 +1,14 @@ +{ + "namespace":"example.avro", + "name": "Sms", + "type": "record", + "fields":[ + { + "name":"phoneNumber", + "type":"string" + },{ + "name":"text", + "type":"string" + } + ] +} \ No newline at end of file