spring-cloudGH-422 Initial support for CloudEvents

Added initial implementation of MessageConverter
At the moment there seem that MessageConverter(s) would be the only thing needed to integrate Cloud Events with various elements of Spring

Author: olegz

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java

@@ -688,6 +688,9 @@ else if (value instanceof Mono) {
 				inputValue = this.extractValueFromOriginalValueHolderIfNecessary(value);
 			}
 
+			if (inputValue instanceof Message && !this.isInputTypeMessage()) {
+				inputValue = ((Message) inputValue).getPayload();
+			}
 			Object result = ((Function) this.target).apply(inputValue);
 
 			return value instanceof OriginalMessageHolder
@@ -985,7 +988,9 @@ private Object convertInputMessageIfNecessary(Message message, Type type) {
 					convertedInput = message;
 				}
 				else {
-					convertedInput = MessageBuilder.withPayload(convertedInput).copyHeaders(message.getHeaders()).build();
+					if (!(convertedInput instanceof Message)) {
+						convertedInput = MessageBuilder.withPayload(convertedInput).copyHeaders(message.getHeaders()).build();
+					}
 				}
 			}
 			return convertedInput;

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/CloudEventJsonMessageConverter.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2020-2020 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.cloud.function.context.config;
+
+import java.lang.reflect.Type;
+import java.util.Collection;
+import java.util.Map;
+
+import org.springframework.cloud.function.json.JsonMapper;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.MimeType;
+
+/**
+ * Implementation of {@link MessageConverter} which uses Jackson or Gson libraries to do the
+ * actual conversion via {@link JsonMapper} instance.
+ *
+ * @author Oleg Zhurakousky
+ *
+ * @since 3.1.0
+ */
+public class CloudEventJsonMessageConverter extends JsonMessageConverter {
+
+	private final JsonMapper mapper;
+
+	public CloudEventJsonMessageConverter(JsonMapper jsonMapper) {
+		super(jsonMapper, new MimeType("application", "cloudevents+json"));
+		this.mapper = jsonMapper;
+	}
+
+	@Override
+	protected Object convertFromInternal(Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
+		Type convertToType = conversionHint == null ? targetClass : (Type) conversionHint;
+		String jsonString = (String) message.getPayload();
+		Map<String, Object> mapEvent = this.mapper.fromJson(jsonString, Map.class);
+		Object payload = this.mapper.fromJson(this.mapper.toJson(mapEvent.get("data")), convertToType);
+		mapEvent.remove("data");
+		return MessageBuilder.withPayload(payload).copyHeaders(mapEvent).build();
+	}
+
+	@Override
+	protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers,
+			@Nullable Object conversionHint) {
+		throw new UnsupportedOperationException("Temporarily not supported as this converter is work in progress");
+	}
+
+}

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java

@@ -97,11 +97,11 @@ public FunctionRegistry functionCatalog(List<MessageConverter> messageConverters
 				.collect(Collectors.toList());
 
 		mcList.add(new JsonMessageConverter(jsonMapper));
+		mcList.add(new CloudEventJsonMessageConverter(jsonMapper));
 		mcList.add(new ByteArrayMessageConverter());
 		mcList.add(new StringMessageConverter());
 		mcList.add(new PrimitiveTypesFromStringMessageConverter(conversionService));
 
-
 		if (!CollectionUtils.isEmpty(mcList)) {
 			messageConverter = new SmartCompositeMessageConverter(mcList);
 		}

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java

@@ -78,14 +78,20 @@ protected Object convertFromInternal(Message<?> message, Class<?> targetClass, @
 			return message.getPayload();
 		}
 		Type convertToType = conversionHint == null ? targetClass : (Type) conversionHint;
-		try {
-			return this.jsonMapper.fromJson(message.getPayload(), convertToType);
+		if (targetClass == byte[].class && message.getPayload() instanceof String) {
+			return ((String) message.getPayload()).getBytes(StandardCharsets.UTF_8);
 		}
-		catch (Exception e) {
-			if (message.getPayload() instanceof byte[] && targetClass.isAssignableFrom(String.class)) {
-				return new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
+		else {
+			try {
+				return this.jsonMapper.fromJson(message.getPayload(), convertToType);
+			}
+			catch (Exception e) {
+				if (message.getPayload() instanceof byte[] && targetClass.isAssignableFrom(String.class)) {
+					return new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
+				}
 			}
 		}
+
 		return null;
 	}