GH-1294 Remove usage of TARGET_PROTOCOL header

Resolves #1294

Author: olegz

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventMessageUtils.java

@@ -21,6 +21,7 @@
 import java.time.OffsetDateTime;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -361,8 +362,22 @@ static String determinePrefixToUse(Map<String, Object> messageHeaders) {
 		return determinePrefixToUse(messageHeaders, false);
 	}
 
+	static String extractTargetProtocol(Map<String, Object> messageHeaders) {
+		Iterator<String> keyIterator = messageHeaders.keySet().iterator();
+		for (; keyIterator.hasNext();) {
+			String key = keyIterator.next();
+			if (key.startsWith("kafka_")) {
+				return Protocols.KAFKA;
+			}
+			else if (key.startsWith("amqp")) {
+				return Protocols.AMQP;
+			}
+		}
+		return null;
+	}
+
 	static String determinePrefixToUse(Map<String, Object> messageHeaders, boolean strict) {
-		String targetProtocol = (String) messageHeaders.get(MessageUtils.TARGET_PROTOCOL);
+		String targetProtocol = extractTargetProtocol(messageHeaders);
 		String prefix = determinePrefixToUse(targetProtocol);
 		if (StringUtils.hasText(prefix) && (strict || StringUtils.hasText((String) messageHeaders.get(prefix + _SPECVERSION)))) {
 			return prefix;

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventsFunctionInvocationHelper.java

@@ -71,7 +71,7 @@ public class CloudEventsFunctionInvocationHelper implements FunctionInvocationHe
 
 	@Override
 	public boolean isRetainOutputAsMessage(Message<?> message) {
-		return message.getHeaders().containsKey(MessageUtils.TARGET_PROTOCOL) || (message.getHeaders().containsKey(MessageUtils.MESSAGE_TYPE)
+		return (message.getHeaders().containsKey(MessageUtils.MESSAGE_TYPE)
 			&& message.getHeaders().get(MessageUtils.MESSAGE_TYPE).equals(CloudEventMessageUtils.CLOUDEVENT_VALUE));
 	}
 
@@ -101,8 +101,8 @@ public Message<?> postProcessResult(Object result, Message<?> input) {
 		if (input != null) {
 			targetPrefix = CloudEventMessageUtils.determinePrefixToUse(input.getHeaders(), true);
 		}
-		else if (result instanceof Message) {
-			targetPrefix = CloudEventMessageUtils.determinePrefixToUse(((Message<?>) result).getHeaders(), true);
+		else if (result instanceof Message resultMessage) {
+			targetPrefix = CloudEventMessageUtils.determinePrefixToUse(resultMessage.getHeaders(), true);
 		}
 
 		Assert.hasText(targetPrefix, "Unable to determine prefix for Cloud Event atttributes, "

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/message/MessageUtils.java

@@ -31,12 +31,6 @@ public abstract class MessageUtils {
 	 * Value for 'message-type' typically use as header key.
 	 */
 	public static String MESSAGE_TYPE = "message-type";
-
-	/**
-	 * Value for 'target-protocol' typically use as header key.
-	 */
-	public static String TARGET_PROTOCOL = "target-protocol";
-
 	/**
 	 * Value for 'target-protocol' typically use as header key.
 	 */

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/cloudevent/CloudEventFunctionTests.java

@@ -29,7 +29,6 @@
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.function.context.FunctionCatalog;
-import org.springframework.cloud.function.context.message.MessageUtils;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -99,7 +98,7 @@ public void testBinaryPojoToPojoDefaultOutputHeaderProviderImperative() {
 			.setId(id)
 			.setSource("https://spring.io/")
 			.setType("org.springframework")
-			.setHeader(MessageUtils.TARGET_PROTOCOL, CloudEventMessageUtils.Protocols.KAFKA)
+			.setHeader("kafka_foo", "blah")
 			.build(CloudEventMessageUtils.AMQP_ATTR_PREFIX);
 
 		assertThat(CloudEventMessageUtils.isCloudEvent(inputMessage)).isTrue();
@@ -136,7 +135,7 @@ public void testBinaryPojoToPojoDefaultOutputHeaderProviderReactive() {
 			.setId(id)
 			.setSource("https://spring.io/")
 			.setType("org.springframework")
-			.setHeader(MessageUtils.TARGET_PROTOCOL, CloudEventMessageUtils.Protocols.KAFKA)
+			.setHeader("kafka_foo", "blah")
 			.build(CloudEventMessageUtils.AMQP_ATTR_PREFIX);
 
 		assertThat(CloudEventMessageUtils.isCloudEvent(inputMessage)).isTrue();
@@ -173,7 +172,7 @@ public void testBinaryPojoToPojoDefaultOutputHeaderProviderReactiveMono() {
 			.setId(id)
 			.setSource("https://spring.io/")
 			.setType("org.springframework")
-			.setHeader(MessageUtils.TARGET_PROTOCOL, CloudEventMessageUtils.Protocols.KAFKA)
+			.setHeader("kafka_foo", "blah")
 			.build(CloudEventMessageUtils.AMQP_ATTR_PREFIX);
 
 		assertThat(CloudEventMessageUtils.isCloudEvent(inputMessage)).isTrue();

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/cloudevent/CloudEventMessageUtilsAndBuilderTests.java

@@ -47,7 +47,7 @@ public void testHeaderKeyInsensitivity() {
 	@Test// see https://github.com/spring-cloud/spring-cloud-function/issues/680
 	public void testProperAttributeExtractionRegardlessOfTargetProtocol() {
 		Message<String> ceMessage = CloudEventMessageBuilder.withData("foo").build();
-		ceMessage = MessageBuilder.fromMessage(ceMessage).setHeader("target-protocol", "kafka").build();
+		ceMessage = MessageBuilder.fromMessage(ceMessage).setHeader("kafka_foo", "blah").build();
 
 		String prefix = CloudEventMessageUtils.determinePrefixToUse(ceMessage.getHeaders());
 		assertThat(prefix).isEqualTo("ce-");