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

Remove use of MonoProcessor.fromSinks

Browse Source 
See gh-25884
This commit is contained in:
Rossen Stoyanchev
2020-10-09 20:45:27 +01:00
parent cdd48ddd7f
commit e73e489fd8
30 changed files with 300 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/MessagingRSocket.java
View File

@@ -27,8 +27,6 @@ import io.rsocket.frame.FrameType;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.Sinks;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
@@ -162,8 +160,9 @@ class MessagingRSocket implements RSocket {
((NettyDataBuffer) dataBuffer).getNativeBuffer().refCnt() : 1;
}
@SuppressWarnings("deprecation")
private Flux<Payload> handleAndReply(Payload firstPayload, FrameType frameType, Flux<Payload> payloads) {
MonoProcessor<Flux<Payload>> replyMono = MonoProcessor.fromSink(Sinks.one());
reactor.core.publisher.MonoProcessor<Flux<Payload>> replyMono = reactor.core.publisher.MonoProcessor.create();
MessageHeaders headers = createHeaders(firstPayload, frameType, replyMono);
AtomicBoolean read = new AtomicBoolean();
@@ -186,8 +185,9 @@ class MessagingRSocket implements RSocket {
return PayloadUtils.retainDataAndReleasePayload(payload, this.strategies.dataBufferFactory());
}
@SuppressWarnings("deprecation")
private MessageHeaders createHeaders(Payload payload, FrameType frameType,
@Nullable MonoProcessor<?> replyMono) {
@Nullable reactor.core.publisher.MonoProcessor<?> replyMono) {
MessageHeaderAccessor headers = new MessageHeaderAccessor();
headers.setLeaveMutable(true);

20
spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketPayloadReturnValueHandler.java
View File

@@ -21,7 +21,6 @@ import java.util.List;
import io.rsocket.Payload;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import org.springframework.core.MethodParameter;
import org.springframework.core.ReactiveAdapterRegistry;
@@ -36,7 +35,7 @@ import org.springframework.util.Assert;
/**
* Extension of {@link AbstractEncoderMethodReturnValueHandler} that
* {@link #handleEncodedContent handles} encoded content by wrapping data buffers
* as RSocket payloads and by passing those to the {@link MonoProcessor}
* as RSocket payloads and by passing those to the {@link reactor.core.publisher.MonoProcessor}
* from the {@link #RESPONSE_HEADER} header.
*
* @author Rossen Stoyanchev
@@ -45,7 +44,7 @@ import org.springframework.util.Assert;
public class RSocketPayloadReturnValueHandler extends AbstractEncoderMethodReturnValueHandler {
/**
* Message header name that is expected to have a {@link MonoProcessor}
* Message header name that is expected to have a {@link reactor.core.publisher.MonoProcessor}
* which will receive the {@code Flux<Payload>} that represents the response.
*/
public static final String RESPONSE_HEADER = "rsocketResponse";
@@ -57,11 +56,11 @@ public class RSocketPayloadReturnValueHandler extends AbstractEncoderMethodRetur
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings({"unchecked", "deprecation"})
protected Mono<Void> handleEncodedContent(
Flux<DataBuffer> encodedContent, MethodParameter returnType, Message<?> message) {
MonoProcessor<Flux<Payload>> replyMono = getReplyMono(message);
reactor.core.publisher.MonoProcessor<Flux<Payload>> replyMono = getReplyMono(message);
Assert.notNull(replyMono, "Missing '" + RESPONSE_HEADER + "'");
replyMono.onNext(encodedContent.map(PayloadUtils::createPayload));
replyMono.onComplete();
@@ -69,8 +68,9 @@ public class RSocketPayloadReturnValueHandler extends AbstractEncoderMethodRetur
}
@Override
@SuppressWarnings("deprecation")
protected Mono<Void> handleNoContent(MethodParameter returnType, Message<?> message) {
MonoProcessor<Flux<Payload>> replyMono = getReplyMono(message);
reactor.core.publisher.MonoProcessor<Flux<Payload>> replyMono = getReplyMono(message);
if (replyMono != null) {
replyMono.onComplete();
}
@@ -78,11 +78,11 @@ public class RSocketPayloadReturnValueHandler extends AbstractEncoderMethodRetur
}
@Nullable
@SuppressWarnings("unchecked")
private MonoProcessor<Flux<Payload>> getReplyMono(Message<?> message) {
@SuppressWarnings({"unchecked", "deprecation"})
private reactor.core.publisher.MonoProcessor<Flux<Payload>> getReplyMono(Message<?> message) {
Object headerValue = message.getHeaders().get(RESPONSE_HEADER);
Assert.state(headerValue == null || headerValue instanceof MonoProcessor, "Expected MonoProcessor");
return (MonoProcessor<Flux<Payload>>) headerValue;
Assert.state(headerValue == null || headerValue instanceof reactor.core.publisher.MonoProcessor, "Expected MonoProcessor");
return (reactor.core.publisher.MonoProcessor<Flux<Payload>>) headerValue;
}
}

31
spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java
View File

@@ -19,8 +19,8 @@ package org.springframework.messaging.tcp.reactor;
import java.time.Duration;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import io.netty.buffer.ByteBuf;
@@ -33,7 +33,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
@@ -53,6 +52,7 @@ import org.springframework.messaging.tcp.TcpConnection;
import org.springframework.messaging.tcp.TcpConnectionHandler;
import org.springframework.messaging.tcp.TcpOperations;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.CompletableToListenableFutureAdapter;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.MonoToListenableFutureAdapter;
import org.springframework.util.concurrent.SettableListenableFuture;
@@ -205,13 +205,13 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
}
// Report first connect to the ListenableFuture
MonoProcessor<Void> connectMono = MonoProcessor.fromSink(Sinks.one());
CompletableFuture<Void> connectFuture = new CompletableFuture<>();
this.tcpClient
.handle(new ReactorNettyHandler(handler))
.connect()
.doOnNext(updateConnectMono(connectMono))
.doOnError(updateConnectMono(connectMono))
.doOnNext(conn -> connectFuture.complete(null))
.doOnError(connectFuture::completeExceptionally)
.doOnError(handler::afterConnectFailure) // report all connect failures to the handler
.flatMap(Connection::onDispose) // post-connect issues
.retryWhen(Retry.from(signals -> signals
@@ -222,7 +222,7 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
.flatMap(attempt -> reconnect(attempt, strategy)))
.subscribe();
return new MonoToListenableFutureAdapter<>(connectMono);
return new CompletableToListenableFutureAdapter<>(connectFuture);
}
private ListenableFuture<Void> handleShuttingDownConnectFailure(TcpConnectionHandler<P> handler) {
@@ -231,19 +231,6 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
return new MonoToListenableFutureAdapter<>(Mono.error(ex));
}
private <T> Consumer<T> updateConnectMono(MonoProcessor<Void> connectMono) {
return o -> {
if (!connectMono.isTerminated()) {
if (o instanceof Throwable) {
connectMono.onError((Throwable) o);
}
else {
connectMono.onComplete();
}
}
};
}
private Publisher<? extends Long> reconnect(Integer attempt, ReconnectStrategy reconnectStrategy) {
Long time = reconnectStrategy.getTimeToNextAttempt(attempt);
return (time != null ? Mono.delay(Duration.ofMillis(time), this.scheduler) : Mono.empty());
@@ -316,8 +303,8 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
logger.debug("Connected to " + conn.address());
}
});
MonoProcessor<Void> completion = MonoProcessor.fromSink(Sinks.one());
TcpConnection<P> connection = new ReactorNettyTcpConnection<>(inbound, outbound, codec, completion);
Sinks.Empty<Void> completionSink = Sinks.empty();
TcpConnection<P> connection = new ReactorNettyTcpConnection<>(inbound, outbound, codec, completionSink);
scheduler.schedule(() -> this.connectionHandler.afterConnected(connection));
inbound.withConnection(conn -> conn.addHandler(new StompMessageDecoder<>(codec)));
@@ -330,7 +317,7 @@ public class ReactorNettyTcpClient<P> implements TcpOperations<P> {
this.connectionHandler::handleFailure,
this.connectionHandler::afterConnectionClosed);
return completion;
return completionSink.asMono();
}
}

