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 287e657f5..b36686ec3 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 @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.schema.avro; +import java.lang.reflect.Constructor; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; @@ -29,6 +31,7 @@ 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; +import org.springframework.util.ReflectionUtils; /** * @author Marius Bogoevici @@ -63,6 +66,21 @@ public class AvroMessageConverterAutoConfiguration { } avroSchemaRegistryClientMessageConverter.setPrefix(this.avroMessageConverterProperties.getPrefix()); avroSchemaRegistryClientMessageConverter.setCacheManager(cacheManager()); + + try { + Class clazz = this.avroMessageConverterProperties.getSubjectNamingStrategy(); + Constructor constructor = ReflectionUtils.accessibleConstructor(clazz); + + avroSchemaRegistryClientMessageConverter.setSubjectNamingStrategy( + (SubjectNamingStrategy) constructor.newInstance() + ); + } catch (Exception ex) { + throw new IllegalStateException("Unable to create SubjectNamingStrategy " + + this.avroMessageConverterProperties.getSubjectNamingStrategy().toString(), + ex + ); + } + 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 index c2d82a78c..2090d6e2e 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 @@ -34,6 +34,8 @@ public class AvroMessageConverterProperties { private String prefix = "vnd"; + private Class subjectNamingStrategy = DefaultSubjectNamingStrategy.class; + public Resource getReaderSchema() { return this.readerSchema; } @@ -67,4 +69,13 @@ public class AvroMessageConverterProperties { public void setPrefix(String prefix) { this.prefix = prefix; } + + public Class getSubjectNamingStrategy() { + return subjectNamingStrategy; + } + + public void setSubjectNamingStrategy(Class subjectNamingStrategy) { + Assert.notNull(subjectNamingStrategy, "cannot be null"); + this.subjectNamingStrategy = subjectNamingStrategy; + } } 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 d8efadf25..fb783e614 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 @@ -100,6 +100,8 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag private String prefix = "vnd"; + private SubjectNamingStrategy subjectNamingStrategy; + /** * @deprecated as of release 1.2.2 in favor of * {@link #AvroSchemaRegistryClientMessageConverter(SchemaRegistryClient, CacheManager)} @@ -205,7 +207,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag } protected String toSubject(Schema schema) { - return schema.getName().toLowerCase(); + return subjectNamingStrategy.toSubject(schema); } @Override @@ -347,4 +349,8 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag Assert.notNull(cacheManager, "'cacheManager' cannot be null"); this.cacheManager = cacheManager; } + + public void setSubjectNamingStrategy(SubjectNamingStrategy subjectNamingStrategy) { + this.subjectNamingStrategy = subjectNamingStrategy; + } } diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/DefaultSubjectNamingStrategy.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/DefaultSubjectNamingStrategy.java new file mode 100644 index 000000000..eecb544ed --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/DefaultSubjectNamingStrategy.java @@ -0,0 +1,30 @@ +/* + * Copyright 2016-2017 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.apache.avro.Schema; + +/** + * @author David Kalosi + */ +public class DefaultSubjectNamingStrategy implements SubjectNamingStrategy { + + @Override + public String toSubject(Schema schema) { + return schema.getName().toLowerCase(); + } +} diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/SubjectNamingStrategy.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/SubjectNamingStrategy.java new file mode 100644 index 000000000..c1da40b89 --- /dev/null +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/SubjectNamingStrategy.java @@ -0,0 +1,33 @@ +/* + * Copyright 2016-2017 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.apache.avro.Schema; + +/** + * Provides function towards naming schema registry subjects for Avro files. + * @author David Kalosi + */ +public interface SubjectNamingStrategy { + + /** + * Takes the Avro schema on input and returns the generated subject under which the schema should be registered. + * @param schema + * @return subject name + */ + String toSubject(Schema schema); +} 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 ac624b5b8..a75814b21 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 @@ -41,6 +41,7 @@ import org.springframework.cache.support.NoOpCacheManager; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.schema.SchemaReference; import org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter; +import org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy; import org.springframework.cloud.stream.schema.client.DefaultSchemaRegistryClient; import org.springframework.cloud.stream.schema.client.SchemaRegistryClient; import org.springframework.cloud.stream.schema.server.SchemaRegistryServerApplication; @@ -86,6 +87,8 @@ public class AvroMessageConverterSerializationTests { SchemaRegistryClient client = new DefaultSchemaRegistryClient(); AvroSchemaRegistryClientMessageConverter converter = new AvroSchemaRegistryClientMessageConverter( client, new NoOpCacheManager()); + + converter.setSubjectNamingStrategy(new DefaultSubjectNamingStrategy()); converter.setDynamicSchemaGenerationEnabled(false); converter.afterPropertiesSet(); diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/CustomSubjectNamingStrategy.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/CustomSubjectNamingStrategy.java new file mode 100644 index 000000000..e165cb059 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/CustomSubjectNamingStrategy.java @@ -0,0 +1,35 @@ +/* + * Copyright 2016-2017 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.Schema; + +import org.springframework.cloud.stream.schema.avro.SubjectNamingStrategy; + +/** + * @author David Kalosi + */ +class CustomSubjectNamingStrategy implements SubjectNamingStrategy { + + CustomSubjectNamingStrategy() { + } + + @Override + public String toSubject(Schema schema) { + return schema.getFullName(); + } +} diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/SubjectNamingStrategyTest.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/SubjectNamingStrategyTest.java new file mode 100644 index 000000000..14efc6313 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/SubjectNamingStrategyTest.java @@ -0,0 +1,76 @@ +/* + * Copyright 2016-2017 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.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.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 David Kalosi + */ +public class SubjectNamingStrategyTest { + + static StubSchemaRegistryClient stubSchemaRegistryClient = new StubSchemaRegistryClient(); + + @Test + public void testCustomNamingStrategy() throws Exception { + ConfigurableApplicationContext sourceContext = SpringApplication.run(AvroSourceApplication.class, + "--server.port=0", + "--debug", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.output.contentType=application/*+avro", + "--spring.cloud.stream.schema.avro.subjectNamingStrategy=org.springframework.cloud.schema.avro.CustomSubjectNamingStrategy", + "--spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true"); + + Source source = sourceContext.getBean(Source.class); + User1 user1 = new User1(); + user1.setFavoriteColor("foo" + UUID.randomUUID().toString()); + user1.setName("foo" + UUID.randomUUID().toString()); + source.output().send(MessageBuilder.withPayload(user1).build()); + + MessageCollector barSourceMessageCollector = sourceContext.getBean(MessageCollector.class); + Message message = barSourceMessageCollector.forChannel(source.output()).poll(1000, TimeUnit.MILLISECONDS); + + assertThat(message.getHeaders().get("contentType")) + .isEqualTo("application/vnd.org.springframework.cloud.schema.avro.User1.v1+avro"); + } + + @EnableBinding(Source.class) + @EnableAutoConfiguration + public static class AvroSourceApplication { + + @Bean + public SchemaRegistryClient schemaRegistryClient() { + return stubSchemaRegistryClient; + } + } +}