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

Switch to Reactor 3 snapshots and Netty 4.1.3

Browse Source
This commit is contained in:
Rossen Stoyanchev
2016-07-15 17:15:33 -04:00
parent dc1664939c
commit 028be2a298
50 changed files with 113 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractRequestBodyPublisher.java
View File

@@ -28,7 +28,7 @@ import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.util.BackpressureUtils;
import reactor.core.publisher.Operators;
import org.springframework.core.io.buffer.DataBuffer;
@@ -106,7 +106,7 @@ abstract class AbstractRequestBodyPublisher implements Publisher<DataBuffer> {
while (hasDemand()) {
DataBuffer dataBuffer = read();
if (dataBuffer != null) {
BackpressureUtils.getAndSub(this.demand, 1L);
Operators.getAndSub(this.demand, 1L);
this.subscriber.onNext(dataBuffer);
}
else {
@@ -214,8 +214,8 @@ abstract class AbstractRequestBodyPublisher implements Publisher<DataBuffer> {
NO_DEMAND {
@Override
void request(AbstractRequestBodyPublisher publisher, long n) {
if (BackpressureUtils.checkRequest(n, publisher.subscriber)) {
BackpressureUtils.addAndGet(publisher.demand, n);
if (Operators.checkRequest(n, publisher.subscriber)) {
Operators.addAndGet(publisher.demand, n);
if (publisher.changeState(this, DEMAND)) {
publisher.checkOnDataAvailable();
}
@@ -249,8 +249,8 @@ abstract class AbstractRequestBodyPublisher implements Publisher<DataBuffer> {
READING {
@Override
void request(AbstractRequestBodyPublisher publisher, long n) {
if (BackpressureUtils.checkRequest(n, publisher.subscriber)) {
BackpressureUtils.addAndGet(publisher.demand, n);
if (Operators.checkRequest(n, publisher.subscriber)) {
Operators.addAndGet(publisher.demand, n);
}
}
},

8
spring-web/src/main/java/org/springframework/http/server/reactive/ChannelSendOperator.java
View File

@@ -21,8 +21,8 @@ import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.MonoSource;
import reactor.core.subscriber.SubscriberBarrier;
import reactor.core.util.EmptySubscription;
import reactor.core.publisher.OperatorAdapter;
import reactor.core.publisher.Operators;
import org.springframework.util.Assert;
@@ -54,7 +54,7 @@ public class ChannelSendOperator<T> extends MonoSource<T, Void> {
source.subscribe(new WriteWithBarrier(s));
}
private class WriteWithBarrier extends SubscriberBarrier<T, Void> implements Publisher<T> {
private class WriteWithBarrier extends OperatorAdapter<T, Void> implements Publisher<T> {
/**
* We've at at least one emission, we've called the write function, the write
@@ -161,7 +161,7 @@ public class ChannelSendOperator<T> extends MonoSource<T, Void> {
this.writeSubscriber = writeSubscriber;
if (this.error != null || this.completed) {
this.writeSubscriber.onSubscribe(EmptySubscription.INSTANCE);
this.writeSubscriber.onSubscribe(Operators.emptySubscription());
emitCachedSignals();
}
else {

4
spring-web/src/main/java/org/springframework/http/server/reactive/ResponseBodyWriteResultPublisher.java
View File

@@ -24,7 +24,7 @@ import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.util.BackpressureUtils;
import reactor.core.publisher.Operators;
/**
* Publisher returned from {@link ServerHttpResponse#writeWith(Publisher)}.
@@ -145,7 +145,7 @@ class ResponseBodyWriteResultPublisher implements Publisher<Void> {
SUBSCRIBED {
@Override
void request(ResponseBodyWriteResultPublisher publisher, long n) {
BackpressureUtils.checkRequest(n, publisher.subscriber);
Operators.checkRequest(n, publisher.subscriber);
}
@Override

4
spring-web/src/main/java/org/springframework/http/server/reactive/RxNettyHttpHandlerAdapter.java
View File

@@ -21,7 +21,7 @@ import io.reactivex.netty.protocol.http.server.HttpServerRequest;
import io.reactivex.netty.protocol.http.server.HttpServerResponse;
import io.reactivex.netty.protocol.http.server.RequestHandler;
import org.reactivestreams.Publisher;
import reactor.core.converter.RxJava1ObservableConverter;
import reactor.adapter.RxJava1Adapter;
import rx.Observable;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
@@ -50,7 +50,7 @@ public class RxNettyHttpHandlerAdapter implements RequestHandler<ByteBuf, ByteBu
RxNettyServerHttpResponse adaptedResponse =
new RxNettyServerHttpResponse(response, dataBufferFactory);
Publisher<Void> result = this.httpHandler.handle(adaptedRequest, adaptedResponse);
return RxJava1ObservableConverter.fromPublisher(result);
return RxJava1Adapter.publisherToObservable(result);
}
}

4
spring-web/src/main/java/org/springframework/http/server/reactive/RxNettyServerHttpRequest.java
View File

@@ -22,7 +22,7 @@ import java.net.URISyntaxException;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.cookie.Cookie;
import io.reactivex.netty.protocol.http.server.HttpServerRequest;
import reactor.core.converter.RxJava1ObservableConverter;
import reactor.adapter.RxJava1Adapter;
import reactor.core.publisher.Flux;
import rx.Observable;
@@ -95,7 +95,7 @@ public class RxNettyServerHttpRequest extends AbstractServerHttpRequest {
@Override
public Flux<DataBuffer> getBody() {
Observable<DataBuffer> content = this.request.getContent().map(dataBufferFactory::wrap);
return RxJava1ObservableConverter.toPublisher(content);
return RxJava1Adapter.observableToFlux(content);
}
}

12
spring-web/src/main/java/org/springframework/http/server/reactive/RxNettyServerHttpResponse.java
View File

@@ -23,8 +23,9 @@ import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.cookie.Cookie;
import io.netty.handler.codec.http.cookie.DefaultCookie;
import io.reactivex.netty.protocol.http.server.HttpServerResponse;
import io.reactivex.netty.protocol.http.server.ResponseContentWriter;
import org.reactivestreams.Publisher;
import reactor.core.converter.RxJava1ObservableConverter;
import reactor.adapter.RxJava1Adapter;
import reactor.core.publisher.Mono;
import rx.Observable;
@@ -72,12 +73,15 @@ public class RxNettyServerHttpResponse extends AbstractServerHttpResponse {
@Override
protected Mono<Void> writeWithInternal(Publisher<DataBuffer> body) {
Observable<ByteBuf> content = RxJava1ObservableConverter.fromPublisher(body).map(this::toByteBuf);
return RxJava1ObservableConverter.toPublisher(this.response.write(content, bb -> bb instanceof FlushingByteBuf)).then();
Observable<ByteBuf> content = RxJava1Adapter.publisherToObservable(body).map(this::toByteBuf);
ResponseContentWriter<ByteBuf> writer = this.response.write(content, bb -> bb instanceof FlushingByteBuf);
return RxJava1Adapter.observableToFlux(writer).then();
}
private ByteBuf toByteBuf(DataBuffer buffer) {
ByteBuf byteBuf = (buffer instanceof NettyDataBuffer ? ((NettyDataBuffer) buffer).getNativeBuffer() : Unpooled.wrappedBuffer(buffer.asByteBuffer()));
ByteBuf byteBuf = (buffer instanceof NettyDataBuffer ?
((NettyDataBuffer) buffer).getNativeBuffer() :
Unpooled.wrappedBuffer(buffer.asByteBuffer()));
return (buffer instanceof FlushingDataBuffer ? new FlushingByteBuf(byteBuf) : byteBuf);
}

7
spring-web/src/main/java/org/springframework/web/client/reactive/support/RxJava1ClientWebRequestBuilder.java
View File

@@ -18,8 +18,7 @@ package org.springframework.web.client.reactive.support;
import java.net.URI;
import reactor.core.converter.RxJava1ObservableConverter;
import reactor.core.converter.RxJava1SingleConverter;
import reactor.adapter.RxJava1Adapter;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.Single;
@@ -149,7 +148,7 @@ public class RxJava1ClientWebRequestBuilder implements ClientWebRequestBuilder {
* as type information for the element published by this reactive stream
*/
public RxJava1ClientWebRequestBuilder body(Single<?> content, ResolvableType elementType) {
this.delegate.body(RxJava1SingleConverter.toPublisher(content), elementType);
this.delegate.body(RxJava1Adapter.singleToMono(content), elementType);
return this;
}
@@ -158,7 +157,7 @@ public class RxJava1ClientWebRequestBuilder implements ClientWebRequestBuilder {
* as type information for the elements published by this reactive stream
*/
public RxJava1ClientWebRequestBuilder body(Observable<?> content, ResolvableType elementType) {
this.delegate.body(RxJava1ObservableConverter.toPublisher(content), elementType);
this.delegate.body(RxJava1Adapter.observableToFlux(content), elementType);
return this;
}

31
spring-web/src/main/java/org/springframework/web/client/reactive/support/RxJava1ResponseExtractors.java
View File

@@ -19,6 +19,12 @@ package org.springframework.web.client.reactive.support;
import java.util.List;
import java.util.Optional;
import reactor.adapter.RxJava1Adapter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.Single;
import org.springframework.core.ResolvableType;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
@@ -27,13 +33,6 @@ import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.web.client.reactive.ResponseExtractor;
import reactor.core.converter.RxJava1ObservableConverter;
import reactor.core.converter.RxJava1SingleConverter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.Single;
/**
* Static factory methods for {@link ResponseExtractor}
* based on the {@link Observable} and {@link Single} APIs.
@@ -50,8 +49,8 @@ public class RxJava1ResponseExtractors {
public static <T> ResponseExtractor<Single<T>> body(Class<T> sourceClass) {
ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
return (clientResponse, messageConverters) -> (Single<T>) RxJava1SingleConverter
.fromPublisher(clientResponse
return (clientResponse, messageConverters) -> (Single<T>) RxJava1Adapter
.publisherToSingle(clientResponse
.flatMap(resp -> decodeResponseBody(resp, resolvableType, messageConverters)).next());
}
@@ -61,8 +60,8 @@ public class RxJava1ResponseExtractors {
public static <T> ResponseExtractor<Observable<T>> bodyStream(Class<T> sourceClass) {
ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
return (clientResponse, messageConverters) -> RxJava1ObservableConverter
.fromPublisher(clientResponse
return (clientResponse, messageConverters) -> RxJava1Adapter
.publisherToObservable(clientResponse
.flatMap(resp -> decodeResponseBody(resp, resolvableType, messageConverters)));
}
@@ -75,7 +74,7 @@ public class RxJava1ResponseExtractors {
ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
return (clientResponse, messageConverters) ->
RxJava1SingleConverter.fromPublisher(clientResponse
RxJava1Adapter.publisherToSingle(clientResponse
.then(response ->
Mono.when(
decodeResponseBody(response, resolvableType, messageConverters).next(),
@@ -91,9 +90,9 @@ public class RxJava1ResponseExtractors {
*/
public static <T> ResponseExtractor<Single<ResponseEntity<Observable<T>>>> responseStream(Class<T> sourceClass) {
ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
return (clientResponse, messageConverters) -> RxJava1SingleConverter.fromPublisher(
return (clientResponse, messageConverters) -> RxJava1Adapter.publisherToSingle(
clientResponse.map(response -> new ResponseEntity<>(
RxJava1ObservableConverter.fromPublisher(
RxJava1Adapter.publisherToObservable(
RxJava1ResponseExtractors.<T> decodeResponseBody(response, resolvableType, messageConverters)),
response.getHeaders(),
response.getStatusCode())));
@@ -103,8 +102,8 @@ public class RxJava1ResponseExtractors {
* Extract the response headers as an {@code HttpHeaders} instance.
*/
public static ResponseExtractor<Single<HttpHeaders>> headers() {
return (clientResponse, messageConverters) -> RxJava1SingleConverter
.fromPublisher(clientResponse.map(resp -> resp.getHeaders()));
return (clientResponse, messageConverters) -> RxJava1Adapter
.publisherToSingle(clientResponse.map(resp -> resp.getHeaders()));
}
@SuppressWarnings("unchecked")