Handle Structured CloudEvent binary payload convertion

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

@@ -287,7 +287,7 @@ static Message<?> toCanonical(Message<?> inputMessage, MessageConverter messageC
 						.fromMessage(cloudEventMessage, Map.class);
 				canonicalizeHeaders(structuredCloudEvent, true);
 				return buildBinaryMessageFromStructuredMap(structuredCloudEvent,
-						inputMessage.getHeaders());
+						inputMessage.getHeaders(), dataContentType, messageConverter);
 			}
 		}
 		else if (StringUtils.hasText(inputContentType)) {
@@ -461,11 +461,22 @@ private static boolean isAttribute(String key) {
 	}
 
 	private static Message<?> buildBinaryMessageFromStructuredMap(Map<String, Object> structuredCloudEvent,
-			MessageHeaders originalHeaders) {
+			MessageHeaders originalHeaders, String dataContentType, MessageConverter messageConverter) {
 		Object payload = structuredCloudEvent.remove(DATA);
 		if (payload == null) {
 			payload = Collections.emptyMap();
 		}
+		// The structured envelope was fully deserialized, so 'data' is now a Java object (e.g. a Map).
+		// A binary-mode CloudEvent carries its data as serialized bytes, and downstream binary readers
+		// (e.g. the CNCF CloudEventMessageConverter) only read byte[]/String payloads. Re-serialize the
+		// data using the data content-type so the binary-mode message is well-formed.
+		if (!(payload instanceof byte[]) && !(payload instanceof String) && StringUtils.hasText(dataContentType)) {
+			Message<?> serializedData = messageConverter.toMessage(payload,
+					new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, dataContentType)));
+			if (serializedData != null) {
+				payload = serializedData.getPayload();
+			}
+		}
 
 		CloudEventMessageBuilder<?> messageBuilder = CloudEventMessageBuilder
 				.withData(payload)
@@ -477,6 +488,12 @@ private static Message<?> buildBinaryMessageFromStructuredMap(Map<String, Object
 			}
 		}
 
+		// The message has been transformed from structured-mode to binary-mode, so the content-type
+		// must reflect the data content-type.
+		if (StringUtils.hasText(dataContentType)) {
+			messageBuilder.setHeader(MessageHeaders.CONTENT_TYPE, dataContentType);
+		}
+
 		return messageBuilder.build();
 	}
 

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

@@ -20,8 +20,11 @@
 import java.text.SimpleDateFormat;
 import java.time.OffsetDateTime;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
+
+import io.cloudevents.CloudEvent;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -333,6 +336,56 @@ public void testStructuredPojoToPojoDefaultOutputAttributeProviderNoDataContentT
 		assertThat(CloudEventMessageUtils.getSource(resultMessage)).isEqualTo(URI.create("http://spring.io/"));
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test // see https://github.com/spring-cloud/spring-cloud-function/issues/1455
+	public void testStructuredToCloudEventTypedPayload() throws Exception {
+		String payload = "{\n" +
+				"    \"specversion\" : \"1.0\",\n" +
+				"    \"type\" : \"org.springframework\",\n" +
+				"    \"source\" : \"https://spring.io/\",\n" +
+				"    \"id\" : \"A234-1234-1234\",\n" +
+				"    \"datacontenttype\" : \"application/json\",\n" +
+				"    \"data\" : {\n" +
+				"        \"version\" : \"1.0\",\n" +
+				"        \"releaseName\" : \"Spring Framework\",\n" +
+				"        \"releaseDate\" : \"24-03-2004\"\n" +
+				"    }\n" +
+				"}";
+		Function<Object, Object> function = this.lookup("echoCloudEvent", TestConfiguration.class);
+
+		Message<String> inputMessage = MessageBuilder
+				.withPayload(payload)
+				.setHeader(MessageHeaders.CONTENT_TYPE, CloudEventMessageUtils.APPLICATION_CLOUDEVENTS_VALUE + "+json")
+				.build();
+
+		// structured-mode: attributes live in the body, not the headers
+		assertThat(CloudEventMessageUtils.isCloudEvent(inputMessage)).isFalse();
+
+		Message<CloudEvent> resultMessage = (Message<CloudEvent>) function.apply(inputMessage);
+
+		/*
+		 * When the target type is the CNCF io.cloudevents.CloudEvent, toCanonical converts the
+		 * structured envelope to binary-mode (data as payload, attributes as ce-* headers) and must
+		 * also reset content-type to the datacontenttype. Otherwise the CNCF CloudEventMessageConverter
+		 * re-dispatches on the stale application/cloudevents+json content-type, picks the structured
+		 * reader and fails to read the (now Map) payload as the serialized envelope.
+		 */
+		assertThat(resultMessage).isNotNull();
+		CloudEvent cloudEvent = resultMessage.getPayload();
+		assertThat(cloudEvent).isNotNull();
+		assertThat(cloudEvent.getSpecVersion().toString()).isEqualTo("1.0");
+		assertThat(cloudEvent.getType()).isEqualTo("org.springframework");
+		assertThat(cloudEvent.getSource()).isEqualTo(URI.create("https://spring.io/"));
+		assertThat(cloudEvent.getId()).isEqualTo("A234-1234-1234");
+
+		// the CloudEvent actually delivered to the user function must carry the data
+		CloudEvent received = TestConfiguration.RECEIVED_CLOUD_EVENT.get();
+		assertThat(received).isNotNull();
+		assertThat(received.getType()).isEqualTo("org.springframework");
+		assertThat(received.getData()).isNotNull();
+		assertThat(new String(received.getData().toBytes())).contains("Spring Framework");
+	}
+
 	private Function<Object, Object> lookup(String functionDefinition, Class<?>... configClass) {
 		ApplicationContext context = new SpringApplicationBuilder(configClass).run(
 				"--logging.level.org.springframework.cloud.function=DEBUG", "--spring.main.lazy-initialization=true");
@@ -342,11 +395,23 @@ private Function<Object, Object> lookup(String functionDefinition, Class<?>... c
 	@EnableAutoConfiguration
 	@Configuration
 	public static class TestConfiguration {
+
+		static final AtomicReference<CloudEvent> RECEIVED_CLOUD_EVENT =
+				new AtomicReference<>();
+
 		@Bean
 		Function<Message<Person>, Message<Person>> echo() {
 			return Function.identity();
 		}
 
+		@Bean
+		Function<Message<CloudEvent>, Message<CloudEvent>> echoCloudEvent() {
+			return message -> {
+				RECEIVED_CLOUD_EVENT.set(message.getPayload());
+				return message;
+			};
+		}
+
 		@Bean
 		Function<SpringReleaseEvent, SpringReleaseEvent> springRelease() {
 			return event -> {