Add type mapping to SimpleToAvroTransformer

* Fix setter names for expression Strings

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());

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) -> 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();
 	}
 

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";
 

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();

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