SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve RSocketForwardingFunction

Browse Source 
* Use `Mono<RSocket>` for lazy connection on target subscription
returned from the `RSocketForwardingFunction`
* Propagate `retry` into an `RSocketConnector`

Resolves #566
This commit is contained in:
Artem Bilan
2020-07-27 16:44:39 -04:00
committed by Oleg Zhurakousky
parent 6ca9c2f072
commit 8d316f906c
1 changed files with 19 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketForwardingFunction.java
View File

@@ -29,6 +29,7 @@ import io.rsocket.util.DefaultPayload;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher; import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry; import reactor.util.retry.Retry;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
@@ -36,41 +37,44 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder; import org.springframework.messaging.support.MessageBuilder;
/** /**
* *
* @author Oleg Zhurakousky * @author Oleg Zhurakousky
* @author Artem Bilan
*
* @since 3.1 * @since 3.1
* *
*/ */
class RSocketForwardingFunction implements Function<Message<byte[]>, Publisher<Message<byte[]>>> { class RSocketForwardingFunction implements Function<Message<byte[]>, Publisher<Message<byte[]>>> {
private static Log logger = LogFactory.getLog(RSocketForwardingFunction.class);
private final RSocket rSocket; private static final Log LOGGER = LogFactory.getLog(RSocketForwardingFunction.class);
private final Mono<RSocket> rsocketMono;
private final FunctionInvocationWrapper targetFunction; private final FunctionInvocationWrapper targetFunction;
RSocketForwardingFunction(FunctionInvocationWrapper targetFunction, InetSocketAddress outputAddress) { RSocketForwardingFunction(FunctionInvocationWrapper targetFunction, InetSocketAddress outputAddress) {
this.targetFunction = targetFunction; this.targetFunction = targetFunction;
this.rSocket = outputAddress == null ? null this.rsocketMono =
: RSocketConnector.connectWith(TcpClientTransport.create(outputAddress)) outputAddress == null
.log() ? null
.retryWhen(Retry.backoff(5, Duration.ofSeconds(1))) : RSocketConnector.create()
.block(); .reconnect(Retry.backoff(5, Duration.ofSeconds(1)))
.connect(TcpClientTransport.create(outputAddress));
} }
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Override @Override
public Publisher<Message<byte[]>> apply(Message<byte[]> input) { public Publisher<Message<byte[]>> apply(Message<byte[]> input) {
if (logger.isDebugEnabled()) { if (LOGGER.isDebugEnabled()) {
logger.debug("Executiing: " + this.targetFunction); LOGGER.debug("Executing: " + this.targetFunction);
} }
Object rawResult = this.targetFunction.apply(input); Object rawResult = this.targetFunction.apply(input);
Publisher<Message<byte[]>> resultMessage = this.rSocket return this.rsocketMono
.requestStream(DefaultPayload.create(((Message<byte[]>) rawResult).getPayload())) .flatMapMany((rsocket) ->
.map(this::buildResultMessage); rsocket.requestStream(DefaultPayload.create(((Message<byte[]>) rawResult).getPayload())))
return resultMessage; .map(this::buildResultMessage);
} }
private Message<byte[]> buildResultMessage(Payload payload) { private Message<byte[]> buildResultMessage(Payload payload) {
@@ -79,4 +83,5 @@ class RSocketForwardingFunction implements Function<Message<byte[]>, Publisher<M
payloadBuffer.get(payloadData); payloadBuffer.get(payloadData);
return MessageBuilder.withPayload(payloadData).build(); return MessageBuilder.withPayload(payloadData).build();
} }
} }