Add simple Avro transformers
Unsophisticated Avro transformers for `SpecificRecord` implementations. * * Fix DSL Factory - transformer was changed to `<? extends GenericContainer>`; change the DSL to match * * Restore test log4j * * Revert to supporting only SpecificRecord * * Fix assert * * Add multi-type deserialization with fluent API * Polishing for PR comments * * Remove type mappings; Exxpression now returns the type (or class name) * Cache types created from class names * Add type header in "toAvro" transformer * * Remove the type cache (already handled by the class loader) * * Don't convert the class to String in the "toAvro" transformer
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
91156444c4
commit
9a2b75bae3
@@ -101,6 +101,7 @@ subprojects { subproject ->
|
||||
ext {
|
||||
activeMqVersion = '5.15.9'
|
||||
apacheSshdVersion = '2.2.0'
|
||||
avroVersion = '1.8.2'
|
||||
aspectjVersion = '1.9.4'
|
||||
assertjVersion = '3.12.2'
|
||||
assertkVersion = '0.17'
|
||||
@@ -397,6 +398,7 @@ project('spring-integration-core') {
|
||||
compile("com.esotericsoftware:kryo-shaded:$kryoShadedVersion", optional)
|
||||
compile("io.micrometer:micrometer-core:$micrometerVersion", optional)
|
||||
compile("io.github.resilience4j:resilience4j-ratelimiter:$resilience4jVersion", optional)
|
||||
compile("org.apache.avro:avro:$avroVersion", optional)
|
||||
|
||||
testCompile ("org.aspectj:aspectjweaver:$aspectjVersion")
|
||||
testCompile "io.projectreactor:reactor-test:$reactorVersion"
|
||||
|
||||
@@ -35,7 +35,7 @@ public abstract class AbstractTransformer extends IntegrationObjectSupport imple
|
||||
return null;
|
||||
}
|
||||
return (result instanceof Message) ? (Message<?>) result
|
||||
: this.getMessageBuilderFactory().withPayload(result).copyHeaders(message.getHeaders()).build();
|
||||
: getMessageBuilderFactory().withPayload(result).copyHeaders(message.getHeaders()).build();
|
||||
}
|
||||
catch (MessageTransformationException e) { // NOSONAR - catch and throw
|
||||
throw e;
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
/*
|
||||
* Copyright 2019 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
|
||||
*
|
||||
* https://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.integration.transformer;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.UncheckedIOException;
|
||||
|
||||
import org.apache.avro.io.DatumReader;
|
||||
import org.apache.avro.io.DecoderFactory;
|
||||
import org.apache.avro.specific.SpecificDatumReader;
|
||||
import org.apache.avro.specific.SpecificRecord;
|
||||
|
||||
import org.springframework.beans.factory.BeanClassLoaderAware;
|
||||
import org.springframework.expression.EvaluationContext;
|
||||
import org.springframework.expression.Expression;
|
||||
import org.springframework.integration.context.IntegrationContextUtils;
|
||||
import org.springframework.integration.expression.FunctionExpression;
|
||||
import org.springframework.integration.transformer.support.AvroHeaders;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ClassUtils;
|
||||
|
||||
/**
|
||||
* An Apache Avro transformer to create generated {@link SpecificRecord} objects
|
||||
* from {@code byte[]}.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @since 5.2
|
||||
*
|
||||
*/
|
||||
public class SimpleFromAvroTransformer extends AbstractTransformer implements BeanClassLoaderAware {
|
||||
|
||||
private final Class<? extends SpecificRecord> defaultType;
|
||||
|
||||
private final DecoderFactory decoderFactory = new DecoderFactory();
|
||||
|
||||
private Expression typeIdExpression = new FunctionExpression<Message<?>>(
|
||||
msg -> msg.getHeaders().get(AvroHeaders.TYPE));
|
||||
|
||||
private EvaluationContext evaluationContext;
|
||||
|
||||
private ClassLoader beanClassLoader;
|
||||
|
||||
/**
|
||||
* Construct an instance with the supplied default type to create.
|
||||
* @param defaultType the type.
|
||||
*/
|
||||
public SimpleFromAvroTransformer(Class<? extends SpecificRecord> defaultType) {
|
||||
Assert.notNull(defaultType, "'defaultType' must not be null");
|
||||
this.defaultType = defaultType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanClassLoader(ClassLoader classLoader) {
|
||||
this.beanClassLoader = classLoader;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the expression to evaluate against the message to determine the type.
|
||||
* Default {@code headers['avro_type']}.
|
||||
* @param expression the expression.
|
||||
* @return the transformer
|
||||
*/
|
||||
public SimpleFromAvroTransformer typeExpression(Expression expression) {
|
||||
Assert.notNull(expression, "'expression' must not be null");
|
||||
this.typeIdExpression = expression;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the expression to evaluate against the message to determine the type id.
|
||||
* Default {@code headers['avro_type']}.
|
||||
* @param expression the expression.
|
||||
* @return the transformer
|
||||
*/
|
||||
public SimpleFromAvroTransformer typeExpression(String expression) {
|
||||
Assert.notNull(expression, "'expression' must not be null");
|
||||
this.typeIdExpression = EXPRESSION_PARSER.parseExpression(expression);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the expression to evaluate against the message to determine the type.
|
||||
* Default {@code headers['avro_type']}.
|
||||
* @param expression the expression.
|
||||
*/
|
||||
public void setTypeExpression(Expression expression) {
|
||||
Assert.notNull(expression, "'expression' must not be null");
|
||||
this.typeIdExpression = expression;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the expression to evaluate against the message to determine the type id.
|
||||
* Default {@code headers['avro_type']}.
|
||||
* @param expression the expression.
|
||||
*/
|
||||
public void setTypeExpression(String expression) {
|
||||
Assert.notNull(expression, "'expression' must not be null");
|
||||
this.typeIdExpression = EXPRESSION_PARSER.parseExpression(expression);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onInit() {
|
||||
this.evaluationContext = IntegrationContextUtils.getEvaluationContext(getBeanFactory());
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Override
|
||||
protected Object doTransform(Message<?> message) {
|
||||
Assert.state(message.getPayload() instanceof byte[], "Payload must be a byte[]");
|
||||
Class<? extends SpecificRecord> type = null;
|
||||
Object value = this.typeIdExpression.getValue(this.evaluationContext, message);
|
||||
if (value instanceof Class) {
|
||||
type = (Class<? extends SpecificRecord>) value;
|
||||
}
|
||||
else if (value instanceof String) {
|
||||
try {
|
||||
type = (Class<? extends SpecificRecord>) ClassUtils.forName((String) value, this.beanClassLoader);
|
||||
}
|
||||
catch (ClassNotFoundException | LinkageError e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
if (type == null) {
|
||||
type = this.defaultType;
|
||||
}
|
||||
DatumReader<?> reader = new SpecificDatumReader<>(type);
|
||||
try {
|
||||
return reader.read(null, this.decoderFactory.binaryDecoder((byte[]) message.getPayload(), null));
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new UncheckedIOException(e);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
/*
|
||||
* Copyright 2019 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
|
||||
*
|
||||
* https://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.integration.transformer;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.UncheckedIOException;
|
||||
|
||||
import org.apache.avro.io.BinaryEncoder;
|
||||
import org.apache.avro.io.DatumWriter;
|
||||
import org.apache.avro.io.EncoderFactory;
|
||||
import org.apache.avro.specific.SpecificDatumWriter;
|
||||
import org.apache.avro.specific.SpecificRecord;
|
||||
|
||||
import org.springframework.integration.transformer.support.AvroHeaders;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* An Apache Avro transformer for generated {@link SpecificRecord} objects.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @since 5.2
|
||||
*
|
||||
*/
|
||||
public class SimpleToAvroTransformer extends AbstractTransformer {
|
||||
|
||||
private final EncoderFactory encoderFactory = new EncoderFactory();
|
||||
|
||||
@Override
|
||||
protected Object doTransform(Message<?> message) {
|
||||
Assert.state(message.getPayload() instanceof SpecificRecord,
|
||||
"Payload must be an implementation of 'SpecificRecord'");
|
||||
SpecificRecord specific = (SpecificRecord) message.getPayload();
|
||||
ByteArrayOutputStream out = new ByteArrayOutputStream();
|
||||
BinaryEncoder encoder = this.encoderFactory.directBinaryEncoder(out, null);
|
||||
DatumWriter<Object> writer = new SpecificDatumWriter<>(specific.getSchema());
|
||||
try {
|
||||
writer.write(specific, encoder);
|
||||
encoder.flush();
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new UncheckedIOException(e);
|
||||
}
|
||||
return getMessageBuilderFactory().withPayload(out.toByteArray())
|
||||
.copyHeaders(message.getHeaders())
|
||||
.setHeader(AvroHeaders.TYPE, specific.getClass())
|
||||
.build();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
/*
|
||||
* Copyright 2019 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
|
||||
*
|
||||
* https://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.integration.transformer.support;
|
||||
|
||||
/**
|
||||
* Pre-defined names and prefixes for Apache Avro related headers.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @since 5.2
|
||||
*/
|
||||
public final class AvroHeaders {
|
||||
|
||||
private AvroHeaders() {
|
||||
super();
|
||||
}
|
||||
|
||||
public static final String PREFIX = "avro";
|
||||
|
||||
/**
|
||||
* The {@code SpecificRecord} type.
|
||||
*/
|
||||
public static final String TYPE = PREFIX + "_type";
|
||||
|
||||
}
|
||||
@@ -0,0 +1,308 @@
|
||||
/**
|
||||
* Autogenerated by Avro
|
||||
*
|
||||
* DO NOT EDIT DIRECTLY
|
||||
*/
|
||||
package org.springframework.integration.transformer;
|
||||
|
||||
import org.apache.avro.specific.SpecificData;
|
||||
import org.apache.avro.message.BinaryMessageEncoder;
|
||||
import org.apache.avro.message.BinaryMessageDecoder;
|
||||
import org.apache.avro.message.SchemaStore;
|
||||
|
||||
@SuppressWarnings("all")
|
||||
@org.apache.avro.specific.AvroGenerated
|
||||
public class AvroTestClass1 extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
|
||||
private static final long serialVersionUID = 72441923701471492L;
|
||||
public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"AvroTestClass1\",\"namespace\":\"org.springframework.integration.transformer\",\"fields\":[{\"name\":\"bar\",\"type\":[\"null\",{\"type\":\"string\",\"avro.java.string\":\"String\"}]},{\"name\":\"qux\",\"type\":[\"null\",{\"type\":\"string\",\"avro.java.string\":\"String\"}],\"default\":null}]}");
|
||||
public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
|
||||
|
||||
private static SpecificData MODEL$ = new SpecificData();
|
||||
|
||||
private static final BinaryMessageEncoder<AvroTestClass1> ENCODER =
|
||||
new BinaryMessageEncoder<AvroTestClass1>(MODEL$, SCHEMA$);
|
||||
|
||||
private static final BinaryMessageDecoder<AvroTestClass1> DECODER =
|
||||
new BinaryMessageDecoder<AvroTestClass1>(MODEL$, SCHEMA$);
|
||||
|
||||
/**
|
||||
* Return the BinaryMessageDecoder instance used by this class.
|
||||
*/
|
||||
public static BinaryMessageDecoder<AvroTestClass1> getDecoder() {
|
||||
return DECODER;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
|
||||
* @param resolver a {@link SchemaStore} used to find schemas by fingerprint
|
||||
*/
|
||||
public static BinaryMessageDecoder<AvroTestClass1> createDecoder(SchemaStore resolver) {
|
||||
return new BinaryMessageDecoder<AvroTestClass1>(MODEL$, SCHEMA$, resolver);
|
||||
}
|
||||
|
||||
/** Serializes this AvroTestClass1 to a ByteBuffer. */
|
||||
public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
|
||||
return ENCODER.encode(this);
|
||||
}
|
||||
|
||||
/** Deserializes a AvroTestClass1 from a ByteBuffer. */
|
||||
public static AvroTestClass1 fromByteBuffer(
|
||||
java.nio.ByteBuffer b) throws java.io.IOException {
|
||||
return DECODER.decode(b);
|
||||
}
|
||||
|
||||
@Deprecated public java.lang.String bar;
|
||||
@Deprecated public java.lang.String qux;
|
||||
|
||||
/**
|
||||
* Default constructor. Note that this does not initialize fields
|
||||
* to their default values from the schema. If that is desired then
|
||||
* one should use <code>newBuilder()</code>.
|
||||
*/
|
||||
public AvroTestClass1() {}
|
||||
|
||||
/**
|
||||
* All-args constructor.
|
||||
* @param bar The new value for bar
|
||||
* @param qux The new value for qux
|
||||
*/
|
||||
public AvroTestClass1(java.lang.String bar, java.lang.String qux) {
|
||||
this.bar = bar;
|
||||
this.qux = qux;
|
||||
}
|
||||
|
||||
public org.apache.avro.Schema getSchema() { return SCHEMA$; }
|
||||
// Used by DatumWriter. Applications should not call.
|
||||
public java.lang.Object get(int field$) {
|
||||
switch (field$) {
|
||||
case 0: return bar;
|
||||
case 1: return qux;
|
||||
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
|
||||
}
|
||||
}
|
||||
|
||||
// Used by DatumReader. Applications should not call.
|
||||
@SuppressWarnings(value="unchecked")
|
||||
public void put(int field$, java.lang.Object value$) {
|
||||
switch (field$) {
|
||||
case 0: bar = (java.lang.String)value$; break;
|
||||
case 1: qux = (java.lang.String)value$; break;
|
||||
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'bar' field.
|
||||
* @return The value of the 'bar' field.
|
||||
*/
|
||||
public java.lang.String getBar() {
|
||||
return bar;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'bar' field.
|
||||
* @param value the value to set.
|
||||
*/
|
||||
public void setBar(java.lang.String value) {
|
||||
this.bar = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'qux' field.
|
||||
* @return The value of the 'qux' field.
|
||||
*/
|
||||
public java.lang.String getQux() {
|
||||
return qux;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'qux' field.
|
||||
* @param value the value to set.
|
||||
*/
|
||||
public void setQux(java.lang.String value) {
|
||||
this.qux = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new AvroTestClass1 RecordBuilder.
|
||||
* @return A new AvroTestClass1 RecordBuilder
|
||||
*/
|
||||
public static org.springframework.integration.transformer.AvroTestClass1.Builder newBuilder() {
|
||||
return new org.springframework.integration.transformer.AvroTestClass1.Builder();
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new AvroTestClass1 RecordBuilder by copying an existing Builder.
|
||||
* @param other The existing builder to copy.
|
||||
* @return A new AvroTestClass1 RecordBuilder
|
||||
*/
|
||||
public static org.springframework.integration.transformer.AvroTestClass1.Builder newBuilder(org.springframework.integration.transformer.AvroTestClass1.Builder other) {
|
||||
return new org.springframework.integration.transformer.AvroTestClass1.Builder(other);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new AvroTestClass1 RecordBuilder by copying an existing AvroTestClass1 instance.
|
||||
* @param other The existing instance to copy.
|
||||
* @return A new AvroTestClass1 RecordBuilder
|
||||
*/
|
||||
public static org.springframework.integration.transformer.AvroTestClass1.Builder newBuilder(org.springframework.integration.transformer.AvroTestClass1 other) {
|
||||
return new org.springframework.integration.transformer.AvroTestClass1.Builder(other);
|
||||
}
|
||||
|
||||
/**
|
||||
* RecordBuilder for AvroTestClass1 instances.
|
||||
*/
|
||||
public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<AvroTestClass1>
|
||||
implements org.apache.avro.data.RecordBuilder<AvroTestClass1> {
|
||||
|
||||
private java.lang.String bar;
|
||||
private java.lang.String qux;
|
||||
|
||||
/** Creates a new Builder */
|
||||
private Builder() {
|
||||
super(SCHEMA$);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a Builder by copying an existing Builder.
|
||||
* @param other The existing Builder to copy.
|
||||
*/
|
||||
private Builder(org.springframework.integration.transformer.AvroTestClass1.Builder other) {
|
||||
super(other);
|
||||
if (isValidValue(fields()[0], other.bar)) {
|
||||
this.bar = data().deepCopy(fields()[0].schema(), other.bar);
|
||||
fieldSetFlags()[0] = true;
|
||||
}
|
||||
if (isValidValue(fields()[1], other.qux)) {
|
||||
this.qux = data().deepCopy(fields()[1].schema(), other.qux);
|
||||
fieldSetFlags()[1] = true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a Builder by copying an existing AvroTestClass1 instance
|
||||
* @param other The existing instance to copy.
|
||||
*/
|
||||
private Builder(org.springframework.integration.transformer.AvroTestClass1 other) {
|
||||
super(SCHEMA$);
|
||||
if (isValidValue(fields()[0], other.bar)) {
|
||||
this.bar = data().deepCopy(fields()[0].schema(), other.bar);
|
||||
fieldSetFlags()[0] = true;
|
||||
}
|
||||
if (isValidValue(fields()[1], other.qux)) {
|
||||
this.qux = data().deepCopy(fields()[1].schema(), other.qux);
|
||||
fieldSetFlags()[1] = true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'bar' field.
|
||||
* @return The value.
|
||||
*/
|
||||
public java.lang.String getBar() {
|
||||
return bar;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'bar' field.
|
||||
* @param value The value of 'bar'.
|
||||
* @return This builder.
|
||||
*/
|
||||
public org.springframework.integration.transformer.AvroTestClass1.Builder setBar(java.lang.String value) {
|
||||
validate(fields()[0], value);
|
||||
this.bar = value;
|
||||
fieldSetFlags()[0] = true;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether the 'bar' field has been set.
|
||||
* @return True if the 'bar' field has been set, false otherwise.
|
||||
*/
|
||||
public boolean hasBar() {
|
||||
return fieldSetFlags()[0];
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Clears the value of the 'bar' field.
|
||||
* @return This builder.
|
||||
*/
|
||||
public org.springframework.integration.transformer.AvroTestClass1.Builder clearBar() {
|
||||
bar = null;
|
||||
fieldSetFlags()[0] = false;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'qux' field.
|
||||
* @return The value.
|
||||
*/
|
||||
public java.lang.String getQux() {
|
||||
return qux;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'qux' field.
|
||||
* @param value The value of 'qux'.
|
||||
* @return This builder.
|
||||
*/
|
||||
public org.springframework.integration.transformer.AvroTestClass1.Builder setQux(java.lang.String value) {
|
||||
validate(fields()[1], value);
|
||||
this.qux = value;
|
||||
fieldSetFlags()[1] = true;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether the 'qux' field has been set.
|
||||
* @return True if the 'qux' field has been set, false otherwise.
|
||||
*/
|
||||
public boolean hasQux() {
|
||||
return fieldSetFlags()[1];
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Clears the value of the 'qux' field.
|
||||
* @return This builder.
|
||||
*/
|
||||
public org.springframework.integration.transformer.AvroTestClass1.Builder clearQux() {
|
||||
qux = null;
|
||||
fieldSetFlags()[1] = false;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public AvroTestClass1 build() {
|
||||
try {
|
||||
AvroTestClass1 record = new AvroTestClass1();
|
||||
record.bar = fieldSetFlags()[0] ? this.bar : (java.lang.String) defaultValue(fields()[0]);
|
||||
record.qux = fieldSetFlags()[1] ? this.qux : (java.lang.String) defaultValue(fields()[1]);
|
||||
return record;
|
||||
} catch (java.lang.Exception e) {
|
||||
throw new org.apache.avro.AvroRuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static final org.apache.avro.io.DatumWriter<AvroTestClass1>
|
||||
WRITER$ = (org.apache.avro.io.DatumWriter<AvroTestClass1>)MODEL$.createDatumWriter(SCHEMA$);
|
||||
|
||||
@Override public void writeExternal(java.io.ObjectOutput out)
|
||||
throws java.io.IOException {
|
||||
WRITER$.write(this, SpecificData.getEncoder(out));
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static final org.apache.avro.io.DatumReader<AvroTestClass1>
|
||||
READER$ = (org.apache.avro.io.DatumReader<AvroTestClass1>)MODEL$.createDatumReader(SCHEMA$);
|
||||
|
||||
@Override public void readExternal(java.io.ObjectInput in)
|
||||
throws java.io.IOException {
|
||||
READER$.read(this, SpecificData.getDecoder(in));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,308 @@
|
||||
/**
|
||||
* Autogenerated by Avro
|
||||
*
|
||||
* DO NOT EDIT DIRECTLY
|
||||
*/
|
||||
package org.springframework.integration.transformer;
|
||||
|
||||
import org.apache.avro.specific.SpecificData;
|
||||
import org.apache.avro.message.BinaryMessageEncoder;
|
||||
import org.apache.avro.message.BinaryMessageDecoder;
|
||||
import org.apache.avro.message.SchemaStore;
|
||||
|
||||
@SuppressWarnings("all")
|
||||
@org.apache.avro.specific.AvroGenerated
|
||||
public class AvroTestClass2 extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
|
||||
private static final long serialVersionUID = -5029139830458327575L;
|
||||
public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"AvroTestClass2\",\"namespace\":\"org.springframework.integration.transformer\",\"fields\":[{\"name\":\"bar\",\"type\":[\"null\",{\"type\":\"string\",\"avro.java.string\":\"String\"}]},{\"name\":\"qux\",\"type\":[\"null\",{\"type\":\"string\",\"avro.java.string\":\"String\"}],\"default\":null}]}");
|
||||
public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
|
||||
|
||||
private static SpecificData MODEL$ = new SpecificData();
|
||||
|
||||
private static final BinaryMessageEncoder<AvroTestClass2> ENCODER =
|
||||
new BinaryMessageEncoder<AvroTestClass2>(MODEL$, SCHEMA$);
|
||||
|
||||
private static final BinaryMessageDecoder<AvroTestClass2> DECODER =
|
||||
new BinaryMessageDecoder<AvroTestClass2>(MODEL$, SCHEMA$);
|
||||
|
||||
/**
|
||||
* Return the BinaryMessageDecoder instance used by this class.
|
||||
*/
|
||||
public static BinaryMessageDecoder<AvroTestClass2> getDecoder() {
|
||||
return DECODER;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
|
||||
* @param resolver a {@link SchemaStore} used to find schemas by fingerprint
|
||||
*/
|
||||
public static BinaryMessageDecoder<AvroTestClass2> createDecoder(SchemaStore resolver) {
|
||||
return new BinaryMessageDecoder<AvroTestClass2>(MODEL$, SCHEMA$, resolver);
|
||||
}
|
||||
|
||||
/** Serializes this AvroTestClass2 to a ByteBuffer. */
|
||||
public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
|
||||
return ENCODER.encode(this);
|
||||
}
|
||||
|
||||
/** Deserializes a AvroTestClass2 from a ByteBuffer. */
|
||||
public static AvroTestClass2 fromByteBuffer(
|
||||
java.nio.ByteBuffer b) throws java.io.IOException {
|
||||
return DECODER.decode(b);
|
||||
}
|
||||
|
||||
@Deprecated public java.lang.String bar;
|
||||
@Deprecated public java.lang.String qux;
|
||||
|
||||
/**
|
||||
* Default constructor. Note that this does not initialize fields
|
||||
* to their default values from the schema. If that is desired then
|
||||
* one should use <code>newBuilder()</code>.
|
||||
*/
|
||||
public AvroTestClass2() {}
|
||||
|
||||
/**
|
||||
* All-args constructor.
|
||||
* @param bar The new value for bar
|
||||
* @param qux The new value for qux
|
||||
*/
|
||||
public AvroTestClass2(java.lang.String bar, java.lang.String qux) {
|
||||
this.bar = bar;
|
||||
this.qux = qux;
|
||||
}
|
||||
|
||||
public org.apache.avro.Schema getSchema() { return SCHEMA$; }
|
||||
// Used by DatumWriter. Applications should not call.
|
||||
public java.lang.Object get(int field$) {
|
||||
switch (field$) {
|
||||
case 0: return bar;
|
||||
case 1: return qux;
|
||||
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
|
||||
}
|
||||
}
|
||||
|
||||
// Used by DatumReader. Applications should not call.
|
||||
@SuppressWarnings(value="unchecked")
|
||||
public void put(int field$, java.lang.Object value$) {
|
||||
switch (field$) {
|
||||
case 0: bar = (java.lang.String)value$; break;
|
||||
case 1: qux = (java.lang.String)value$; break;
|
||||
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'bar' field.
|
||||
* @return The value of the 'bar' field.
|
||||
*/
|
||||
public java.lang.String getBar() {
|
||||
return bar;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'bar' field.
|
||||
* @param value the value to set.
|
||||
*/
|
||||
public void setBar(java.lang.String value) {
|
||||
this.bar = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'qux' field.
|
||||
* @return The value of the 'qux' field.
|
||||
*/
|
||||
public java.lang.String getQux() {
|
||||
return qux;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'qux' field.
|
||||
* @param value the value to set.
|
||||
*/
|
||||
public void setQux(java.lang.String value) {
|
||||
this.qux = value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new AvroTestClass2 RecordBuilder.
|
||||
* @return A new AvroTestClass2 RecordBuilder
|
||||
*/
|
||||
public static org.springframework.integration.transformer.AvroTestClass2.Builder newBuilder() {
|
||||
return new org.springframework.integration.transformer.AvroTestClass2.Builder();
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new AvroTestClass2 RecordBuilder by copying an existing Builder.
|
||||
* @param other The existing builder to copy.
|
||||
* @return A new AvroTestClass2 RecordBuilder
|
||||
*/
|
||||
public static org.springframework.integration.transformer.AvroTestClass2.Builder newBuilder(org.springframework.integration.transformer.AvroTestClass2.Builder other) {
|
||||
return new org.springframework.integration.transformer.AvroTestClass2.Builder(other);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new AvroTestClass2 RecordBuilder by copying an existing AvroTestClass2 instance.
|
||||
* @param other The existing instance to copy.
|
||||
* @return A new AvroTestClass2 RecordBuilder
|
||||
*/
|
||||
public static org.springframework.integration.transformer.AvroTestClass2.Builder newBuilder(org.springframework.integration.transformer.AvroTestClass2 other) {
|
||||
return new org.springframework.integration.transformer.AvroTestClass2.Builder(other);
|
||||
}
|
||||
|
||||
/**
|
||||
* RecordBuilder for AvroTestClass2 instances.
|
||||
*/
|
||||
public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<AvroTestClass2>
|
||||
implements org.apache.avro.data.RecordBuilder<AvroTestClass2> {
|
||||
|
||||
private java.lang.String bar;
|
||||
private java.lang.String qux;
|
||||
|
||||
/** Creates a new Builder */
|
||||
private Builder() {
|
||||
super(SCHEMA$);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a Builder by copying an existing Builder.
|
||||
* @param other The existing Builder to copy.
|
||||
*/
|
||||
private Builder(org.springframework.integration.transformer.AvroTestClass2.Builder other) {
|
||||
super(other);
|
||||
if (isValidValue(fields()[0], other.bar)) {
|
||||
this.bar = data().deepCopy(fields()[0].schema(), other.bar);
|
||||
fieldSetFlags()[0] = true;
|
||||
}
|
||||
if (isValidValue(fields()[1], other.qux)) {
|
||||
this.qux = data().deepCopy(fields()[1].schema(), other.qux);
|
||||
fieldSetFlags()[1] = true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a Builder by copying an existing AvroTestClass2 instance
|
||||
* @param other The existing instance to copy.
|
||||
*/
|
||||
private Builder(org.springframework.integration.transformer.AvroTestClass2 other) {
|
||||
super(SCHEMA$);
|
||||
if (isValidValue(fields()[0], other.bar)) {
|
||||
this.bar = data().deepCopy(fields()[0].schema(), other.bar);
|
||||
fieldSetFlags()[0] = true;
|
||||
}
|
||||
if (isValidValue(fields()[1], other.qux)) {
|
||||
this.qux = data().deepCopy(fields()[1].schema(), other.qux);
|
||||
fieldSetFlags()[1] = true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'bar' field.
|
||||
* @return The value.
|
||||
*/
|
||||
public java.lang.String getBar() {
|
||||
return bar;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'bar' field.
|
||||
* @param value The value of 'bar'.
|
||||
* @return This builder.
|
||||
*/
|
||||
public org.springframework.integration.transformer.AvroTestClass2.Builder setBar(java.lang.String value) {
|
||||
validate(fields()[0], value);
|
||||
this.bar = value;
|
||||
fieldSetFlags()[0] = true;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether the 'bar' field has been set.
|
||||
* @return True if the 'bar' field has been set, false otherwise.
|
||||
*/
|
||||
public boolean hasBar() {
|
||||
return fieldSetFlags()[0];
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Clears the value of the 'bar' field.
|
||||
* @return This builder.
|
||||
*/
|
||||
public org.springframework.integration.transformer.AvroTestClass2.Builder clearBar() {
|
||||
bar = null;
|
||||
fieldSetFlags()[0] = false;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the value of the 'qux' field.
|
||||
* @return The value.
|
||||
*/
|
||||
public java.lang.String getQux() {
|
||||
return qux;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the value of the 'qux' field.
|
||||
* @param value The value of 'qux'.
|
||||
* @return This builder.
|
||||
*/
|
||||
public org.springframework.integration.transformer.AvroTestClass2.Builder setQux(java.lang.String value) {
|
||||
validate(fields()[1], value);
|
||||
this.qux = value;
|
||||
fieldSetFlags()[1] = true;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether the 'qux' field has been set.
|
||||
* @return True if the 'qux' field has been set, false otherwise.
|
||||
*/
|
||||
public boolean hasQux() {
|
||||
return fieldSetFlags()[1];
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Clears the value of the 'qux' field.
|
||||
* @return This builder.
|
||||
*/
|
||||
public org.springframework.integration.transformer.AvroTestClass2.Builder clearQux() {
|
||||
qux = null;
|
||||
fieldSetFlags()[1] = false;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public AvroTestClass2 build() {
|
||||
try {
|
||||
AvroTestClass2 record = new AvroTestClass2();
|
||||
record.bar = fieldSetFlags()[0] ? this.bar : (java.lang.String) defaultValue(fields()[0]);
|
||||
record.qux = fieldSetFlags()[1] ? this.qux : (java.lang.String) defaultValue(fields()[1]);
|
||||
return record;
|
||||
} catch (java.lang.Exception e) {
|
||||
throw new org.apache.avro.AvroRuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static final org.apache.avro.io.DatumWriter<AvroTestClass2>
|
||||
WRITER$ = (org.apache.avro.io.DatumWriter<AvroTestClass2>)MODEL$.createDatumWriter(SCHEMA$);
|
||||
|
||||
@Override public void writeExternal(java.io.ObjectOutput out)
|
||||
throws java.io.IOException {
|
||||
WRITER$.write(this, SpecificData.getEncoder(out));
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static final org.apache.avro.io.DatumReader<AvroTestClass2>
|
||||
READER$ = (org.apache.avro.io.DatumReader<AvroTestClass2>)MODEL$.createDatumReader(SCHEMA$);
|
||||
|
||||
@Override public void readExternal(java.io.ObjectInput in)
|
||||
throws java.io.IOException {
|
||||
READER$.read(this, SpecificData.getDecoder(in));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,189 @@
|
||||
/*
|
||||
* Copyright 2019 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
|
||||
*
|
||||
* https://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.integration.transformer;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.config.EnableIntegration;
|
||||
import org.springframework.integration.dsl.IntegrationFlow;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.integration.transformer.support.AvroHeaders;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 5.2
|
||||
*
|
||||
*/
|
||||
@SpringJUnitConfig
|
||||
public class AvroTests {
|
||||
|
||||
@Test
|
||||
void testTransformers(@Autowired Config config) {
|
||||
AvroTestClass1 test = new AvroTestClass1("baz", "fiz");
|
||||
config.in1().send(new GenericMessage<>(test));
|
||||
assertThat(config.tapped().receive(0))
|
||||
.isNotNull()
|
||||
.extracting(msg -> msg.getPayload())
|
||||
.isInstanceOf(byte[].class);
|
||||
assertThat(config.out().receive(0))
|
||||
.isNotNull()
|
||||
.extracting(msg -> msg.getPayload())
|
||||
.isEqualTo(test)
|
||||
.isNotSameAs(test);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testMultiTypeTransformers(@Autowired Config config) {
|
||||
AvroTestClass1 test = new AvroTestClass1("baz", "fiz");
|
||||
config.in2().send(new GenericMessage<>(test));
|
||||
assertThat(config.tapped().receive(0))
|
||||
.isNotNull()
|
||||
.extracting(msg -> msg.getPayload())
|
||||
.isInstanceOf(byte[].class);
|
||||
assertThat(config.out().receive(0))
|
||||
.isNotNull()
|
||||
.extracting(msg -> msg.getPayload())
|
||||
.isNotEqualTo(test)
|
||||
.isInstanceOf(AvroTestClass2.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testMultiTypeTransformersClassName(@Autowired Config config) {
|
||||
AvroTestClass1 test = new AvroTestClass1("baz", "fiz");
|
||||
config.in3().send(new GenericMessage<>(test));
|
||||
assertThat(config.tapped().receive(0))
|
||||
.isNotNull()
|
||||
.extracting(msg -> msg.getPayload())
|
||||
.isInstanceOf(byte[].class);
|
||||
assertThat(config.out().receive(0))
|
||||
.isNotNull()
|
||||
.extracting(msg -> msg.getPayload())
|
||||
.isNotEqualTo(test)
|
||||
.isInstanceOf(AvroTestClass2.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testTransformersNoHeaderPresent(@Autowired Config config) {
|
||||
AvroTestClass1 test = new AvroTestClass1("baz", "fiz");
|
||||
config.in4().send(new GenericMessage<>(test));
|
||||
assertThat(config.tapped().receive(0))
|
||||
.isNotNull()
|
||||
.extracting(msg -> msg.getPayload())
|
||||
.isInstanceOf(byte[].class);
|
||||
assertThat(config.out().receive(0))
|
||||
.isNotNull()
|
||||
.extracting(msg -> msg.getPayload())
|
||||
.isEqualTo(test)
|
||||
.isNotSameAs(test);
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@EnableIntegration
|
||||
public static class Config {
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow flow1() {
|
||||
return IntegrationFlows.from(in1())
|
||||
.transform(new SimpleToAvroTransformer())
|
||||
.wireTap(tapped())
|
||||
.transform(transformer())
|
||||
.channel(out())
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow flow2() {
|
||||
return IntegrationFlows.from(in2())
|
||||
.transform(new SimpleToAvroTransformer())
|
||||
.wireTap(tapped())
|
||||
.enrichHeaders(h -> h.header(AvroHeaders.TYPE, AvroTestClass2.class, true))
|
||||
.transform(transformer())
|
||||
.channel(out())
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow flow3() {
|
||||
return IntegrationFlows.from(in3())
|
||||
.transform(new SimpleToAvroTransformer())
|
||||
.wireTap(tapped())
|
||||
.enrichHeaders(h -> h.header(AvroHeaders.TYPE, AvroTestClass2.class.getName(), true))
|
||||
.transform(transformer())
|
||||
.channel(out())
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow flow4() {
|
||||
return IntegrationFlows.from(in4())
|
||||
.transform(new SimpleToAvroTransformer())
|
||||
.wireTap(tapped())
|
||||
.enrichHeaders(h -> h.header(AvroHeaders.TYPE, null, true)
|
||||
.shouldSkipNulls(false))
|
||||
.transform(transformer())
|
||||
.channel(out())
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public SimpleFromAvroTransformer transformer() {
|
||||
return new SimpleFromAvroTransformer(AvroTestClass1.class);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel in1() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel in2() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel in3() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel in4() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public PollableChannel tapped() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public PollableChannel out() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
{
|
||||
"type" : "record",
|
||||
"name" : "AvroTestClass",
|
||||
"namespace" : "org.springframework.integration.transformer",
|
||||
"fields" : [ {
|
||||
"name" : "bar",
|
||||
"type" : [ "null", "string" ]
|
||||
},
|
||||
{
|
||||
"name" : "qux",
|
||||
"type" : [ "null", "string" ],
|
||||
"default" : null
|
||||
} ]
|
||||
}
|
||||
@@ -4,6 +4,7 @@
|
||||
"https://checkstyle.org/dtds/suppressions_1_2.dtd">
|
||||
<suppressions>
|
||||
<suppress files="package-info\.java" checks=".*"/>
|
||||
<suppress files="AvroTestClass.*" checks=".*"/>
|
||||
<suppress files="[\\/]test[\\/]" checks="RequireThis"/>
|
||||
<suppress files="[\\/]test[\\/]" checks="FinalClass"/>
|
||||
<suppress files="[\\/]test[\\/]" checks="InnerTypeLast"/>
|
||||
|
||||
@@ -373,7 +373,7 @@ public class Thing1 {
|
||||
----
|
||||
====
|
||||
|
||||
NOTE: The Boon support has been deprecated since version 5.2.
|
||||
NOTE: Boon support has been deprecated since version 5.2.
|
||||
|
||||
You may wish to consider using a `FactoryBean` or a factory method to create the `JsonObjectMapper` with the required characteristics.
|
||||
The following example shows how to use such a factory:
|
||||
@@ -453,6 +453,19 @@ Starting with version 5.2, the `JsonToObjectTransformer` can be configured with
|
||||
Also this component now consults request message headers first for the presence of the `JsonHeaders.RESOLVABLE_TYPE` or `JsonHeaders.TYPE_ID` and falls back to the configured type otherwise.
|
||||
The `ObjectToJsonTransformer` now also populates a `JsonHeaders.RESOLVABLE_TYPE` header based on the request message payload for any possible downstream scenarios.
|
||||
|
||||
[[Avro-transformers]]
|
||||
===== Apache Avro Transformers
|
||||
|
||||
Version 5.2 added simple transformers to transform to/from Apache Avro.
|
||||
|
||||
They are unsophisticated in that there is no schema registry; the transformers simply use the schema embedded in the `SpecificRecord` implementation generated from the Avro schema.
|
||||
|
||||
Messages sent to the `SimpleToAvroTransformer` must have a payload that implements `SpecificRecord`; the transformer can handle multiple types.
|
||||
The `SimpleFromAvroTransformer` must be configured with a `SpecificRecord` class which is used as the default type to deserialize.
|
||||
You can also specify a SpEL expression to determine the type to deserialize.
|
||||
The default SpEL expression is `headers[avro_type]` (`AvroHeaders.TYPE`).
|
||||
If the expression returns `null`, the `defaultType` is used.
|
||||
|
||||
[[transformer-annotation]]
|
||||
==== Configuring a Transformer with Annotations
|
||||
|
||||
|
||||
@@ -116,3 +116,9 @@ See <<./webflux.adoc#webflux,WebFlux Support>> for more information.
|
||||
|
||||
The `MongoDbMessageStore` can now be configured with custom converters.
|
||||
See <<./mongodb.adoc#mongodb, MongoDB Support>> for more information.
|
||||
|
||||
[[x5.2-avro]]
|
||||
==== Avro Transformers
|
||||
|
||||
Simple Apache Avro transformers are now provided.
|
||||
See <<./transformers.adoc#avro-transformers, Avro Transformers>> for more information.
|
||||
|
||||
Reference in New Issue
Block a user