Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Progress #24

Browse Source 
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
This commit is contained in:
Stephane Maldini
2015-10-05 15:35:59 +01:00
parent 0dabdb8207
commit 464ff1d960
12 changed files with 121 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-web-reactive/build.gradle
View File

@@ -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"

25
spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoder.java
View File

@@ -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<Object> {
@Override
public Publisher<Object> decode(Publisher<ByteBuffer> 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);
}
});
}
}

43
spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/Jaxb2Decoder.java
View File

@@ -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<Object> {
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));
}
}

37
spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/JsonObjectDecoder.java
View File

@@ -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<ByteBuffer> {
@Override
public Publisher<ByteBuffer> decode(Publisher<ByteBuffer> inputStream, ResolvableType type, MediaType mediaType, Object... hints) {
return Streams.wrap(inputStream).flatMap(new Function<ByteBuffer, Publisher<? extends ByteBuffer>>() {
return Publishers.flatMap(inputStream, new Function<ByteBuffer, Publisher<? extends ByteBuffer>>() {
int openBraces;
int idx;
@@ -121,17 +120,17 @@ public class JsonObjectDecoder implements ByteToMessageDecoder<ByteBuffer> {
}
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<ByteBuffer> {
}
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);
}
/**

15
spring-web-reactive/src/main/java/org/springframework/reactive/codec/decoder/StringDecoder.java
View File

@@ -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<String> {
@Override
public Publisher<String> decode(Publisher<ByteBuffer> 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));
}
}

18
spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoder.java
View File

@@ -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<Object> {
@Override
public Publisher<ByteBuffer> encode(Publisher<? extends Object> messageStream, ResolvableType type, MediaType mediaType, Object... hints) {
Stream<ByteBuffer> 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;
}
}

36
spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/Jaxb2Encoder.java
View File

@@ -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<Object> {
@Override
public Publisher<ByteBuffer> encode(Publisher<? extends Object> 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<Object> {
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);
}
});
}

13
spring-web-reactive/src/main/java/org/springframework/reactive/codec/encoder/StringEncoder.java
View File

@@ -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<String> {
@Override
public Publisher<ByteBuffer> encode(Publisher<? extends String> 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)));
}
}

19
spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/DispatcherHandler.java
View File

@@ -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);
}
}

8
spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/RequestBodyArgumentResolver.java
View File

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

38
spring-web-reactive/src/main/java/org/springframework/reactive/web/dispatch/method/annotation/ResponseBodyResultHandler.java
View File

@@ -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<ByteBuffer> 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) {

4
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/reactor/ReactorServerHttpResponse.java
View File

@@ -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<Void> writeWith(Publisher<ByteBuffer> contentPublisher) {
writeHeaders();
return this.channel.writeWith(Streams.wrap(contentPublisher).map(Buffer::new));
return this.channel.writeWith(Publishers.map(contentPublisher, Buffer::new));
}
private void writeHeaders() {