From 9a2b75bae3b7670995256481e145cbb78186780c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 9 Jul 2019 14:59:01 -0400 Subject: [PATCH] Add simple Avro transformers Unsophisticated Avro transformers for `SpecificRecord` implementations. * * Fix DSL Factory - transformer was changed to ``; 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 --- build.gradle | 2 + .../transformer/AbstractTransformer.java | 2 +- .../SimpleFromAvroTransformer.java | 151 +++++++++ .../transformer/SimpleToAvroTransformer.java | 65 ++++ .../transformer/support/AvroHeaders.java | 38 +++ .../transformer/AvroTestClass1.java | 308 ++++++++++++++++++ .../transformer/AvroTestClass2.java | 308 ++++++++++++++++++ .../integration/transformer/AvroTests.java | 189 +++++++++++ .../test/resources/avro/AvroTestClass.avsc | 14 + src/checkstyle/checkstyle-suppressions.xml | 1 + src/reference/asciidoc/transformer.adoc | 15 +- src/reference/asciidoc/whats-new.adoc | 6 + 12 files changed, 1097 insertions(+), 2 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/transformer/SimpleFromAvroTransformer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/transformer/SimpleToAvroTransformer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/transformer/support/AvroHeaders.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTestClass1.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTestClass2.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTests.java create mode 100644 spring-integration-core/src/test/resources/avro/AvroTestClass.avsc diff --git a/build.gradle b/build.gradle index 4f03586b82..199a1b360d 100644 --- a/build.gradle +++ b/build.gradle @@ -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" diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractTransformer.java index ef14a6a580..142fa5579f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractTransformer.java @@ -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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/SimpleFromAvroTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/SimpleFromAvroTransformer.java new file mode 100644 index 0000000000..e2cc4d8e6e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/SimpleFromAvroTransformer.java @@ -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 defaultType; + + private final DecoderFactory decoderFactory = new DecoderFactory(); + + private Expression typeIdExpression = new FunctionExpression>( + 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 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 type = null; + Object value = this.typeIdExpression.getValue(this.evaluationContext, message); + if (value instanceof Class) { + type = (Class) value; + } + else if (value instanceof String) { + try { + type = (Class) 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); + } + } + + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/SimpleToAvroTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/SimpleToAvroTransformer.java new file mode 100644 index 0000000000..f25e56fd6a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/SimpleToAvroTransformer.java @@ -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 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(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/support/AvroHeaders.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/support/AvroHeaders.java new file mode 100644 index 0000000000..f4ac872b90 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/support/AvroHeaders.java @@ -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"; + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTestClass1.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTestClass1.java new file mode 100644 index 0000000000..3c172f258f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTestClass1.java @@ -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 ENCODER = + new BinaryMessageEncoder(MODEL$, SCHEMA$); + + private static final BinaryMessageDecoder DECODER = + new BinaryMessageDecoder(MODEL$, SCHEMA$); + + /** + * Return the BinaryMessageDecoder instance used by this class. + */ + public static BinaryMessageDecoder 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 createDecoder(SchemaStore resolver) { + return new BinaryMessageDecoder(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 newBuilder(). + */ + 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 + implements org.apache.avro.data.RecordBuilder { + + 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 + WRITER$ = (org.apache.avro.io.DatumWriter)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 + READER$ = (org.apache.avro.io.DatumReader)MODEL$.createDatumReader(SCHEMA$); + + @Override public void readExternal(java.io.ObjectInput in) + throws java.io.IOException { + READER$.read(this, SpecificData.getDecoder(in)); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTestClass2.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTestClass2.java new file mode 100644 index 0000000000..42bfc1c913 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTestClass2.java @@ -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 ENCODER = + new BinaryMessageEncoder(MODEL$, SCHEMA$); + + private static final BinaryMessageDecoder DECODER = + new BinaryMessageDecoder(MODEL$, SCHEMA$); + + /** + * Return the BinaryMessageDecoder instance used by this class. + */ + public static BinaryMessageDecoder 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 createDecoder(SchemaStore resolver) { + return new BinaryMessageDecoder(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 newBuilder(). + */ + 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 + implements org.apache.avro.data.RecordBuilder { + + 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 + WRITER$ = (org.apache.avro.io.DatumWriter)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 + READER$ = (org.apache.avro.io.DatumReader)MODEL$.createDatumReader(SCHEMA$); + + @Override public void readExternal(java.io.ObjectInput in) + throws java.io.IOException { + READER$.read(this, SpecificData.getDecoder(in)); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTests.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTests.java new file mode 100644 index 0000000000..273b36f158 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTests.java @@ -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(); + } + + } + +} diff --git a/spring-integration-core/src/test/resources/avro/AvroTestClass.avsc b/spring-integration-core/src/test/resources/avro/AvroTestClass.avsc new file mode 100644 index 0000000000..58ba9084d2 --- /dev/null +++ b/spring-integration-core/src/test/resources/avro/AvroTestClass.avsc @@ -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 + } ] +} diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index 0e100dbddb..8414a71994 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -4,6 +4,7 @@ "https://checkstyle.org/dtds/suppressions_1_2.dtd"> + diff --git a/src/reference/asciidoc/transformer.adoc b/src/reference/asciidoc/transformer.adoc index 38632c7a8d..24ead56a73 100644 --- a/src/reference/asciidoc/transformer.adoc +++ b/src/reference/asciidoc/transformer.adoc @@ -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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index b72c051f8e..44145b0abc 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -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.