From 464ff1d9606a3189a41bae80e504929dec091d33 Mon Sep 17 00:00:00 2001 From: Stephane Maldini Date: Mon, 5 Oct 2015 15:35:59 +0100 Subject: [PATCH] Progress #24 Replace Streams fail, empty and Stream map with Publishers.error, empty and map. Replace Streams calls by Publishers.flatMap, just and Publishers.from Precise cancel and size to toReadQueue to simulate Promise Update build --- spring-web-reactive/build.gradle | 5 ++- .../codec/decoder/JacksonJsonDecoder.java | 25 +++++------ .../reactive/codec/decoder/Jaxb2Decoder.java | 43 ++++++++----------- .../codec/decoder/JsonObjectDecoder.java | 37 ++++++++-------- .../reactive/codec/decoder/StringDecoder.java | 15 +++---- .../codec/encoder/JacksonJsonEncoder.java | 18 +++----- .../reactive/codec/encoder/Jaxb2Encoder.java | 36 +++++++--------- .../reactive/codec/encoder/StringEncoder.java | 13 +++--- .../web/dispatch/DispatcherHandler.java | 19 ++++---- .../RequestBodyArgumentResolver.java | 8 ++-- .../annotation/ResponseBodyResultHandler.java | 38 ++++++++-------- .../reactor/ReactorServerHttpResponse.java | 4 +- 12 files changed, 121 insertions(+), 140 deletions(-) diff --git a/spring-web-reactive/build.gradle b/spring-web-reactive/build.gradle index 5e25378a19..d227418046 100644 --- a/spring-web-reactive/build.gradle +++ b/spring-web-reactive/build.gradle @@ -34,9 +34,10 @@ dependencies { compile "org.springframework:spring-core:4.2.0.RELEASE" compile "org.springframework:spring-web:4.2.0.RELEASE" compile "org.reactivestreams:reactive-streams:1.0.0" - compile "io.projectreactor:reactor-stream:2.1.0.BUILD-SNAPSHOT" + compile "io.projectreactor:reactor-core:2.1.0.BUILD-SNAPSHOT" compile "commons-logging:commons-logging:1.2" - compile "com.fasterxml.jackson.core:jackson-databind:2.6.1" + + optional "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/decoder/JacksonJsonDecoder.java b/spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoder.java index 150bf33a6e..0562653bdc 100644 --- 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 @@ -16,19 +16,18 @@ 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; +import reactor.Publishers; + +import java.io.IOException; +import java.nio.ByteBuffer; /** * Decode from a bytes stream of JSON objects to a stream of {@code Object} (POJO). @@ -56,15 +55,13 @@ public class JacksonJsonDecoder implements ByteToMessageDecoder { @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); - } - }); + return Publishers.map(inputStream, 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 index 77ae96a315..bff1ef0334 100644 --- 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 @@ -16,33 +16,27 @@ 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.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.ByteBufferPublisherInputStream; import org.springframework.util.Assert; +import org.xml.sax.InputSource; +import org.xml.sax.SAXException; +import org.xml.sax.XMLReader; +import org.xml.sax.helpers.XMLReaderFactory; +import reactor.Publishers; + +import javax.xml.bind.*; +import javax.xml.bind.annotation.XmlRootElement; +import javax.xml.transform.Source; +import javax.xml.transform.sax.SAXSource; +import javax.xml.transform.stream.StreamSource; +import java.nio.ByteBuffer; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; /** * Decode from a bytes stream of XML elements to a stream of {@code Object} (POJO). @@ -66,18 +60,19 @@ public class Jaxb2Decoder implements ByteToMessageDecoder { Source source = processSource(new StreamSource(new ByteBufferPublisherInputStream(inputStream))); Unmarshaller unmarshaller = createUnmarshaller(outputClass); if (outputClass.isAnnotationPresent(XmlRootElement.class)) { - return Streams.just(unmarshaller.unmarshal(source)); + return Publishers.just(unmarshaller.unmarshal(source)); } else { JAXBElement jaxbElement = unmarshaller.unmarshal(source, outputClass); - return Streams.just(jaxbElement.getValue()); + return Publishers.just(jaxbElement.getValue()); } } catch (UnmarshalException ex) { - return Streams.fail(new CodecException("Could not unmarshal to [" + outputClass + "]: " + ex.getMessage(), ex)); + return Publishers.error( + 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)); + return Publishers.error(new CodecException("Could not instantiate JAXBContext: " + ex.getMessage(), ex)); } } 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 index 4aa1edca46..c1f9c35926 100644 --- 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 @@ -16,22 +16,21 @@ 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.Promise; -import reactor.rx.Streams; -import rx.Observable; - import org.springframework.core.ResolvableType; import org.springframework.http.MediaType; import org.springframework.reactive.codec.encoder.JsonObjectEncoder; +import reactor.Publishers; +import reactor.fn.Function; +import reactor.rx.Promise; +import rx.Observable; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; /** * Decode an arbitrary split byte stream representing JSON objects to a bye stream @@ -98,7 +97,7 @@ public class JsonObjectDecoder implements ByteToMessageDecoder { @Override public Publisher decode(Publisher inputStream, ResolvableType type, MediaType mediaType, Object... hints) { - return Streams.wrap(inputStream).flatMap(new Function>() { + return Publishers.flatMap(inputStream, new Function>() { int openBraces; int idx; @@ -121,17 +120,17 @@ public class JsonObjectDecoder implements ByteToMessageDecoder { } if (state == ST_CORRUPTED) { in.skipBytes(in.readableBytes()); - return Streams.fail(new IllegalStateException("Corrupted stream")); + return Publishers.error(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")); + return Publishers.error(new IllegalStateException( + "object length exceeds " + maxObjectLength + ": " + + wrtIdx + + " bytes discarded")); } for (/* use current idx */; idx < wrtIdx; idx++) { @@ -202,16 +201,16 @@ public class JsonObjectDecoder implements ByteToMessageDecoder { } else { state = ST_CORRUPTED; - return Streams.fail(new IllegalStateException( - "invalid JSON received at byte position " + idx + - ": " + ByteBufUtil.hexDump(in))); + return Publishers.error(new IllegalStateException( + "invalid JSON received at byte position " + idx + + ": " + ByteBufUtil.hexDump(in))); } } if (in.readableBytes() == 0) { idx = 0; } - return Streams.from(chunks); + return Publishers.from(chunks); } /** 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 index 3943bf9359..b64bd81edb 100644 --- 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 @@ -16,18 +16,17 @@ 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; +import reactor.Publishers; +import reactor.io.buffer.Buffer; + +import java.nio.ByteBuffer; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; /** * Decode from a bytes stream to a String stream. @@ -48,7 +47,7 @@ public class StringDecoder implements ByteToMessageDecoder { @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)); + return Publishers.map(inputStream, 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 index 3fb9200140..efdb652be0 100644 --- 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 @@ -16,20 +16,18 @@ 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; +import reactor.Publishers; +import reactor.io.buffer.Buffer; + +import java.io.IOException; +import java.nio.ByteBuffer; /** * Encode from an {@code Object} stream to a byte stream of JSON objects. @@ -56,19 +54,17 @@ public class JacksonJsonEncoder implements MessageToByteEncoder { @Override public Publisher encode(Publisher messageStream, ResolvableType type, MediaType mediaType, Object... hints) { - Stream stream = Streams.wrap(messageStream).map(value -> { + return Publishers.map(messageStream, value -> { Buffer buffer = new Buffer(); BufferOutputStream outputStream = new BufferOutputStream(buffer); try { this.mapper.writeValue(outputStream, value); - } - catch (IOException e) { + } 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 index a1fffb1b9e..3a2887e21f 100644 --- 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 @@ -16,20 +16,7 @@ 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; @@ -37,6 +24,17 @@ import org.springframework.reactive.codec.decoder.Jaxb2Decoder; import org.springframework.reactive.io.BufferOutputStream; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; +import reactor.Publishers; +import reactor.io.buffer.Buffer; + +import javax.xml.bind.JAXBContext; +import javax.xml.bind.JAXBException; +import javax.xml.bind.MarshalException; +import javax.xml.bind.Marshaller; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; /** * Encode from an {@code Object} stream to a byte stream of XML elements. @@ -55,7 +53,7 @@ public class Jaxb2Encoder implements MessageToByteEncoder { @Override public Publisher encode(Publisher messageStream, ResolvableType type, MediaType mediaType, Object... hints) { - return Streams.wrap(messageStream).map(value -> { + return Publishers.map(messageStream, value -> { try { Buffer buffer = new Buffer(); BufferOutputStream outputStream = new BufferOutputStream(buffer); @@ -65,14 +63,12 @@ public class Jaxb2Encoder implements MessageToByteEncoder { marshaller.marshal(value, outputStream); buffer.flip(); return buffer.byteBuffer(); - } - catch (MarshalException ex) { + } catch (MarshalException ex) { throw new CodecException( - "Could not marshal [" + value + "]: " + ex.getMessage(), ex); - } - catch (JAXBException ex) { + "Could not marshal [" + value + "]: " + ex.getMessage(), ex); + } catch (JAXBException ex) { throw new CodecException( - "Could not instantiate JAXBContext: " + ex.getMessage(), ex); + "Could not instantiate JAXBContext: " + ex.getMessage(), ex); } }); } 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 index a27adf6e36..487007f7d3 100644 --- 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 @@ -16,17 +16,16 @@ 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; +import reactor.Publishers; + +import java.nio.ByteBuffer; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; /** * Encode from a String stream to a bytes stream. @@ -47,7 +46,7 @@ public class StringEncoder implements MessageToByteEncoder { @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))); + return Publishers.map(elementStream, s -> ByteBuffer.wrap(s.getBytes(charset))); } } diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/DispatcherHandler.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/DispatcherHandler.java index a7bc0904ef..8d1fe02ab9 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/DispatcherHandler.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/DispatcherHandler.java @@ -15,15 +15,9 @@ */ package org.springframework.reactive.web.dispatch; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; -import reactor.rx.Streams; - import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactoryUtils; import org.springframework.context.ApplicationContext; @@ -33,6 +27,11 @@ import org.springframework.http.HttpStatus; import org.springframework.reactive.web.http.HttpHandler; import org.springframework.reactive.web.http.ServerHttpRequest; import org.springframework.reactive.web.http.ServerHttpResponse; +import reactor.Publishers; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; /** * Central dispatcher for HTTP request handlers/controllers. Dispatches to registered @@ -102,7 +101,7 @@ public class DispatcherHandler implements HttpHandler, ApplicationContextAware { if (handler == null) { // No exception handling mechanism yet response.setStatusCode(HttpStatus.NOT_FOUND); - return Streams.empty(); + return Publishers.empty(); } HandlerAdapter handlerAdapter = getHandlerAdapter(handler); @@ -114,11 +113,11 @@ public class DispatcherHandler implements HttpHandler, ApplicationContextAware { return resultHandler.handleResult(request, response, result); } } - return Streams.fail(new IllegalStateException( - "No HandlerResultHandler for " + result.getValue())); + return Publishers.error(new IllegalStateException( + "No HandlerResultHandler for " + result.getValue())); } catch(Exception ex) { - return Streams.fail(ex); + return Publishers.error(ex); } } 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 index da6d1e1a93..f0607240ac 100644 --- 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 @@ -22,8 +22,10 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.reactivestreams.Publisher; +import reactor.Publishers; import reactor.rx.Promise; import reactor.rx.Stream; import reactor.rx.Streams; @@ -115,13 +117,13 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve } else { try { - return Streams.wrap(elementStream).next().await(); + return Publishers.toReadQueue(elementStream, 1, true).poll(30, TimeUnit.SECONDS); } catch(InterruptedException ex) { - return Streams.fail(new IllegalStateException("Timeout before getter the value")); + return Publishers.error(new IllegalStateException("Timeout before getter the value")); } } } - return Streams.fail(new IllegalStateException("Argument type not supported: " + type)); + return Publishers.error(new IllegalStateException("Argument type not supported: " + type)); } private MediaType resolveMediaType(ServerHttpRequest request) { 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 00ccc67ace..9e4431d116 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 @@ -15,22 +15,7 @@ */ package org.springframework.reactive.web.dispatch.method.annotation; -import java.lang.reflect.Method; -import java.lang.reflect.Type; -import java.nio.ByteBuffer; -import java.nio.charset.Charset; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.CompletableFuture; - import org.reactivestreams.Publisher; -import reactor.rx.Promise; -import reactor.rx.Streams; -import rx.Observable; -import rx.RxReactiveStreams; -import rx.Single; - import org.springframework.core.MethodParameter; import org.springframework.core.Ordered; import org.springframework.core.ParameterizedTypeReference; @@ -46,6 +31,19 @@ import org.springframework.reactive.web.http.ServerHttpRequest; import org.springframework.reactive.web.http.ServerHttpResponse; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.method.HandlerMethod; +import reactor.Publishers; +import reactor.rx.Promise; +import rx.Observable; +import rx.RxReactiveStreams; +import rx.Single; + +import java.lang.reflect.Type; +import java.nio.ByteBuffer; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; /** @@ -104,7 +102,7 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered MethodParameter returnType = handlerMethod.getReturnValueType(value); if (value == null) { - return Streams.empty(); + return Publishers.empty(); } MediaType mediaType = resolveMediaType(request); @@ -132,7 +130,7 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered elementStream = (Publisher)value; } else { - elementStream = Streams.just(value); + elementStream = Publishers.just(value); } Publisher outputStream = serializer.encode(elementStream, type, mediaType, hints.toArray()); @@ -141,10 +139,10 @@ public class ResponseBodyResultHandler implements HandlerResultHandler, Ordered outputStream = postProcessor.encode(outputStream, type, mediaType, hints.toArray()); } response.getHeaders().setContentType(mediaType); - return response.writeWith(Streams.wrap(outputStream)); + return response.writeWith(outputStream); } - return Streams.fail(new IllegalStateException( - "Return value type not supported: " + returnType)); + return Publishers.error(new IllegalStateException( + "Return value type not supported: " + returnType)); } private MediaType resolveMediaType(ServerHttpRequest request) { diff --git a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java index 647994c6bb..66761cabd5 100644 --- a/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java +++ b/spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java @@ -20,10 +20,10 @@ import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatus; import org.springframework.reactive.web.http.ServerHttpResponse; import org.springframework.util.Assert; +import reactor.Publishers; import reactor.io.buffer.Buffer; import reactor.io.net.http.HttpChannel; import reactor.io.net.http.model.Status; -import reactor.rx.Streams; import java.nio.ByteBuffer; @@ -59,7 +59,7 @@ public class ReactorServerHttpResponse implements ServerHttpResponse { @Override public Publisher writeWith(Publisher contentPublisher) { writeHeaders(); - return this.channel.writeWith(Streams.wrap(contentPublisher).map(Buffer::new)); + return this.channel.writeWith(Publishers.map(contentPublisher, Buffer::new)); } private void writeHeaders() {