11
spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpConnection.java
View File

@@ -18,7 +18,7 @@ package org.springframework.messaging.tcp.reactor;
import io.netty.buffer.ByteBuf;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.Sinks;
import reactor.netty.NettyInbound;
import reactor.netty.NettyOutbound;
@@ -42,16 +42,16 @@ public class ReactorNettyTcpConnection<P> implements TcpConnection<P> {
private final ReactorNettyCodec<P> codec;
private final MonoProcessor<Void> closeProcessor;
private final Sinks.Empty<Void> completionSink;
public ReactorNettyTcpConnection(NettyInbound inbound, NettyOutbound outbound,
ReactorNettyCodec<P> codec, MonoProcessor<Void> closeProcessor) {
ReactorNettyCodec<P> codec, Sinks.Empty<Void> completionSink) {
this.inbound = inbound;
this.outbound = outbound;
this.codec = codec;
this.closeProcessor = closeProcessor;
this.completionSink = completionSink;
}
@@ -75,7 +75,8 @@ public class ReactorNettyTcpConnection<P> implements TcpConnection<P> {
@Override
public void close() {
this.closeProcessor.onComplete();
// Ignore result: can't overflow, ok if not first or no one listens
this.completionSink.tryEmitEmpty();
}
}

11
spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java
View File

@@ -28,7 +28,6 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
@@ -126,15 +125,15 @@ class RSocketServerToClientIntegrationTests {
static class ServerController {
// Must be initialized by @Test method...
volatile MonoProcessor<Void> result;
volatile Sinks.Empty<Void> resultSink;
void reset() {
this.result = MonoProcessor.fromSink(Sinks.one());
this.resultSink = Sinks.empty();
}
void await(Duration duration) {
this.result.block(duration);
this.resultSink.asMono().block(duration);
}
@@ -201,8 +200,8 @@ class RSocketServerToClientIntegrationTests {
private void runTest(Runnable testEcho) {
Mono.fromRunnable(testEcho)
.doOnError(ex -> result.onError(ex))
.doOnSuccess(o -> result.onComplete())
.doOnError(ex -> resultSink.emitError(ex, Sinks.EmitFailureHandler.FAIL_FAST))
.doOnSuccess(o -> resultSink.emitEmpty(Sinks.EmitFailureHandler.FAIL_FAST))
.subscribeOn(Schedulers.boundedElastic()) // StepVerifier will block
.subscribe();
}

23
spring-messaging/src/test/java/org/springframework/messaging/simp/annotation/support/SimpAnnotationMethodMessageHandlerTests.java
View File

@@ -34,7 +34,6 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.Sinks;
import org.springframework.context.support.StaticApplicationContext;
@@ -342,8 +341,8 @@ public class SimpAnnotationMethodMessageHandlerTests {
Message<?> message = createMessage("/app1/mono");
this.messageHandler.handleMessage(message);
assertThat(controller.monoProcessor).isNotNull();
controller.monoProcessor.onNext("foo");
assertThat(controller.oneSink).isNotNull();
controller.oneSink.emitValue("foo", Sinks.EmitFailureHandler.FAIL_FAST);
verify(this.converter).toMessage(this.payloadCaptor.capture(), any(MessageHeaders.class));
assertThat(this.payloadCaptor.getValue()).isEqualTo("foo");
}
@@ -357,7 +356,7 @@ public class SimpAnnotationMethodMessageHandlerTests {
Message<?> message = createMessage("/app1/mono");
this.messageHandler.handleMessage(message);
controller.monoProcessor.onError(new IllegalStateException());
controller.oneSink.emitError(new IllegalStateException(), Sinks.EmitFailureHandler.FAIL_FAST);
assertThat(controller.exceptionCaught).isTrue();
}
@@ -370,8 +369,8 @@ public class SimpAnnotationMethodMessageHandlerTests {
Message<?> message = createMessage("/app1/flux");
this.messageHandler.handleMessage(message);
assertThat(controller.fluxSink).isNotNull();
controller.fluxSink.tryEmitNext("foo");
assertThat(controller.manySink).isNotNull();
controller.manySink.tryEmitNext("foo");
verify(this.converter, never()).toMessage(any(), any(MessageHeaders.class));
}
@@ -585,22 +584,22 @@ public class SimpAnnotationMethodMessageHandlerTests {
@Controller
private static class ReactiveController {
private MonoProcessor<String> monoProcessor;
private Sinks.One<String> oneSink;
private Sinks.Many<String> fluxSink;
private Sinks.Many<String> manySink;
private boolean exceptionCaught = false;
@MessageMapping("mono")
public Mono<String> handleMono() {
this.monoProcessor = MonoProcessor.fromSink(Sinks.one());
return this.monoProcessor;
this.oneSink = Sinks.one();
return this.oneSink.asMono();
}
@MessageMapping("flux")
public Flux<String> handleFlux() {
this.fluxSink = Sinks.many().unicast().onBackpressureBuffer();
return this.fluxSink.asFlux();
this.manySink = Sinks.many().unicast().onBackpressureBuffer();
return this.manySink.asFlux();
}
@MessageExceptionHandler(IllegalStateException.class)