From 745d270ddedd0493f6784db48564ecb48fee892a Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Tue, 30 May 2023 19:25:27 -0400
Subject: [PATCH] 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/aggregator-function/pom.xml          |  5 ++
 .../main/java/functions/JsonBytesToMap.java   | 80 +++++++++++++++++++
 2 files changed, 85 insertions(+)
 create mode 100644 function/payload-converter-function/src/main/java/functions/JsonBytesToMap.java

diff --git a/function/aggregator-function/pom.xml b/function/aggregator-function/pom.xml
index ce11a214..57a24503 100644
--- a/function/aggregator-function/pom.xml
+++ b/function/aggregator-function/pom.xml
@@ -15,6 +15,11 @@
     <description>Spring Native Function for Aggregator</description>
 
     <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>payload-converter-function</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.integration</groupId>
             <artifactId>spring-integration-core</artifactId>
diff --git a/function/payload-converter-function/src/main/java/functions/JsonBytesToMap.java b/function/payload-converter-function/src/main/java/functions/JsonBytesToMap.java
new file mode 100644
index 00000000..a1bdc9df
--- /dev/null
+++ b/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;
+		}
+	}
+
+}