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 index 6e304ed960..3213893306 100644 --- 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 @@ -83,10 +83,6 @@ public class SimpleFromAvroTransformer extends AbstractTransformer implements Be return this; } - private void assertExpressionNotNull(Object expression) { - Assert.notNull(expression, "'expression' must not be null"); - } - /** * Set the expression to evaluate against the message to determine the type id. * Default {@code headers['avro_type']}. @@ -114,11 +110,15 @@ public class SimpleFromAvroTransformer extends AbstractTransformer implements Be * Default {@code headers['avro_type']}. * @param expression the expression. */ - public void setTypeExpression(String expression) { + public void setTypeExpressionString(String expression) { assertExpressionNotNull(expression); this.typeIdExpression = EXPRESSION_PARSER.parseExpression(expression); } + private void assertExpressionNotNull(Object expression) { + Assert.notNull(expression, "'expression' must not be null"); + } + @Override protected void onInit() { this.evaluationContext = IntegrationContextUtils.getEvaluationContext(getBeanFactory()); 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 index f25e56fd6a..ad98f4fd89 100644 --- 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 @@ -26,6 +26,10 @@ import org.apache.avro.io.EncoderFactory; import org.apache.avro.specific.SpecificDatumWriter; import org.apache.avro.specific.SpecificRecord; +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; @@ -41,6 +45,64 @@ public class SimpleToAvroTransformer extends AbstractTransformer { private final EncoderFactory encoderFactory = new EncoderFactory(); + private Expression typeIdExpression = + new FunctionExpression>((message) -> message.getPayload().getClass()); + + private EvaluationContext evaluationContext; + + /** + * Set the expression to evaluate against the message to determine the value + * for the {@link AvroHeaders#TYPE} header. + * @param expression the expression. + * @return the transformer + */ + public SimpleToAvroTransformer typeExpression(Expression expression) { + assertExpressionNotNull(expression); + this.typeIdExpression = expression; + return this; + } + + /** + * Set the expression to evaluate against the message to determine the value + * for the {@link AvroHeaders#TYPE} header. + * @param expression the expression. + * @return the transformer + */ + public SimpleToAvroTransformer typeExpression(String expression) { + assertExpressionNotNull(expression); + this.typeIdExpression = EXPRESSION_PARSER.parseExpression(expression); + return this; + } + + /** + * Set the expression to evaluate against the message to determine the value + * for the {@link AvroHeaders#TYPE} header. + * @param expression the expression. + */ + public void setTypeExpression(Expression expression) { + assertExpressionNotNull(expression); + this.typeIdExpression = expression; + } + + /** + * Set the expression to evaluate against the message to determine the value + * for the {@link AvroHeaders#TYPE} header. + * @param expression the expression. + */ + public void setTypeExpressionString(String expression) { + assertExpressionNotNull(expression); + this.typeIdExpression = EXPRESSION_PARSER.parseExpression(expression); + } + + private void assertExpressionNotNull(Object expression) { + Assert.notNull(expression, "'expression' must not be null"); + } + + @Override + protected void onInit() { + this.evaluationContext = IntegrationContextUtils.getEvaluationContext(getBeanFactory()); + } + @Override protected Object doTransform(Message message) { Assert.state(message.getPayload() instanceof SpecificRecord, @@ -58,7 +120,7 @@ public class SimpleToAvroTransformer extends AbstractTransformer { } return getMessageBuilderFactory().withPayload(out.toByteArray()) .copyHeaders(message.getHeaders()) - .setHeader(AvroHeaders.TYPE, specific.getClass()) + .setHeader(AvroHeaders.TYPE, this.typeIdExpression.getValue(this.evaluationContext, message)) .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 index 15d349b76b..b4cf4975a5 100644 --- 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 @@ -36,7 +36,8 @@ public final class AvroHeaders { public static final String PREFIX = "avro_"; /** - * The {@code SpecificRecord} type. + * The {@code SpecificRecord} type. By default it's the fully qualified + * SpecificRecord type but can be a key that is mapped to the actual type. */ public static final String TYPE = PREFIX + "type"; 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 index 273b36f158..245fd560ca 100644 --- 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 @@ -29,6 +29,7 @@ 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.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @@ -49,11 +50,13 @@ public class AvroTests { .isNotNull() .extracting(msg -> msg.getPayload()) .isInstanceOf(byte[].class); - assertThat(config.out().receive(0)) + Message received = config.out().receive(0); + assertThat(received) .isNotNull() .extracting(msg -> msg.getPayload()) .isEqualTo(test) .isNotSameAs(test); + assertThat(received.getHeaders().get("flow")).isEqualTo("flow1"); } @Test @@ -64,11 +67,13 @@ public class AvroTests { .isNotNull() .extracting(msg -> msg.getPayload()) .isInstanceOf(byte[].class); - assertThat(config.out().receive(0)) + Message received = config.out().receive(0); + assertThat(received) .isNotNull() .extracting(msg -> msg.getPayload()) .isNotEqualTo(test) .isInstanceOf(AvroTestClass2.class); + assertThat(received.getHeaders().get("flow")).isEqualTo("flow2"); } @Test @@ -79,11 +84,13 @@ public class AvroTests { .isNotNull() .extracting(msg -> msg.getPayload()) .isInstanceOf(byte[].class); - assertThat(config.out().receive(0)) + Message received = config.out().receive(0); + assertThat(received) .isNotNull() .extracting(msg -> msg.getPayload()) .isNotEqualTo(test) .isInstanceOf(AvroTestClass2.class); + assertThat(received.getHeaders().get("flow")).isEqualTo("flow3"); } @Test @@ -94,11 +101,47 @@ public class AvroTests { .isNotNull() .extracting(msg -> msg.getPayload()) .isInstanceOf(byte[].class); - assertThat(config.out().receive(0)) + Message received = config.out().receive(0); + assertThat(received) .isNotNull() .extracting(msg -> msg.getPayload()) .isEqualTo(test) .isNotSameAs(test); + assertThat(received.getHeaders().get("flow")).isEqualTo("flow4"); + } + + @Test + void testTransformWithTypeMappingExpressions(@Autowired Config config) { + AvroTestClass1 test = new AvroTestClass1("baz", "fiz"); + config.in5().send(new GenericMessage<>(test)); + assertThat(config.tapped().receive(0)) + .isNotNull() + .extracting(msg -> msg.getPayload()) + .isInstanceOf(byte[].class); + Message received = config.out().receive(0); + assertThat(received) + .isNotNull() + .extracting(msg -> msg.getPayload()) + .isNotEqualTo(test) + .isInstanceOf(AvroTestClass2.class); + assertThat(received.getHeaders().get("flow")).isEqualTo("flow5"); + } + + @Test + void testTransformersFallbackWhenNoTypeMappingMatch(@Autowired Config config) { + AvroTestClass1 test = new AvroTestClass1("baz", "fiz"); + config.in6().send(new GenericMessage<>(test)); + assertThat(config.tapped().receive(0)) + .isNotNull() + .extracting(msg -> msg.getPayload()) + .isInstanceOf(byte[].class); + Message received = config.out().receive(0); + assertThat(received) + .isNotNull() + .extracting(msg -> msg.getPayload()) + .isEqualTo(test) + .isNotSameAs(test); + assertThat(received.getHeaders().get("flow")).isEqualTo("flow6"); } @Configuration @@ -110,7 +153,8 @@ public class AvroTests { return IntegrationFlows.from(in1()) .transform(new SimpleToAvroTransformer()) .wireTap(tapped()) - .transform(transformer()) + .transform(fromTransformer()) + .enrichHeaders(h -> h.header("flow", "flow1")) .channel(out()) .get(); } @@ -121,7 +165,8 @@ public class AvroTests { .transform(new SimpleToAvroTransformer()) .wireTap(tapped()) .enrichHeaders(h -> h.header(AvroHeaders.TYPE, AvroTestClass2.class, true)) - .transform(transformer()) + .transform(fromTransformer()) + .enrichHeaders(h -> h.header("flow", "flow2")) .channel(out()) .get(); } @@ -132,7 +177,8 @@ public class AvroTests { .transform(new SimpleToAvroTransformer()) .wireTap(tapped()) .enrichHeaders(h -> h.header(AvroHeaders.TYPE, AvroTestClass2.class.getName(), true)) - .transform(transformer()) + .transform(fromTransformer()) + .enrichHeaders(h -> h.header("flow", "flow3")) .channel(out()) .get(); } @@ -144,13 +190,40 @@ public class AvroTests { .wireTap(tapped()) .enrichHeaders(h -> h.header(AvroHeaders.TYPE, null, true) .shouldSkipNulls(false)) - .transform(transformer()) + .transform(fromTransformer()) + .enrichHeaders(h -> h.header("flow", "flow4")) .channel(out()) .get(); } @Bean - public SimpleFromAvroTransformer transformer() { + public IntegrationFlow flow5() { + return IntegrationFlows.from(in5()) + .transform(new SimpleToAvroTransformer().typeExpression("'avroTest'")) + .wireTap(tapped()) + .transform(new SimpleFromAvroTransformer(AvroTestClass1.class) + .typeExpression("'avroTest' == headers[avro_type] ? '" + + AvroTestClass2.class.getName() + "' : null")) + .enrichHeaders(h -> h.header("flow", "flow5")) + .channel(out()) + .get(); + } + + @Bean + public IntegrationFlow flow6() { + return IntegrationFlows.from(in6()) + .transform(new SimpleToAvroTransformer().typeExpression("'wontFindThisHeader'")) + .wireTap(tapped()) + .transform(new SimpleFromAvroTransformer(AvroTestClass1.class) + .typeExpression("'avroTest' == headers[avro_type] ? '" + + AvroTestClass2.class.getName() + "' : null")) + .enrichHeaders(h -> h.header("flow", "flow6")) + .channel(out()) + .get(); + } + + @Bean + public SimpleFromAvroTransformer fromTransformer() { return new SimpleFromAvroTransformer(AvroTestClass1.class); } @@ -174,6 +247,16 @@ public class AvroTests { return new DirectChannel(); } + @Bean + public DirectChannel in5() { + return new DirectChannel(); + } + + @Bean + public DirectChannel in6() { + return new DirectChannel(); + } + @Bean public PollableChannel tapped() { return new QueueChannel(); diff --git a/src/reference/asciidoc/transformer.adoc b/src/reference/asciidoc/transformer.adoc index 24ead56a73..aaf411e256 100644 --- a/src/reference/asciidoc/transformer.adoc +++ b/src/reference/asciidoc/transformer.adoc @@ -462,10 +462,13 @@ They are unsophisticated in that there is no schema registry; the transformers s 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`). +You can also specify a SpEL expression to determine the type to deserialize using the `setTypeExpression` method. +The default SpEL expression is `headers[avro_type]` (`AvroHeaders.TYPE`) which, by default, is populated by the `SimpleToAvroTransformer` with the fully qualified class name of the source class. If the expression returns `null`, the `defaultType` is used. +The `SimpleToAvroTransformer` also has a `setTypeExpression` method. +This allows decoupling of the producer and consumer where the sender can set the header to some token representing the type and the consumer then maps that token to a type. + [[transformer-annotation]] ==== Configuring a Transformer with Annotations