From f566cb9749c6c69a6d0ef9f3218def4328cc542b Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 1 Nov 2018 13:11:39 +0100 Subject: [PATCH] Polishing GH-1517 Resolves #1517 Resolves #1519 --- .../avro/AbstractAvroMessageConverter.java | 14 +-- .../avro/AvroMessageConverterProperties.java | 2 +- ...oSchemaRegistryClientMessageConverter.java | 87 +++++++++---------- 3 files changed, 42 insertions(+), 61 deletions(-) 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 2e22ec936..7522a5a4b 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,7 +43,6 @@ 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 @@ -59,8 +58,6 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver */ private Schema.Parser schemaParser = new Schema.Parser(); - protected Resource[] schemaImports = new Resource[]{}; - protected AbstractAvroMessageConverter(MimeType supportedMimeType) { this(Collections.singletonList(supportedMimeType)); } @@ -71,16 +68,7 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver } 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; + return this.schemaParser.parse(r.getInputStream()); } @Override 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 da1c71cef..451d8c215 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2018 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. 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 3853405f1..4d843debc 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 @@ -97,7 +97,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag private boolean dynamicSchemaGenerationEnabled; - private CacheManager cacheManager; + private final CacheManager cacheManager; private Schema readerSchema; @@ -109,6 +109,8 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag private SubjectNamingStrategy subjectNamingStrategy; + protected Resource[] schemaImports = new Resource[]{}; + /** * Creates a new instance, configuring it with {@link SchemaRegistryClient} and * {@link CacheManager}. @@ -171,6 +173,20 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag this.prefix = prefix; } + 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); + } + } + + public void setSubjectNamingStrategy(SubjectNamingStrategy subjectNamingStrategy) { + this.subjectNamingStrategy = subjectNamingStrategy; + } + @Override public void afterPropertiesSet() throws Exception { this.versionedSchema = Pattern.compile("application/" + this.prefix @@ -212,22 +228,6 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag } } - private void registerSchema(Resource schemaLocation, Schema schema) { - 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()); - if (this.logger.isInfoEnabled()) { - this.logger.info("Schema " + schema.getName() - + " registered with id " + schema); - } - this.cacheManager.getCache(REFLECTION_CACHE_NAME) - .put(schema.getNamespace() + "." + schema.getName(), schema); - } - protected String toSubject(Schema schema) { return subjectNamingStrategy.toSubject(schema); } @@ -282,17 +282,6 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag 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) { @@ -320,16 +309,6 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag 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()) { @@ -360,16 +339,30 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag return schema; } - /** - * @deprecated as of release 1.0.4. Please use the constructor to inject CacheManager - */ - @Deprecated - public void setCacheManager(CacheManager cacheManager) { - Assert.notNull(cacheManager, "'cacheManager' cannot be null"); - this.cacheManager = cacheManager; + private void registerSchema(Resource schemaLocation, Schema schema) { + 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()); + if (this.logger.isInfoEnabled()) { + this.logger.info("Schema " + schema.getName() + + " registered with id " + schema); + } + this.cacheManager.getCache(REFLECTION_CACHE_NAME) + .put(schema.getNamespace() + "." + schema.getName(), schema); } - public void setSubjectNamingStrategy(SubjectNamingStrategy subjectNamingStrategy) { - this.subjectNamingStrategy = subjectNamingStrategy; + 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; } }