Remove dependency on web classes from new Encoder/Decoder implementation

The web dependency still need to be there since it appears that spring-rsocket requires it

spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/MessageAwareJsonDecoder.java

@@ -16,27 +16,35 @@
 
 package org.springframework.cloud.function.rsocket;
 
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.lang.reflect.Type;
 import java.util.Map;
 
+import org.reactivestreams.Publisher;
 import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
 import org.springframework.cloud.function.json.JsonMapper;
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.AbstractDecoder;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.http.codec.json.Jackson2JsonDecoder;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
+import reactor.core.publisher.Flux;
+
 /**
  *
  * @author Oleg Zhurakousky
  * @since 3.1
  *
  */
-class MessageAwareJsonDecoder extends Jackson2JsonDecoder {
+class MessageAwareJsonDecoder extends AbstractDecoder<Object> {
 
 	private final JsonMapper jsonMapper;
 
@@ -49,22 +57,25 @@ class MessageAwareJsonDecoder extends Jackson2JsonDecoder {
 		return mimeType.isCompatibleWith(MimeTypeUtils.APPLICATION_JSON);
 	}
 
-
 	@SuppressWarnings("unchecked")
 	@Override
 	public Object decode(DataBuffer dataBuffer, ResolvableType targetType,
-			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) throws DecodingException {
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints)
+			throws DecodingException {
 
-		ResolvableType type = ResolvableType.forClassWithGenerics(Map.class, String.class, Object.class);
+		ResolvableType type = ResolvableType.forClassWithGenerics(Map.class, String.class,
-		Map<String, Object> messageMap = (Map<String, Object>) super.decode(dataBuffer, type, mimeType, hints);
+				Object.class);
+		Map<String, Object> messageMap = (Map<String, Object>) doDecode(dataBuffer, type,
+				mimeType, hints);
 		if (messageMap.containsKey(FunctionRSocketUtils.PAYLOAD)) {
 			Type requestedType = FunctionTypeUtils.getGenericType(targetType.getType());
-			Object payload = this.jsonMapper.fromJson(messageMap.get(FunctionRSocketUtils.PAYLOAD), requestedType);
+			Object payload = this.jsonMapper.fromJson(
+					messageMap.get(FunctionRSocketUtils.PAYLOAD), requestedType);
 
 			if (FunctionTypeUtils.isMessage(targetType.getType())) {
-				return MessageBuilder.withPayload(payload)
+				return MessageBuilder.withPayload(payload).copyHeaders(
-					.copyHeaders((Map<String, ?>) messageMap.get(FunctionRSocketUtils.HEADERS))
+						(Map<String, ?>) messageMap.get(FunctionRSocketUtils.HEADERS))
-					.build();
+						.build();
 			}
 			else {
 				return payload;
@@ -73,6 +84,45 @@ class MessageAwareJsonDecoder extends Jackson2JsonDecoder {
 		else {
 			return messageMap;
 		}
+	}
 
+	private Object doDecode(DataBuffer dataBuffer, ResolvableType targetType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints)
+			throws DecodingException {
+
+		try {
+			byte[] data = toByteArray(dataBuffer.asInputStream());
+			return this.jsonMapper.fromJson(data, targetType.getType());
+		}
+		catch (IOException ex) {
+			throw new IllegalStateException(ex);
+		}
+		finally {
+			DataBufferUtils.release(dataBuffer);
+		}
+	}
+
+	private byte[] toByteArray(final InputStream input) throws IOException {
+		try (ByteArrayOutputStream output = new ByteArrayOutputStream()) {
+			copyLarge(input, output, new byte[2048]);
+			return output.toByteArray();
+		}
+	}
+
+	private long copyLarge(final InputStream input, final OutputStream output,
+			final byte[] buffer) throws IOException {
+		long count = 0;
+		int n;
+		while (-1 != (n = input.read(buffer))) {
+			output.write(buffer, 0, n);
+			count += n;
+		}
+		return count;
+	}
+
+	@Override
+	public Flux<Object> decode(Publisher<DataBuffer> inputStream,
+			ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) {
+		return Flux.from(inputStream).map(buffer -> decode(buffer, elementType, mimeType, hints));
 	}
 }

spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/MessageAwareJsonEncoder.java

@@ -20,29 +20,36 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
+import org.reactivestreams.Publisher;
 import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
 import org.springframework.cloud.function.json.JsonMapper;
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.AbstractEncoder;
+import org.springframework.core.codec.ByteArrayEncoder;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
-import org.springframework.http.codec.json.Jackson2JsonEncoder;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
+import reactor.core.publisher.Flux;
+
 /**
  *
+ * An extended implementation
  * @author Oleg Zhurakousky
  * @since 3.1
  *
  */
-class MessageAwareJsonEncoder extends Jackson2JsonEncoder {
+class MessageAwareJsonEncoder extends AbstractEncoder<Object> {
 
 	private final JsonMapper mapper;
 
 	private final boolean isClient;
 
+	private final ByteArrayEncoder byteArrayEncoder;
+
 	MessageAwareJsonEncoder(JsonMapper mapper) {
 		this(mapper, false);
 	}
@@ -50,6 +57,7 @@ class MessageAwareJsonEncoder extends Jackson2JsonEncoder {
 	MessageAwareJsonEncoder(JsonMapper mapper, boolean isClient) {
 		this.mapper = mapper;
 		this.isClient = isClient;
+		this.byteArrayEncoder = new ByteArrayEncoder();
 	}
 
 	@Override
@@ -81,6 +89,15 @@ class MessageAwareJsonEncoder extends Jackson2JsonEncoder {
 			}
 			value = Collections.singletonMap(FunctionRSocketUtils.PAYLOAD, value);
 		}
-		return super.encodeValue(value, bufferFactory, valueType, mimeType, hints);
+		byte[] data = this.mapper.toJson(value);
+		return this.byteArrayEncoder.encodeValue(data, bufferFactory, valueType, mimeType, hints);
+	}
+
+	@Override
+	public Flux<DataBuffer> encode(Publisher<? extends Object> inputStream,
+			DataBufferFactory bufferFactory, ResolvableType elementType,
+			MimeType mimeType, Map<String, Object> hints) {
+		return Flux.from(inputStream).map(value ->
+			encodeValue(value, bufferFactory, elementType, mimeType, hints));
 	}
 }