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() {