Added decode methods

Author: matiasah

src/main/java/org/springframework/data/web/ProjectingJackson2JsonDecoder.java

@@ -15,23 +15,31 @@
  */
 package org.springframework.data.web;
 
+import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.jayway.jsonpath.spi.json.JacksonJsonProvider;
 import com.jayway.jsonpath.spi.mapper.JacksonMappingProvider;
+import org.reactivestreams.Publisher;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.codec.DecodingException;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.http.codec.json.Jackson2JsonDecoder;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ConcurrentReferenceHashMap;
 import org.springframework.util.MimeType;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
+import java.math.BigDecimal;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -129,4 +137,33 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 		}
 	}
 
+	@Override
+	public Flux<Object> decode(Publisher<DataBuffer> input, ResolvableType elementType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		ObjectMapper mapper = selectObjectMapper(elementType, mimeType);
+		if (mapper == null) {
+			throw new IllegalStateException("No ObjectMapper for " + elementType);
+		}
+
+		Flux<DataBuffer> processed = processInput(input, elementType, mimeType, hints);
+
+		return DataBufferUtils.join(processed, this.getMaxInMemorySize())
+				.flatMap(dataBuffer -> Mono.just(decode(dataBuffer, elementType, mimeType, hints)))
+				.expand(object -> {
+					if (object instanceof Iterable) {
+						return Flux.fromIterable((Iterable) object);
+					}
+					return Flux.just(object);
+				});
+	}
+
+	@Override
+	public Object decode(DataBuffer dataBuffer, ResolvableType targetType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) throws DecodingException {
+
+		return projectionFactory.createProjection(ResolvableType.forType(targetType.getType()).resolve(Object.class),
+				dataBuffer.asInputStream());
+	}
+
 }