diff --git a/spring-web-reactive/build.gradle b/spring-web-reactive/build.gradle index 07501e6280..2e6c3722eb 100644 --- a/spring-web-reactive/build.gradle +++ b/spring-web-reactive/build.gradle @@ -24,6 +24,7 @@ dependencies { compile "org.reactivestreams:reactive-streams:1.0.0" compile "io.projectreactor:reactor-stream:2.0.5.RELEASE" compile "commons-logging:commons-logging:1.2" + compile "com.fasterxml.jackson.core:jackson-databind:2.6.1" optional "io.reactivex:rxnetty:0.5.0-SNAPSHOT" optional "io.reactivex:rxjava-reactive-streams:1.0.1" diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/CodecException.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/CodecException.java new file mode 100644 index 0000000000..3ed102ea04 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/CodecException.java @@ -0,0 +1,36 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec; + +import org.springframework.core.NestedRuntimeException; + +/** + * Codec related exception, usually used as a wrapper for a cause exception. + * + * @author Sebastien Deleuze + */ +public class CodecException extends NestedRuntimeException { + + public CodecException(String msg, Throwable cause) { + super(msg, cause); + } + + public CodecException(String msg) { + super(msg); + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/ByteToMessageDecoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/ByteToMessageDecoder.java new file mode 100644 index 0000000000..888bb69571 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/ByteToMessageDecoder.java @@ -0,0 +1,56 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.nio.ByteBuffer; + +import org.reactivestreams.Publisher; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.encoder.MessageToByteEncoder; + +/** + * Decode from a bytes stream to a message stream. + * + * @author Sebastien Deleuze + * @see MessageToByteEncoder + */ +public interface ByteToMessageDecoder { + + /** + * Indicate whether the given type and media type can be processed by this decoder. + * @param type the (potentially generic) type to ultimately decode to. + * Could be different from {@code T} type. + * @param mediaType the media type to decode from. + * Typically the value of a {@code Content-Type} header for HTTP request. + * @param hints Additional information about how to do decode, optional. + * @return {@code true} if decodable; {@code false} otherwise + */ + boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints); + + /** + * Decode a bytes stream to a message stream. + * @param inputStream the input stream that represent the whole object to decode. + * @param type the (potentially generic) type to ultimately decode to. + * Could be different from {@code T} type. + * @param hints Additional information about how to do decode, optional. + * @return the decoded message stream + */ + Publisher decode(Publisher inputStream, ResolvableType type, MediaType mediaType, Object... hints); + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoder.java new file mode 100644 index 0000000000..150bf33a6e --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoder.java @@ -0,0 +1,70 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.io.IOException; +import java.nio.ByteBuffer; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.ObjectReader; +import org.reactivestreams.Publisher; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.CodecException; +import org.springframework.reactive.codec.encoder.JacksonJsonEncoder; +import org.springframework.reactive.io.ByteBufferInputStream; + +/** + * Decode from a bytes stream of JSON objects to a stream of {@code Object} (POJO). + * + * @author Sebastien Deleuze + * @see JacksonJsonEncoder + */ +public class JacksonJsonDecoder implements ByteToMessageDecoder { + + private final ObjectMapper mapper; + + public JacksonJsonDecoder() { + this(new ObjectMapper()); + } + + public JacksonJsonDecoder(ObjectMapper mapper) { + this.mapper = mapper; + } + + @Override + public boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints) { + return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON); + } + + @Override + public Publisher decode(Publisher inputStream, ResolvableType type, MediaType mediaType, Object... hints) { + ObjectReader reader = mapper.readerFor(type.getRawClass()); + return Streams.wrap(inputStream) + .map(chunk -> { + try { + return reader.readValue(new ByteBufferInputStream(chunk)); + } + catch (IOException e) { + throw new CodecException("Error while reading the data", e); + } + }); + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/Jaxb2Decoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/Jaxb2Decoder.java new file mode 100644 index 0000000000..462baab1c9 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/Jaxb2Decoder.java @@ -0,0 +1,132 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.nio.ByteBuffer; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import javax.xml.bind.JAXBContext; +import javax.xml.bind.JAXBElement; +import javax.xml.bind.JAXBException; +import javax.xml.bind.UnmarshalException; +import javax.xml.bind.Unmarshaller; +import javax.xml.bind.annotation.XmlRootElement; +import javax.xml.transform.Source; +import javax.xml.transform.sax.SAXSource; +import javax.xml.transform.stream.StreamSource; + +import org.reactivestreams.Publisher; +import org.xml.sax.InputSource; +import org.xml.sax.SAXException; +import org.xml.sax.XMLReader; +import org.xml.sax.helpers.XMLReaderFactory; +import reactor.io.buffer.Buffer; +import reactor.rx.Stream; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.CodecException; +import org.springframework.reactive.codec.encoder.Jaxb2Encoder; +import org.springframework.reactive.io.ByteArrayPublisherInputStream; +import org.springframework.util.Assert; + +/** + * Decode from a bytes stream of XML elements to a stream of {@code Object} (POJO). + * + * @author Sebastien Deleuze + * @see Jaxb2Encoder + */ +public class Jaxb2Decoder implements ByteToMessageDecoder { + + private final ConcurrentMap, JAXBContext> jaxbContexts = new ConcurrentHashMap, JAXBContext>(64); + + + @Override + public boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints) { + return mediaType.isCompatibleWith(MediaType.APPLICATION_XML) || mediaType.isCompatibleWith(MediaType.TEXT_XML); + } + + @Override + public Publisher decode(Publisher inputStream, ResolvableType type, MediaType mediaType, Object... hints) { + Stream stream = Streams.wrap(inputStream).map(chunk -> new Buffer(chunk).asBytes()); + Class outputClass = type.getRawClass(); + try { + Source source = processSource(new StreamSource(new ByteArrayPublisherInputStream(stream))); + Unmarshaller unmarshaller = createUnmarshaller(outputClass); + if (outputClass.isAnnotationPresent(XmlRootElement.class)) { + return Streams.just(unmarshaller.unmarshal(source)); + } + else { + JAXBElement jaxbElement = unmarshaller.unmarshal(source, outputClass); + return Streams.just(jaxbElement.getValue()); + } + } + catch (UnmarshalException ex) { + return Streams.fail(new CodecException("Could not unmarshal to [" + outputClass + "]: " + ex.getMessage(), ex)); + } + catch (JAXBException ex) { + return Streams.fail(new CodecException("Could not instantiate JAXBContext: " + ex.getMessage(), ex)); + } + } + + protected Source processSource(Source source) { + if (source instanceof StreamSource) { + StreamSource streamSource = (StreamSource) source; + InputSource inputSource = new InputSource(streamSource.getInputStream()); + try { + XMLReader xmlReader = XMLReaderFactory.createXMLReader(); + return new SAXSource(xmlReader, inputSource); + } + catch (SAXException ex) { + throw new CodecException("Error while processing the source", ex); + } + } + else { + return source; + } + } + + protected final Unmarshaller createUnmarshaller(Class clazz) throws JAXBException { + try { + JAXBContext jaxbContext = getJaxbContext(clazz); + Unmarshaller unmarshaller = jaxbContext.createUnmarshaller(); + return unmarshaller; + } + catch (JAXBException ex) { + throw new CodecException( + "Could not create Unmarshaller for class [" + clazz + "]: " + ex.getMessage(), ex); + } + } + + protected final JAXBContext getJaxbContext(Class clazz) { + Assert.notNull(clazz, "'clazz' must not be null"); + JAXBContext jaxbContext = this.jaxbContexts.get(clazz); + if (jaxbContext == null) { + try { + jaxbContext = JAXBContext.newInstance(clazz); + this.jaxbContexts.putIfAbsent(clazz, jaxbContext); + } + catch (JAXBException ex) { + throw new CodecException( + "Could not instantiate JAXBContext for class [" + clazz + "]: " + ex.getMessage(), ex); + } + } + return jaxbContext; + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JsonObjectDecoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JsonObjectDecoder.java new file mode 100644 index 0000000000..6a1f5db21d --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JsonObjectDecoder.java @@ -0,0 +1,263 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufUtil; +import io.netty.buffer.Unpooled; +import org.reactivestreams.Publisher; +import reactor.fn.Function; +import reactor.rx.Streams; +import rx.Observable; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.encoder.JsonObjectEncoder; + +/** + * Decode an arbitrary split byte stream representing JSON objects to a bye stream + * where each chunk is a well-formed JSON object. + * + * If {@code Hints.STREAM_ARRAY_ELEMENTS} is enabled, each element of top level JSON array + * will be streamed as an individual JSON object. + * + * This class does not do any real parsing or validation. A sequence of bytes is considered a JSON object/array + * if it contains a matching number of opening and closing braces/brackets. + * + * Based on Netty {@code JsonObjectDecoder} + * + * @author Sebastien Deleuze + * @see JsonObjectEncoder + */ +public class JsonObjectDecoder implements ByteToMessageDecoder { + + private static final int ST_CORRUPTED = -1; + private static final int ST_INIT = 0; + private static final int ST_DECODING_NORMAL = 1; + private static final int ST_DECODING_ARRAY_STREAM = 2; + + private final int maxObjectLength; + private final boolean streamArrayElements; + + public JsonObjectDecoder() { + // 1 MB + this(1024 * 1024); + } + + public JsonObjectDecoder(int maxObjectLength) { + this(maxObjectLength, true); + } + + public JsonObjectDecoder(boolean streamArrayElements) { + this(1024 * 1024, streamArrayElements); + } + + /** + * @param maxObjectLength maximum number of bytes a JSON object/array may use (including braces and all). + * Objects exceeding this length are dropped and an {@link IllegalStateException} + * is thrown. + * @param streamArrayElements if set to true and the "top level" JSON object is an array, each of its entries + * is passed through the pipeline individually and immediately after it was fully + * received, allowing for arrays with "infinitely" many elements. + * + */ + public JsonObjectDecoder(int maxObjectLength, boolean streamArrayElements) { + if (maxObjectLength < 1) { + throw new IllegalArgumentException("maxObjectLength must be a positive int"); + } + this.maxObjectLength = maxObjectLength; + this.streamArrayElements = streamArrayElements; + } + + + @Override + public boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints) { + return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON) && + (Observable.class.isAssignableFrom(type.getRawClass()) || Publisher.class.isAssignableFrom(type.getRawClass())); + } + + @Override + public Publisher decode(Publisher inputStream, ResolvableType type, MediaType mediaType, Object... hints) { + + return Streams.wrap(inputStream).flatMap(new Function>() { + + int openBraces; + int idx; + int state; + boolean insideString; + ByteBuf in; + Integer wrtIdx; + + @Override + public Publisher apply(ByteBuffer b) { + List chunks = new ArrayList<>(); + + if (in == null) { + in = Unpooled.copiedBuffer(b); + wrtIdx = in.writerIndex(); + } + else { + in = Unpooled.copiedBuffer(in, Unpooled.copiedBuffer(b)); + wrtIdx = in.writerIndex(); + } + if (state == ST_CORRUPTED) { + in.skipBytes(in.readableBytes()); + return Streams.fail(new IllegalStateException("Corrupted stream")); + } + + if (wrtIdx > maxObjectLength) { + // buffer size exceeded maxObjectLength; discarding the complete buffer. + in.skipBytes(in.readableBytes()); + reset(); + return Streams.fail(new IllegalStateException( + "object length exceeds " + maxObjectLength + ": " + + wrtIdx + + " bytes discarded")); + } + + for (/* use current idx */; idx < wrtIdx; idx++) { + byte c = in.getByte(idx); + if (state == ST_DECODING_NORMAL) { + decodeByte(c, in, idx); + + // All opening braces/brackets have been closed. That's enough to conclude + // that the JSON object/array is complete. + if (openBraces == 0) { + ByteBuf json = extractObject(in, in.readerIndex(), + idx + 1 - in.readerIndex()); + if (json != null) { + chunks.add(json.nioBuffer()); + } + + // The JSON object/array was extracted => discard the bytes from + // the input buffer. + in.readerIndex(idx + 1); + // Reset the object state to get ready for the next JSON object/text + // coming along the byte stream. + reset(); + } + } + else if (state == ST_DECODING_ARRAY_STREAM) { + decodeByte(c, in, idx); + + if (!insideString && (openBraces == 1 && c == ',' || + openBraces == 0 && c == ']')) { + // skip leading spaces. No range check is needed and the loop will terminate + // because the byte at position idx is not a whitespace. + for (int i = in.readerIndex(); Character.isWhitespace(in.getByte(i)); i++) { + in.skipBytes(1); + } + + // skip trailing spaces. + int idxNoSpaces = idx - 1; + while (idxNoSpaces >= in.readerIndex() && + Character.isWhitespace(in.getByte(idxNoSpaces))) { + idxNoSpaces--; + } + + ByteBuf json = extractObject(in, in.readerIndex(), + idxNoSpaces + 1 - in.readerIndex()); + if (json != null) { + chunks.add(json.nioBuffer()); + } + + in.readerIndex(idx + 1); + + if (c == ']') { + reset(); + } + } + // JSON object/array detected. Accumulate bytes until all braces/brackets are closed. + } + else if (c == '{' || c == '[') { + initDecoding(c, streamArrayElements); + + if (state == ST_DECODING_ARRAY_STREAM) { + // Discard the array bracket + in.skipBytes(1); + } + // Discard leading spaces in front of a JSON object/array. + } + else if (Character.isWhitespace(c)) { + in.skipBytes(1); + } + else { + state = ST_CORRUPTED; + return Streams.fail(new IllegalStateException( + "invalid JSON received at byte position " + idx + + ": " + ByteBufUtil.hexDump(in))); + } + } + + if (in.readableBytes() == 0) { + idx = 0; + } + return Streams.from(chunks); + } + + /** + * Override this method if you want to filter the json objects/arrays that get passed through the pipeline. + */ + @SuppressWarnings("UnusedParameters") + protected ByteBuf extractObject(ByteBuf buffer, int index, int length) { + return buffer.slice(index, length).retain(); + } + + private void decodeByte(byte c, ByteBuf in, int idx) { + if ((c == '{' || c == '[') && !insideString) { + openBraces++; + } + else if ((c == '}' || c == ']') && !insideString) { + openBraces--; + } + else if (c == '"') { + // start of a new JSON string. It's necessary to detect strings as they may + // also contain braces/brackets and that could lead to incorrect results. + if (!insideString) { + insideString = true; + // If the double quote wasn't escaped then this is the end of a string. + } + else if (in.getByte(idx - 1) != '\\') { + insideString = false; + } + } + } + + private void initDecoding(byte openingBrace, boolean streamArrayElements) { + openBraces = 1; + if (openingBrace == '[' && streamArrayElements) { + state = ST_DECODING_ARRAY_STREAM; + } + else { + state = ST_DECODING_NORMAL; + } + } + + private void reset() { + insideString = false; + state = ST_INIT; + openBraces = 0; + } + + }); + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/StringDecoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/StringDecoder.java new file mode 100644 index 0000000000..3943bf9359 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/StringDecoder.java @@ -0,0 +1,54 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.nio.ByteBuffer; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; + +import org.reactivestreams.Publisher; +import reactor.io.buffer.Buffer; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.encoder.StringEncoder; +import org.springframework.reactive.codec.support.HintUtils; + +/** + * Decode from a bytes stream to a String stream. + * + * @author Sebastien Deleuze + * @see StringEncoder + */ +public class StringDecoder implements ByteToMessageDecoder { + + public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8; + + + @Override + public boolean canDecode(ResolvableType type, MediaType mediaType, Object... hints) { + return mediaType.isCompatibleWith(MediaType.TEXT_PLAIN); + } + + @Override + public Publisher decode(Publisher inputStream, ResolvableType type, MediaType mediaType, Object... hints) { + Charset charset = HintUtils.getHintByClass(Charset.class, hints, DEFAULT_CHARSET); + return Streams.wrap(inputStream).map(chunk -> new String(new Buffer(chunk).asBytes(), charset)); + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoder.java new file mode 100644 index 0000000000..931b853f56 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoder.java @@ -0,0 +1,76 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.io.IOException; +import java.nio.ByteBuffer; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.reactivestreams.Publisher; +import reactor.io.buffer.Buffer; +import reactor.rx.Stream; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.CodecException; +import org.springframework.reactive.codec.decoder.JacksonJsonDecoder; +import org.springframework.reactive.io.BufferOutputStream; + +/** + * Encode from an {@code Object} stream to a byte stream of JSON objects. + * + * @author Sebastien Deleuze + * @see JacksonJsonDecoder + */ +public class JacksonJsonEncoder implements MessageToByteEncoder { + + private final ObjectMapper mapper; + + + public JacksonJsonEncoder() { + this(new ObjectMapper()); + } + + public JacksonJsonEncoder(ObjectMapper mapper) { + this.mapper = mapper; + } + + + @Override + public boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints) { + return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON); + } + + @Override + public Publisher encode(Publisher messageStream, ResolvableType type, MediaType mediaType, Object... hints) { + Stream stream = Streams.wrap(messageStream).map(value -> { + Buffer buffer = new Buffer(); + BufferOutputStream outputStream = new BufferOutputStream(buffer); + try { + this.mapper.writeValue(outputStream, value); + } + catch (IOException e) { + throw new CodecException("Error while writing the data", e); + } + buffer.flip(); + return buffer.byteBuffer(); + }); + return stream; + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/Jaxb2Encoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/Jaxb2Encoder.java new file mode 100644 index 0000000000..b0adac193f --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/Jaxb2Encoder.java @@ -0,0 +1,108 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import javax.xml.bind.JAXBContext; +import javax.xml.bind.JAXBException; +import javax.xml.bind.MarshalException; +import javax.xml.bind.Marshaller; + +import org.reactivestreams.Publisher; +import reactor.io.buffer.Buffer; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.CodecException; +import org.springframework.reactive.codec.decoder.Jaxb2Decoder; +import org.springframework.reactive.io.BufferOutputStream; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +/** + * Encode from an {@code Object} stream to a byte stream of XML elements. + * + * @author Sebastien Deleuze + * @see Jaxb2Decoder + */ +public class Jaxb2Encoder implements MessageToByteEncoder { + + private final ConcurrentMap, JAXBContext> jaxbContexts = new ConcurrentHashMap, JAXBContext>(64); + + + @Override + public boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints) { + return mediaType.isCompatibleWith(MediaType.APPLICATION_XML) || mediaType.isCompatibleWith(MediaType.TEXT_XML); + } + + @Override + public Publisher encode(Publisher messageStream, ResolvableType type, MediaType mediaType, Object... hints) { + return Streams.wrap(messageStream).map(value -> { + try { + Buffer buffer = new Buffer(); + BufferOutputStream outputStream = new BufferOutputStream(buffer); + Class clazz = ClassUtils.getUserClass(value); + Marshaller marshaller = createMarshaller(clazz); + marshaller.setProperty(Marshaller.JAXB_ENCODING, StandardCharsets.UTF_8.name()); + marshaller.marshal(value, outputStream); + buffer.flip(); + return buffer.byteBuffer(); + } + catch (MarshalException ex) { + throw new CodecException( + "Could not marshal [" + value + "]: " + ex.getMessage(), ex); + } + catch (JAXBException ex) { + throw new CodecException( + "Could not instantiate JAXBContext: " + ex.getMessage(), ex); + } + }); + } + + protected final Marshaller createMarshaller(Class clazz) { + try { + JAXBContext jaxbContext = getJaxbContext(clazz); + Marshaller marshaller = jaxbContext.createMarshaller(); + return marshaller; + } + catch (JAXBException ex) { + throw new CodecException( + "Could not create Marshaller for class [" + clazz + "]: " + ex.getMessage(), ex); + } + } + + protected final JAXBContext getJaxbContext(Class clazz) { + Assert.notNull(clazz, "'clazz' must not be null"); + JAXBContext jaxbContext = this.jaxbContexts.get(clazz); + if (jaxbContext == null) { + try { + jaxbContext = JAXBContext.newInstance(clazz); + this.jaxbContexts.putIfAbsent(clazz, jaxbContext); + } + catch (JAXBException ex) { + throw new CodecException( + "Could not instantiate JAXBContext for class [" + clazz + "]: " + ex.getMessage(), ex); + } + } + return jaxbContext; + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/JsonObjectEncoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/JsonObjectEncoder.java new file mode 100644 index 0000000000..974f00b943 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/JsonObjectEncoder.java @@ -0,0 +1,61 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.nio.ByteBuffer; + +import org.reactivestreams.Publisher; + +import rx.Observable; +import rx.RxReactiveStreams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.decoder.JsonObjectDecoder; + +/** + * Encode a bye stream of individual JSON element to a byte stream representing a single + * JSON array when {@code Hints.ENCODE_AS_ARRAY} is enabled. + * + * @author Sebastien Deleuze + * @see JsonObjectDecoder + */ +public class JsonObjectEncoder implements MessageToByteEncoder { + + private final ByteBuffer START_ARRAY = ByteBuffer.wrap("[".getBytes()); + private final ByteBuffer END_ARRAY = ByteBuffer.wrap("]".getBytes()); + private final ByteBuffer COMMA = ByteBuffer.wrap(",".getBytes()); + + + @Override + public boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints) { + return mediaType.isCompatibleWith(MediaType.APPLICATION_JSON) && + (Observable.class.isAssignableFrom(type.getRawClass()) || Publisher.class.isAssignableFrom(type.getRawClass())); + } + + @Override + public Publisher encode(Publisher messageStream, ResolvableType type, MediaType mediaType, Object... hints) { + // TODO We use RxJava Observable because there is no skipLast() operator in Reactor + // TODO Merge some chunks, there is no need to have chunks with only '[', ']' or ',' characters + return RxReactiveStreams.toPublisher( + Observable.concat( + Observable.just(START_ARRAY), + RxReactiveStreams.toObservable(messageStream).flatMap(b -> Observable.just(b, COMMA)).skipLast(1), + Observable.just(END_ARRAY))); + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/MessageToByteEncoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/MessageToByteEncoder.java new file mode 100644 index 0000000000..9a1b67162f --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/MessageToByteEncoder.java @@ -0,0 +1,58 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.nio.ByteBuffer; + +import org.reactivestreams.Publisher; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.decoder.ByteToMessageDecoder; + +/** + * Encode from a message stream to a bytes stream. + * + * @author Sebastien Deleuze + * @see ByteToMessageDecoder + */ +public interface MessageToByteEncoder { + + /** + * Indicate whether the given type and media type can be processed by this encoder. + * @param type the (potentially generic) type to ultimately encode from. + * Could be different from {@code T} type. + * @param mediaType the media type to encode. + * Typically the value of an {@code Accept} header for HTTP request. + * @param hints Additional information about how to encode, optional. + * @return {@code true} if encodable; {@code false} otherwise + */ + boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints); + + /** + * Encode a given message stream to the given output byte stream. + * @param messageStream the message stream to encode. + * @param type the (potentially generic) type to ultimately encode from. + * Could be different from {@code T} type. + * @param mediaType the media type to encode. + * Typically the value of an {@code Accept} header for HTTP request. + * @param hints Additional information about how to encode, optional. + * @return the encoded bytes stream + */ + Publisher encode(Publisher messageStream, ResolvableType type, MediaType mediaType, Object... hints); + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/StringEncoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/StringEncoder.java new file mode 100644 index 0000000000..a27adf6e36 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/StringEncoder.java @@ -0,0 +1,53 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.nio.ByteBuffer; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; + +import org.reactivestreams.Publisher; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.decoder.StringDecoder; +import org.springframework.reactive.codec.support.HintUtils; + +/** + * Encode from a String stream to a bytes stream. + * + * @author Sebastien Deleuze + * @see StringDecoder + */ +public class StringEncoder implements MessageToByteEncoder { + + public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8; + + + @Override + public boolean canEncode(ResolvableType type, MediaType mediaType, Object... hints) { + return mediaType.isCompatibleWith(MediaType.TEXT_PLAIN); + } + + @Override + public Publisher encode(Publisher elementStream, ResolvableType type, MediaType mediaType, Object... hints) { + final Charset charset = HintUtils.getHintByClass(Charset.class, hints, DEFAULT_CHARSET); + return Streams.wrap(elementStream).map(s -> ByteBuffer.wrap(s.getBytes(charset))); + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/codec/support/HintUtils.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/support/HintUtils.java new file mode 100644 index 0000000000..ab63b2f659 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/support/HintUtils.java @@ -0,0 +1,54 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.support; + +import org.springframework.reactive.codec.decoder.ByteToMessageDecoder; +import org.springframework.reactive.codec.encoder.MessageToByteEncoder; + +/** + * Utility methods for dealing with codec hints. + * + * @author Sebastien Deleuze + * @see Hints + * @see MessageToByteEncoder + * @see ByteToMessageDecoder + */ +public abstract class HintUtils { + + public static T getHintByClass(Class clazz, Object[] hints) { + return getHintByClass(clazz, hints, null); + } + + public static T getHintByClass(Class clazz, Object[] hints, T defaultValue) { + for (Object hint : hints) { + if (hint.getClass().isAssignableFrom(clazz)) { + return (T)hint; + } + } + return defaultValue; + } + + public static boolean containsHint(Object hint, Object[] hints) { + for (Object h : hints) { + if (h.equals(hint)) { + return true; + } + } + return false; + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/io/BufferOutputStream.java b/spring-web-reactive/src/main/java/org/springframework/reactive/io/BufferOutputStream.java new file mode 100644 index 0000000000..4d51ec8788 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/io/BufferOutputStream.java @@ -0,0 +1,48 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.io; + +import java.io.IOException; +import java.io.OutputStream; + +import reactor.io.buffer.Buffer; + +/** + * @author Sebastien Deleuze + */ +public class BufferOutputStream extends OutputStream { + + private Buffer buffer; + + public BufferOutputStream(Buffer buffer) { + this.buffer = buffer; + } + + public void write(int b) throws IOException { + buffer.append(b); + } + + public void write(byte[] bytes, int off, int len) + throws IOException { + buffer.append(bytes, off, len); + } + + public Buffer getBuffer() { + return buffer; + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/io/ByteBufferInputStream.java b/spring-web-reactive/src/main/java/org/springframework/reactive/io/ByteBufferInputStream.java new file mode 100644 index 0000000000..dbd2a5e16d --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/io/ByteBufferInputStream.java @@ -0,0 +1,48 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.io; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; + +/** + * Simple {@link InputStream} implementation that exposes currently + * available content of a {@link ByteBuffer}. + * + * From Jackson ByteBufferBackedInputStream + */ +public class ByteBufferInputStream extends InputStream { + + protected final ByteBuffer b; + + public ByteBufferInputStream(ByteBuffer buf) { b = buf; } + + @Override public int available() { return b.remaining(); } + + @Override + public int read() throws IOException { return b.hasRemaining() ? (b.get() & 0xFF) : -1; } + + @Override + public int read(byte[] bytes, int off, int len) throws IOException { + if (!b.hasRemaining()) return -1; + len = Math.min(len, b.remaining()); + b.get(bytes, off, len); + return len; + } + +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/RequestBodyArgumentResolver.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/RequestBodyArgumentResolver.java new file mode 100644 index 0000000000..6a7159d8fd --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/RequestBodyArgumentResolver.java @@ -0,0 +1,134 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.web.dispatch.method.annotation; + +import java.nio.ByteBuffer; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.reactivestreams.Publisher; +import reactor.rx.Stream; +import reactor.rx.Streams; +import rx.Observable; +import rx.RxReactiveStreams; + +import org.springframework.core.MethodParameter; +import org.springframework.core.ResolvableType; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.decoder.ByteToMessageDecoder; +import org.springframework.reactive.web.dispatch.method.HandlerMethodArgumentResolver; +import org.springframework.reactive.web.http.ServerHttpRequest; +import org.springframework.web.bind.annotation.RequestBody; + +/** + * @author Sebastien Deleuze + */ +public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolver { + + private static final Charset UTF_8 = Charset.forName("UTF-8"); + + private final List> deserializers; + private final List> preProcessors; + + + public RequestBodyArgumentResolver(List> deserializers) { + this(deserializers, Collections.EMPTY_LIST); + } + + public RequestBodyArgumentResolver(List> deserializers, List> preProcessors) { + this.deserializers = deserializers; + this.preProcessors = preProcessors; + } + + @Override + public boolean supportsParameter(MethodParameter parameter) { + return parameter.hasParameterAnnotation(RequestBody.class); + } + + @Override + public Object resolveArgument(MethodParameter parameter, ServerHttpRequest request) { + + MediaType mediaType = resolveMediaType(request); + ResolvableType type = ResolvableType.forMethodParameter(parameter); + List hints = new ArrayList<>(); + hints.add(UTF_8); + // TODO: Refactor type conversion + ResolvableType readType = type; + if (Observable.class.isAssignableFrom(type.getRawClass()) || Publisher.class.isAssignableFrom(type.getRawClass())) { + readType = type.getGeneric(0); + } + + ByteToMessageDecoder deserializer = resolveDeserializers(request, type, mediaType, hints.toArray()); + if (deserializer != null) { + + Publisher inputStream = Streams.wrap(request.getBody()).map(bytes -> ByteBuffer.wrap(bytes)); + List> preProcessors = resolvePreProcessors(request, type, mediaType, hints.toArray()); + for (ByteToMessageDecoder preProcessor : preProcessors) { + inputStream = preProcessor.decode(inputStream, type, mediaType, hints.toArray()); + } + Publisher elementStream = deserializer.decode(inputStream, readType, mediaType, UTF_8); + + // TODO: Refactor type conversion + if (Stream.class.isAssignableFrom(type.getRawClass())) { + return Streams.wrap(elementStream); + } + else if (Observable.class.isAssignableFrom(type.getRawClass())) { + return RxReactiveStreams.toObservable(elementStream); + } + else if (Publisher.class.isAssignableFrom(type.getRawClass())) { + return elementStream; + } + else { + try { + return Streams.wrap(elementStream).next().await(); + } catch(InterruptedException ex) { + throw new IllegalStateException("Timeout before getter the value"); + } + } + } + throw new IllegalStateException("Argument type not supported: " + type); + } + + private MediaType resolveMediaType(ServerHttpRequest request) { + String acceptHeader = request.getHeaders().getFirst(HttpHeaders.CONTENT_TYPE); + List mediaTypes = MediaType.parseMediaTypes(acceptHeader); + MediaType.sortBySpecificityAndQuality(mediaTypes); + return ( mediaTypes.size() > 0 ? mediaTypes.get(0) : MediaType.TEXT_PLAIN); + } + + private ByteToMessageDecoder resolveDeserializers(ServerHttpRequest request, ResolvableType type, MediaType mediaType, Object[] hints) { + for (ByteToMessageDecoder deserializer : this.deserializers) { + if (deserializer.canDecode(type, mediaType, hints)) { + return deserializer; + } + } + return null; + } + + private List> resolvePreProcessors(ServerHttpRequest request, ResolvableType type, MediaType mediaType, Object[] hints) { + List> preProcessors = new ArrayList<>(); + for (ByteToMessageDecoder preProcessor : this.preProcessors) { + if (preProcessor.canDecode(type, mediaType, hints)) { + preProcessors.add(preProcessor); + } + } + return preProcessors; + } +} diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/RequestMappingHandlerAdapter.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/RequestMappingHandlerAdapter.java index d57e9e9413..5fcab45fd2 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/RequestMappingHandlerAdapter.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/RequestMappingHandlerAdapter.java @@ -16,12 +16,16 @@ package org.springframework.reactive.web.dispatch.method.annotation; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import org.reactivestreams.Publisher; import reactor.rx.Streams; import org.springframework.beans.factory.InitializingBean; +import org.springframework.reactive.codec.decoder.JacksonJsonDecoder; +import org.springframework.reactive.codec.decoder.JsonObjectDecoder; +import org.springframework.reactive.codec.decoder.StringDecoder; import org.springframework.reactive.web.dispatch.HandlerAdapter; import org.springframework.reactive.web.dispatch.HandlerResult; import org.springframework.reactive.web.dispatch.method.HandlerMethodArgumentResolver; @@ -50,6 +54,7 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin if (this.argumentResolvers == null) { this.argumentResolvers = new ArrayList<>(); this.argumentResolvers.add(new RequestParamArgumentResolver()); + this.argumentResolvers.add(new RequestBodyArgumentResolver(Arrays.asList(new StringDecoder(), new JacksonJsonDecoder()), Arrays.asList(new JsonObjectDecoder(true)))); } } diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/ResponseBodyResultHandler.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/ResponseBodyResultHandler.java index 21d8ddb0d0..69f8d483c9 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/ResponseBodyResultHandler.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/ResponseBodyResultHandler.java @@ -16,16 +16,25 @@ package org.springframework.reactive.web.dispatch.method.annotation; import java.lang.reflect.Method; +import java.nio.ByteBuffer; import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import org.reactivestreams.Publisher; +import reactor.io.buffer.Buffer; import reactor.rx.Streams; +import rx.Observable; +import rx.RxReactiveStreams; import org.springframework.core.MethodParameter; import org.springframework.core.Ordered; import org.springframework.core.ResolvableType; import org.springframework.core.annotation.AnnotatedElementUtils; +import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; +import org.springframework.reactive.codec.encoder.MessageToByteEncoder; import org.springframework.reactive.web.dispatch.HandlerResult; import org.springframework.reactive.web.dispatch.HandlerResultHandler; import org.springframework.reactive.web.http.ServerHttpRequest; @@ -35,8 +44,7 @@ import org.springframework.web.method.HandlerMethod; /** - * For now a simple {@code String} or {@code Publisher} to - * "text/plain;charset=UTF-8" conversion. + * First version using {@link MessageToByteEncoder}s * * @author Rossen Stoyanchev */ @@ -45,9 +53,21 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered private static final Charset UTF_8 = Charset.forName("UTF-8"); + private final List> serializers; + private final List> postProcessors; + private int order = Ordered.LOWEST_PRECEDENCE; + public ResponseBodyResultHandler(List> serializers) { + this(serializers, Collections.EMPTY_LIST); + } + + public ResponseBodyResultHandler(List> serializers, List> postProcessors) { + this.serializers = serializers; + this.postProcessors = postProcessors; + } + public void setOrder(int order) { this.order = order; } @@ -80,20 +100,61 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered return Streams.empty(); } - if (value instanceof String) { - response.getHeaders().setContentType(new MediaType("text", "plain", UTF_8)); - return response.writeWith(Streams.just(((String) value).getBytes(UTF_8))); + MediaType mediaType = resolveMediaType(request); + ResolvableType type = ResolvableType.forMethodParameter(returnType); + List hints = new ArrayList<>(); + hints.add(UTF_8); + MessageToByteEncoder serializer = (MessageToByteEncoder)resolveSerializer(request, type, mediaType, hints.toArray()); + if (serializer != null) { + Publisher elementStream; + + // TODO: Refactor type conversion + if (Observable.class.isAssignableFrom(type.getRawClass())) { + elementStream = RxReactiveStreams.toPublisher((Observable) value); + } + else if (Publisher.class.isAssignableFrom(type.getRawClass())) { + elementStream = (Publisher)value; + } + else { + elementStream = Streams.just(value); + } + + Publisher outputStream = serializer.encode(elementStream, type, mediaType, hints.toArray()); + List> postProcessors = resolvePostProcessors(request, type, mediaType, hints.toArray()); + for (MessageToByteEncoder postProcessor : postProcessors) { + outputStream = postProcessor.encode(outputStream, type, mediaType, hints.toArray()); + } + response.getHeaders().setContentType(mediaType); + return response.writeWith(Streams.wrap(outputStream).map(buffer -> new Buffer(buffer).asBytes())); } - else if (value instanceof Publisher) { - Class type = ResolvableType.forMethodParameter(returnType).resolveGeneric(0); - if (String.class.equals(type)) { - @SuppressWarnings("unchecked") - Publisher content = (Publisher) value; - return response.writeWith(Streams.wrap(content).map(value1 -> value1.getBytes(UTF_8))); + return Streams.fail(new IllegalStateException( + "Return value type not supported: " + returnType)); + } + + private MediaType resolveMediaType(ServerHttpRequest request) { + String acceptHeader = request.getHeaders().getFirst(HttpHeaders.ACCEPT); + List mediaTypes = MediaType.parseMediaTypes(acceptHeader); + MediaType.sortBySpecificityAndQuality(mediaTypes); + return ( mediaTypes.size() > 0 ? mediaTypes.get(0) : MediaType.TEXT_PLAIN); + } + + private MessageToByteEncoder resolveSerializer(ServerHttpRequest request, ResolvableType type, MediaType mediaType, Object[] hints) { + for (MessageToByteEncoder codec : this.serializers) { + if (codec.canEncode(type, mediaType, hints)) { + return codec; } } + return null; + } - return Streams.fail(new IllegalStateException("Return value type not supported: " + returnType)); + private List> resolvePostProcessors(ServerHttpRequest request, ResolvableType type, MediaType mediaType, Object[] hints) { + List> postProcessors = new ArrayList<>(); + for (MessageToByteEncoder postProcessor : this.postProcessors) { + if (postProcessor.canEncode(type, mediaType, hints)) { + postProcessors.add(postProcessor); + } + } + return postProcessors; } } diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoderTests.java new file mode 100644 index 0000000000..b29dc553fe --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoderTests.java @@ -0,0 +1,56 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.nio.ByteBuffer; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import org.junit.Test; +import reactor.io.buffer.Buffer; +import reactor.rx.Stream; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.converter.Pojo; + +/** + * @author Sebastien Deleuze + */ +public class JacksonJsonDecoderTests { + + private final JacksonJsonDecoder decoder = new JacksonJsonDecoder(); + + @Test + public void canDecode() { + assertTrue(decoder.canDecode(null, MediaType.APPLICATION_JSON)); + assertFalse(decoder.canDecode(null, MediaType.APPLICATION_XML)); + } + + @Test + public void decode() throws InterruptedException { + Stream source = Streams.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer()); + List results = Streams.wrap(decoder.decode(source, ResolvableType.forClass(Pojo.class), null)) + .toList().await(); + assertEquals(1, results.size()); + assertEquals("foofoo", ((Pojo) results.get(0)).getFoo()); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/Jaxb2DecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/Jaxb2DecoderTests.java new file mode 100644 index 0000000000..24d9d9cce1 --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/Jaxb2DecoderTests.java @@ -0,0 +1,57 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.nio.ByteBuffer; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import org.junit.Test; +import reactor.io.buffer.Buffer; +import reactor.rx.Stream; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.converter.Pojo; + +/** + * @author Sebastien Deleuze + */ +public class Jaxb2DecoderTests { + + private final Jaxb2Decoder decoder = new Jaxb2Decoder(); + + @Test + public void canDecode() { + assertTrue(decoder.canDecode(null, MediaType.APPLICATION_XML)); + assertTrue(decoder.canDecode(null, MediaType.TEXT_XML)); + assertFalse(decoder.canDecode(null, MediaType.APPLICATION_JSON)); + } + + @Test + public void decode() throws InterruptedException { + Stream source = Streams.just(Buffer.wrap("barbarfoofoo").byteBuffer()); + List results = Streams.wrap(decoder.decode(source, ResolvableType.forClass(Pojo.class), null)) + .toList().await(); + assertEquals(1, results.size()); + assertEquals("foofoo", ((Pojo) results.get(0)).getFoo()); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JsonObjectDecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JsonObjectDecoderTests.java new file mode 100644 index 0000000000..d683796e6d --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JsonObjectDecoderTests.java @@ -0,0 +1,62 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import org.junit.Test; +import reactor.io.buffer.Buffer; +import reactor.rx.Stream; +import reactor.rx.Streams; + +/** + * @author Sebastien Deleuze + */ +public class JsonObjectDecoderTests { + + @Test + public void decodeSingleChunkToArray() throws InterruptedException { + JsonObjectDecoder decoder = new JsonObjectDecoder(true); + Stream source = Streams.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer()); + List results = Streams.wrap(decoder.decode(source, null, null)).map(chunk -> { + byte[] b = new byte[chunk.remaining()]; + chunk.get(b); + return new String(b, StandardCharsets.UTF_8); + }).toList().await(); + assertEquals(2, results.size()); + assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0)); + assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1)); + } + + @Test + public void decodeMultipleChunksToArray() throws InterruptedException { + JsonObjectDecoder decoder = new JsonObjectDecoder(true); + Stream source = Streams.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\"").byteBuffer(), Buffer.wrap(": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer()); + List results = Streams.wrap(decoder.decode(source, null, null)).map(chunk -> { + byte[] b = new byte[chunk.remaining()]; + chunk.get(b); + return new String(b, StandardCharsets.UTF_8); + }).toList().await(); + assertEquals(2, results.size()); + assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0)); + assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1)); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java new file mode 100644 index 0000000000..9ad83fc7b9 --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java @@ -0,0 +1,57 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.decoder; + +import java.nio.ByteBuffer; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import org.junit.Test; +import reactor.io.buffer.Buffer; +import reactor.rx.Stream; +import reactor.rx.Streams; + +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.reactive.converter.Pojo; + +/** + * @author Sebastien Deleuze + */ +public class StringDecoderTests { + + private final StringDecoder decoder = new StringDecoder(); + + @Test + public void canDecode() { + assertTrue(decoder.canDecode(null, MediaType.TEXT_PLAIN)); + assertFalse(decoder.canDecode(null, MediaType.APPLICATION_JSON)); + } + + @Test + public void decode() throws InterruptedException { + Stream source = Streams.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); + List results = Streams.wrap(decoder.decode(source, ResolvableType.forClass(Pojo.class), null)) + .toList().await(); + assertEquals(2, results.size()); + assertEquals("foo", results.get(0)); + assertEquals("bar", results.get(1)); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoderTests.java new file mode 100644 index 0000000000..4207c5997e --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoderTests.java @@ -0,0 +1,58 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.nio.charset.StandardCharsets; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import org.junit.Test; +import reactor.rx.Stream; +import reactor.rx.Streams; + +import org.springframework.http.MediaType; +import org.springframework.reactive.converter.Pojo; + +/** + * @author Sebastien Deleuze + */ +public class JacksonJsonEncoderTests { + + private final JacksonJsonEncoder encoder = new JacksonJsonEncoder(); + + @Test + public void canWrite() { + assertTrue(encoder.canEncode(null, MediaType.APPLICATION_JSON)); + assertFalse(encoder.canEncode(null, MediaType.APPLICATION_XML)); + } + + @Test + public void write() throws InterruptedException { + Stream source = Streams.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar")); + List results = Streams.wrap(encoder.encode(source, null, null)).map(chunk -> { + byte[] b = new byte[chunk.remaining()]; + chunk.get(b); + return new String(b, StandardCharsets.UTF_8); + }).toList().await(); + assertEquals(2, results.size()); + assertEquals("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", results.get(0)); + assertEquals("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}", results.get(1)); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/Jaxb2EncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/Jaxb2EncoderTests.java new file mode 100644 index 0000000000..1710b6708b --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/Jaxb2EncoderTests.java @@ -0,0 +1,60 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.nio.charset.StandardCharsets; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import org.junit.Test; +import reactor.rx.Stream; +import reactor.rx.Streams; + +import org.springframework.http.MediaType; +import org.springframework.reactive.codec.decoder.Jaxb2Decoder; +import org.springframework.reactive.converter.Pojo; + +/** + * @author Sebastien Deleuze + */ +public class Jaxb2EncoderTests { + + private final Jaxb2Encoder encoder = new Jaxb2Encoder(); + + @Test + public void canEncode() { + assertTrue(encoder.canEncode(null, MediaType.APPLICATION_XML)); + assertTrue(encoder.canEncode(null, MediaType.TEXT_XML)); + assertFalse(encoder.canEncode(null, MediaType.APPLICATION_JSON)); + } + + @Test + public void encode() throws InterruptedException { + Stream source = Streams.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar")); + List results = Streams.wrap(encoder.encode(source, null, null)).map(chunk -> { + byte[] b = new byte[chunk.remaining()]; + chunk.get(b); + return new String(b, StandardCharsets.UTF_8); + }).toList().await(); + assertEquals(2, results.size()); + assertEquals("barbarfoofoo", results.get(0)); + assertEquals("barbarbarfoofoofoo", results.get(1)); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JsonObjectEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JsonObjectEncoderTests.java new file mode 100644 index 0000000000..47726014f0 --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JsonObjectEncoderTests.java @@ -0,0 +1,47 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import org.junit.Test; +import reactor.io.buffer.Buffer; +import reactor.rx.Stream; +import reactor.rx.Streams; + +/** + * @author Sebastien Deleuze + */ +public class JsonObjectEncoderTests { + + @Test + public void encodeToArray() throws InterruptedException { + JsonObjectEncoder encoder = new JsonObjectEncoder(); + Stream source = Streams.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer(), Buffer.wrap("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}").byteBuffer()); + List results = Streams.wrap(encoder.encode(source, null, null)).map(chunk -> { + byte[] b = new byte[chunk.remaining()]; + chunk.get(b); + return new String(b, StandardCharsets.UTF_8); + }).toList().await(); + String result = String.join("", results); + assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]", result); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/StringEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/StringEncoderTests.java new file mode 100644 index 0000000000..ce67e5e29a --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/StringEncoderTests.java @@ -0,0 +1,55 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.codec.encoder; + +import java.nio.charset.StandardCharsets; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import org.junit.Test; +import reactor.rx.Streams; + +import org.springframework.http.MediaType; + +/** + * @author Sebastien Deleuze + */ +public class StringEncoderTests { + + private final StringEncoder encoder = new StringEncoder(); + + @Test + public void canWrite() { + assertTrue(encoder.canEncode(null, MediaType.TEXT_PLAIN)); + assertFalse(encoder.canEncode(null, MediaType.APPLICATION_JSON)); + } + + @Test + public void write() throws InterruptedException { + List results = Streams.wrap(encoder.encode(Streams.just("foo"), null, null)) + .map(chunk -> { + byte[] b = new byte[chunk.remaining()]; + chunk.get(b); + return new String(b, StandardCharsets.UTF_8); + }).toList().await(); + assertEquals(1, results.size()); + assertEquals("foo", results.get(0)); + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/converter/Pojo.java b/spring-web-reactive/src/test/java/org/springframework/reactive/converter/Pojo.java new file mode 100644 index 0000000000..913acc4d12 --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/converter/Pojo.java @@ -0,0 +1,55 @@ +/* + * Copyright 2002-2015 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 + * + * http://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.reactive.converter; + +import javax.xml.bind.annotation.XmlRootElement; + +/** + * @author Sebastien Deleuze + */ +@XmlRootElement +public class Pojo { + + private String foo; + + private String bar; + + public Pojo() { + } + + public Pojo(String foo, String bar) { + this.foo = foo; + this.bar = bar; + } + + public String getFoo() { + return this.foo; + } + + public void setFoo(String foo) { + this.foo = foo; + } + + public String getBar() { + return this.bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/web/dispatch/method/annotation/RequestMappingIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/reactive/web/dispatch/method/annotation/RequestMappingIntegrationTests.java index 52f2dacc4f..053fcc716b 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/web/dispatch/method/annotation/RequestMappingIntegrationTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/web/dispatch/method/annotation/RequestMappingIntegrationTests.java @@ -17,41 +17,48 @@ package org.springframework.reactive.web.dispatch.method.annotation; import java.net.URI; -import java.nio.charset.Charset; +import java.util.Arrays; +import java.util.List; +import static org.junit.Assert.assertEquals; import org.junit.Test; import org.reactivestreams.Publisher; +import reactor.rx.Stream; import reactor.rx.Streams; +import rx.Observable; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.http.MediaType; import org.springframework.http.RequestEntity; import org.springframework.http.ResponseEntity; +import org.springframework.reactive.codec.encoder.JacksonJsonEncoder; +import org.springframework.reactive.codec.encoder.JsonObjectEncoder; +import org.springframework.reactive.codec.encoder.StringEncoder; import org.springframework.reactive.web.dispatch.DispatcherHandler; import org.springframework.reactive.web.http.AbstractHttpHandlerIntegrationTests; import org.springframework.reactive.web.http.HttpHandler; import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.client.RestTemplate; import org.springframework.web.context.support.StaticWebApplicationContext; -import static org.junit.Assert.assertArrayEquals; - /** * @author Rossen Stoyanchev + * @author Sebastien Deleuze */ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrationTests { - private static final Charset UTF_8 = Charset.forName("UTF-8"); - - @Override protected HttpHandler createHttpHandler() { StaticWebApplicationContext wac = new StaticWebApplicationContext(); wac.registerSingleton("handlerMapping", RequestMappingHandlerMapping.class); wac.registerSingleton("handlerAdapter", RequestMappingHandlerAdapter.class); - wac.registerSingleton("responseBodyResultHandler", ResponseBodyResultHandler.class); + wac.getDefaultListableBeanFactory().registerSingleton("responseBodyResultHandler", + new ResponseBodyResultHandler(Arrays.asList(new StringEncoder(), new JacksonJsonEncoder()), Arrays.asList(new JsonObjectEncoder()))); wac.registerSingleton("controller", TestController.class); wac.refresh(); @@ -67,9 +74,97 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati URI url = new URI("http://localhost:" + port + "/param?name=George"); RequestEntity request = RequestEntity.get(url).build(); - ResponseEntity response = restTemplate.exchange(request, byte[].class); + ResponseEntity response = restTemplate.exchange(request, String.class); - assertArrayEquals("Hello George!".getBytes(UTF_8), response.getBody()); + assertEquals("Hello George!", response.getBody()); + } + + @Test + public void serializeAsPojo() throws Exception { + + RestTemplate restTemplate = new RestTemplate(); + + URI url = new URI("http://localhost:" + port + "/person"); + RequestEntity request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build(); + ResponseEntity response = restTemplate.exchange(request, Person.class); + + assertEquals(new Person("Robert"), response.getBody()); + } + + @Test + public void serializeAsList() throws Exception { + + RestTemplate restTemplate = new RestTemplate(); + + URI url = new URI("http://localhost:" + port + "/list"); + RequestEntity request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build(); + List results = restTemplate.exchange(request, new ParameterizedTypeReference>(){}).getBody(); + + assertEquals(2, results.size()); + assertEquals(new Person("Robert"), results.get(0)); + assertEquals(new Person("Marie"), results.get(1)); + } + + @Test + public void serializeAsPublisher() throws Exception { + + RestTemplate restTemplate = new RestTemplate(); + + URI url = new URI("http://localhost:" + port + "/publisher"); + RequestEntity request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build(); + List results = restTemplate.exchange(request, new ParameterizedTypeReference>(){}).getBody(); + + assertEquals(2, results.size()); + assertEquals(new Person("Robert"), results.get(0)); + assertEquals(new Person("Marie"), results.get(1)); + } + + @Test + public void serializeAsObservable() throws Exception { + + RestTemplate restTemplate = new RestTemplate(); + + URI url = new URI("http://localhost:" + port + "/observable"); + RequestEntity request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build(); + List results = restTemplate.exchange(request, new ParameterizedTypeReference>() { + }).getBody(); + + assertEquals(2, results.size()); + assertEquals(new Person("Robert"), results.get(0)); + assertEquals(new Person("Marie"), results.get(1)); + } + + @Test + public void serializeAsReactorStream() throws Exception { + + RestTemplate restTemplate = new RestTemplate(); + + URI url = new URI("http://localhost:" + port + "/stream"); + RequestEntity request = RequestEntity.get(url).accept(MediaType.APPLICATION_JSON).build(); + List results = restTemplate.exchange(request, new ParameterizedTypeReference>() { + }).getBody(); + + assertEquals(2, results.size()); + assertEquals(new Person("Robert"), results.get(0)); + assertEquals(new Person("Marie"), results.get(1)); + } + + @Test + public void echo() throws Exception { + RestTemplate restTemplate = new RestTemplate(); + + URI url = new URI("http://localhost:" + port + "/echo"); + List persons = Arrays.asList(new Person("Robert"), new Person("Marie")); + RequestEntity> request = RequestEntity + .post(url) + .contentType(MediaType.APPLICATION_JSON) + .accept(MediaType.APPLICATION_JSON) + .body(persons); + List results = restTemplate.exchange(request, new ParameterizedTypeReference>(){}).getBody(); + + assertEquals(2, results.size()); + assertEquals("Robert", results.get(0).getName()); + assertEquals("Marie", results.get(1).getName()); } @@ -82,6 +177,73 @@ public class RequestMappingIntegrationTests extends AbstractHttpHandlerIntegrati public Publisher handleWithParam(@RequestParam String name) { return Streams.just("Hello ", name, "!"); } + + @RequestMapping("/person") + @ResponseBody + public Person personResponseBody() { + return new Person("Robert"); + } + + @RequestMapping("/list") + @ResponseBody + public List listResponseBody() { + return Arrays.asList(new Person("Robert"), new Person("Marie")); + } + + @RequestMapping("/publisher") + @ResponseBody + public Publisher publisherResponseBody() { + return Streams.just(new Person("Robert"), new Person("Marie")); + } + + @RequestMapping("/observable") + @ResponseBody + public Observable observableResponseBody() { + return Observable.just(new Person("Robert"), new Person("Marie")); + } + + @RequestMapping("/stream") + @ResponseBody + public Stream reactorStreamResponseBody() { + return Streams.just(new Person("Robert"), new Person("Marie")); + } + + @RequestMapping("/echo") + @ResponseBody + public Publisher echo(@RequestBody Publisher persons) { + return persons; + } + + } + + private static class Person { + + private String name; + + public Person() { + } + + public Person(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + @Override + public boolean equals(Object o) { + return name.equals(((Person)o).name); + } + + @Override + public int hashCode() { + return name.hashCode(); + } } }