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

1218 Inconsistent use of AvroSchemaRegistryClientMessageConverter.AVRO_FORMAT

Browse Source
This commit is contained in:
jmaxwell
2018-02-13 14:17:41 -06:00
committed by Soby Chacko
parent 30ad6b659a
commit 688d73597c
1 changed files with 16 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,7 +17,9 @@
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;
@@ -25,6 +27,7 @@ import org.apache.avro.Schema;
import org.apache.avro.generic.GenericContainer;
import org.apache.avro.reflect.ReflectData;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cache.CacheManager;
@@ -35,7 +38,6 @@ 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;
@@ -86,6 +88,8 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
public static final String REFERENCE_CACHE_NAME = CACHE_PREFIX + ".referenceCache";
public static final MimeType DEFAULT_AVRO_MIME_TYPE = new MimeType("application", "*+"+AVRO_FORMAT);
private Pattern versionedSchema;
private boolean dynamicSchemaGenerationEnabled;
@@ -119,7 +123,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
*/
public AvroSchemaRegistryClientMessageConverter(SchemaRegistryClient schemaRegistryClient,
CacheManager cacheManager) {
super(Arrays.asList(new MimeType("application", "*+avro")));
super(Collections.singletonList(DEFAULT_AVRO_MIME_TYPE));
Assert.notNull(schemaRegistryClient, "cannot be null");
Assert.notNull(cacheManager, "'cacheManager' cannot be null");
this.schemaRegistryClient = schemaRegistryClient;
@@ -165,7 +169,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
@Override
public void afterPropertiesSet() throws Exception {
this.versionedSchema = Pattern.compile("application/" + this.prefix
+ "\\.([\\p{Alnum}\\$\\.]+)\\.v(\\p{Digit}+)\\+avro");
+ "\\.([\\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()) {
@@ -220,7 +224,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
return true;
}
MimeType mimeType = getContentTypeResolver().resolve(headers);
return MimeType.valueOf("application/*+avro").includes(mimeType);
return DEFAULT_AVRO_MIME_TYPE.includes(mimeType);
}
@Override
@@ -248,12 +252,13 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
SchemaReference schemaReference = parsedSchema.getRegistration()
.getSchemaReference();
if (headers instanceof MutableMessageHeaders) {
headers.put(MessageHeaders.CONTENT_TYPE,
"application/" + this.prefix + "." + schemaReference.getSubject()
+ ".v" + schemaReference.getVersion() + "+avro");
}
DirectFieldAccessor dfa = new DirectFieldAccessor(headers);
@SuppressWarnings("unchecked")
Map<String, Object> _headers = (Map<String, Object>) dfa.getPropertyValue("headers");
_headers.put(MessageHeaders.CONTENT_TYPE,
"application/" + this.prefix + "." + schemaReference.getSubject()
+ ".v" + schemaReference.getVersion() + "+"+AVRO_FORMAT);
return schema;
}