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

subject naming strategy POC

Browse Source 
checkstyle fix

more checkstyle fixes
This commit is contained in:
David Kalosi
2017-10-24 20:47:06 +02:00
committed by Oleg Zhurakousky
parent d0be34f7cb
commit 08761c3de0
8 changed files with 213 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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;
}

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

@@ -34,6 +34,8 @@ public class AvroMessageConverterProperties {
private String prefix = "vnd";
private Class<? extends SubjectNamingStrategy> 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<? extends SubjectNamingStrategy> subjectNamingStrategy) {
Assert.notNull(subjectNamingStrategy, "cannot be null");
this.subjectNamingStrategy = subjectNamingStrategy;
}
}

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

@@ -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;
}
}

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

@@ -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();
}
}

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

@@ -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);
}

3
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroMessageConverterSerializationTests.java
View File

@@ -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();

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

@@ -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();
}
}

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

@@ -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;
}
}
}