GH-441: Add JsonBytesToMap function (#461)

Add `JsonBytesToMap` function to allow conversion between byte[] and JSON Map.

Fixes https://github.com/spring-cloud/stream-applications/issues/441

* Introduce a `JsonBytesToMap` as a part of a `payload-converter-function` module
which is auto-discovered by Spring Cloud Function scanning algorithm - the `functions` package.
* Add a `payload-converter-function` as dependency into an `aggregator-function`
* Compose `jsonBytesToMap|aggregatorFunction` for the `aggregator-processor`
* Verify a `JsonBytesToMap` function in action with an `AggregatorProcessorTests`
* Mentioned such a payload conversion in the `aggregator-processor` README

function/payload-converter-function/src/main/java/functions/JsonBytesToMap.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2023-2023 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 functions;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.springframework.core.log.LogAccessor;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * The {@link Function} to deserialize {@code byte[]} payload into a Map
+ * if {@link MessageHeaders#CONTENT_TYPE} header is JSON.
+ * Otherwise, the message is returned as is.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class JsonBytesToMap implements Function<Message<?>, Message<?>> {
+
+	private static final LogAccessor logger = new LogAccessor(JsonBytesToMap.class);
+
+	private final ObjectMapper objectMapper;
+
+	public JsonBytesToMap(ObjectMapper objectMapper) {
+		this.objectMapper = objectMapper;
+	}
+
+	@Override
+	public Message<?> apply(Message<?> message) {
+		if (message.getPayload() instanceof byte[] payload) {
+			MessageHeaders headers = message.getHeaders();
+			String contentType =
+					headers.containsKey(MessageHeaders.CONTENT_TYPE)
+							? headers.get(MessageHeaders.CONTENT_TYPE).toString()
+							: MimeTypeUtils.APPLICATION_JSON_VALUE;
+
+			if (contentType.contains("json")) {
+				message = MessageBuilder.withPayload(payloadToMapIfCan(payload))
+						.copyHeaders(message.getHeaders())
+						.build();
+			}
+		}
+
+		return message;
+	}
+
+	private Object payloadToMapIfCan(byte[] payload) {
+		try {
+			return this.objectMapper.readValue(payload, Map.class);
+		}
+		catch (IOException ex) {
+			logger.trace(ex, "Cannot deserialize to Map");
+			// Was not able to construct the map from byte[] -- returning as is
+			return payload;
+		}
+	}
+
+}