Skip to content

Commit 00acd1e

Browse files
rstoyanchevrstoyanchev
authored
Deprecate ResumeStrategy and replace with Reactor Retry (#798)
1 parent 414193f commit 00acd1e

9 files changed

Lines changed: 96 additions & 69 deletions

File tree

‎rsocket-core/src/main/java/io/rsocket/RSocketFactory.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import io.netty.buffer.ByteBuf;
1919
import io.netty.buffer.ByteBufAllocator;
20+
import io.netty.buffer.Unpooled;
2021
import io.rsocket.core.RSocketConnector;
2122
import io.rsocket.core.RSocketServer;
2223
import io.rsocket.core.Resume;
@@ -26,6 +27,7 @@
2627
import io.rsocket.plugins.DuplexConnectionInterceptor;
2728
import io.rsocket.plugins.RSocketInterceptor;
2829
import io.rsocket.plugins.SocketAcceptorInterceptor;
30+
import io.rsocket.resume.ClientResume;
2931
import io.rsocket.resume.ResumableFramesStore;
3032
import io.rsocket.resume.ResumeStrategy;
3133
import io.rsocket.transport.ClientTransport;
@@ -97,6 +99,9 @@ default <T extends Closeable> Start<T> transport(ServerTransport<T> transport) {
9799

98100
/** Factory to create and configure an RSocket client, and connect to a server. */
99101
public static class ClientRSocketFactory implements ClientTransportAcceptor {
102+
private static final ClientResume CLIENT_RESUME =
103+
new ClientResume(Duration.ofMinutes(2), Unpooled.EMPTY_BUFFER);
104+
100105
private final RSocketConnector connector;
101106

102107
private Duration tickPeriod = Duration.ofSeconds(20);
@@ -348,9 +353,11 @@ public ClientRSocketFactory resumeStreamTimeout(Duration streamTimeout) {
348353
return this;
349354
}
350355

351-
public ClientRSocketFactory resumeStrategy(Supplier<ResumeStrategy> resumeStrategy) {
356+
public ClientRSocketFactory resumeStrategy(Supplier<ResumeStrategy> strategy) {
352357
resume();
353-
resume.resumeStrategy(resumeStrategy);
358+
resume.retry(
359+
Retry.from(
360+
signals -> signals.flatMap(s -> strategy.get().apply(CLIENT_RESUME, s.failure()))));
354361
return this;
355362
}
356363

‎rsocket-core/src/main/java/io/rsocket/core/RSocketConnector.java‎

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -254,8 +254,17 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
254254
DuplexConnection wrappedConnection;
255255

256256
if (resume != null) {
257-
ClientRSocketSession session = createSession(connection, connectionMono);
258-
resumeToken = session.token();
257+
resumeToken = resume.getTokenSupplier().get();
258+
ClientRSocketSession session =
259+
new ClientRSocketSession(
260+
connection,
261+
resume.getSessionDuration(),
262+
resume.getRetry(),
263+
resume.getStoreFactory(CLIENT_TAG).apply(resumeToken),
264+
resume.getStreamTimeout(),
265+
resume.isCleanupStoreOnKeepAlive())
266+
.continueWith(connectionMono)
267+
.resumeToken(resumeToken);
259268
keepAliveHandler =
260269
new KeepAliveHandler.ResumableKeepAliveHandler(session.resumableConnection());
261270
wrappedConnection = session.resumableConnection();
@@ -343,18 +352,4 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
343352
}
344353
});
345354
}
346-
347-
private ClientRSocketSession createSession(
348-
DuplexConnection connection, Mono<DuplexConnection> connectionMono) {
349-
ByteBuf resumeToken = resume.getTokenSupplier().get();
350-
ClientRSocketSession session =
351-
new ClientRSocketSession(
352-
connection,
353-
resume.getSessionDuration(),
354-
resume.getResumeStrategySupplier(),
355-
resume.getStoreFactory(CLIENT_TAG).apply(resumeToken),
356-
resume.getStreamTimeout(),
357-
resume.isCleanupStoreOnKeepAlive());
358-
return session.continueWith(connectionMono).resumeToken(resumeToken);
359-
}
360355
}

‎rsocket-core/src/main/java/io/rsocket/core/Resume.java‎

Lines changed: 18 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -17,23 +17,29 @@
1717

1818
import io.netty.buffer.ByteBuf;
1919
import io.rsocket.frame.ResumeFrameFlyweight;
20-
import io.rsocket.resume.ExponentialBackoffResumeStrategy;
2120
import io.rsocket.resume.InMemoryResumableFramesStore;
2221
import io.rsocket.resume.ResumableFramesStore;
23-
import io.rsocket.resume.ResumeStrategy;
2422
import java.time.Duration;
2523
import java.util.function.Function;
2624
import java.util.function.Supplier;
25+
import org.slf4j.Logger;
26+
import org.slf4j.LoggerFactory;
27+
import reactor.util.retry.Retry;
2728

2829
public class Resume {
29-
Duration sessionDuration = Duration.ofMinutes(2);
30-
Duration streamTimeout = Duration.ofSeconds(10);
31-
boolean cleanupStoreOnKeepAlive;
32-
Function<? super ByteBuf, ? extends ResumableFramesStore> storeFactory;
30+
private static final Logger logger = LoggerFactory.getLogger(Resume.class);
31+
32+
private Duration sessionDuration = Duration.ofMinutes(2);
33+
private Duration streamTimeout = Duration.ofSeconds(10);
34+
private boolean cleanupStoreOnKeepAlive;
35+
private Function<? super ByteBuf, ? extends ResumableFramesStore> storeFactory;
3336

3437
private Supplier<ByteBuf> tokenSupplier = ResumeFrameFlyweight::generateResumeToken;
35-
private Supplier<ResumeStrategy> resumeStrategySupplier =
36-
() -> new ExponentialBackoffResumeStrategy(Duration.ofSeconds(1), Duration.ofSeconds(16), 2);
38+
private Retry retry =
39+
Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
40+
.maxBackoff(Duration.ofSeconds(16))
41+
.jitter(1.0)
42+
.doBeforeRetry(signal -> logger.debug("Connection error", signal.failure()));
3743

3844
public Resume() {}
3945

@@ -63,8 +69,8 @@ public Resume token(Supplier<ByteBuf> supplier) {
6369
return this;
6470
}
6571

66-
public Resume resumeStrategy(Supplier<ResumeStrategy> supplier) {
67-
this.resumeStrategySupplier = supplier;
72+
public Resume retry(Retry retry) {
73+
this.retry = retry;
6874
return this;
6975
}
7076

@@ -90,7 +96,7 @@ Supplier<ByteBuf> getTokenSupplier() {
9096
return tokenSupplier;
9197
}
9298

93-
Supplier<ResumeStrategy> getResumeStrategySupplier() {
94-
return resumeStrategySupplier;
99+
Retry getRetry() {
100+
return retry;
95101
}
96102
}

‎rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java‎

Lines changed: 30 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2015-2019 the original author or authors.
2+
* Copyright 2015-2020 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -26,10 +26,11 @@
2626
import io.rsocket.internal.ClientServerInputMultiplexer;
2727
import java.time.Duration;
2828
import java.util.concurrent.atomic.AtomicBoolean;
29-
import java.util.function.Supplier;
3029
import org.slf4j.Logger;
3130
import org.slf4j.LoggerFactory;
31+
import reactor.core.publisher.Flux;
3232
import reactor.core.publisher.Mono;
33+
import reactor.util.retry.Retry;
3334

3435
public class ClientRSocketSession implements RSocketSession<Mono<DuplexConnection>> {
3536
private static final Logger logger = LoggerFactory.getLogger(ClientRSocketSession.class);
@@ -42,7 +43,7 @@ public class ClientRSocketSession implements RSocketSession<Mono<DuplexConnectio
4243
public ClientRSocketSession(
4344
DuplexConnection duplexConnection,
4445
Duration resumeSessionDuration,
45-
Supplier<ResumeStrategy> resumeStrategy,
46+
Retry retry,
4647
ResumableFramesStore resumableFramesStore,
4748
Duration resumeStreamTimeout,
4849
boolean cleanupStoreOnKeepAlive) {
@@ -63,24 +64,13 @@ public ClientRSocketSession(
6364
.flatMap(
6465
err -> {
6566
logger.debug("Client session connection error. Starting new connection");
66-
ResumeStrategy reconnectOnError = resumeStrategy.get();
67-
ClientResume clientResume = new ClientResume(resumeSessionDuration, resumeToken);
6867
AtomicBoolean once = new AtomicBoolean();
6968
return newConnection
7069
.delaySubscription(
7170
once.compareAndSet(false, true)
72-
? reconnectOnError.apply(clientResume, err)
71+
? retry.generateCompanion(Flux.just(new RetrySignal(err)))
7372
: Mono.empty())
74-
.retryWhen(
75-
errors ->
76-
errors
77-
.doOnNext(
78-
retryErr ->
79-
logger.debug("Resumption reconnection error", retryErr))
80-
.flatMap(
81-
retryErr ->
82-
Mono.from(reconnectOnError.apply(clientResume, retryErr))
83-
.doOnNext(v -> logger.debug("Retrying with: {}", v))))
73+
.retryWhen(retry)
8474
.timeout(resumeSessionDuration);
8575
})
8676
.map(ClientServerInputMultiplexer::new)
@@ -177,4 +167,28 @@ private static long remotePos(ByteBuf resumeOkFrame) {
177167
private static ConnectionErrorException errorFrameThrowable(long impliedPos) {
178168
return new ConnectionErrorException("resumption_server_pos=[" + impliedPos + "]");
179169
}
170+
171+
private static class RetrySignal implements Retry.RetrySignal {
172+
173+
private final Throwable ex;
174+
175+
RetrySignal(Throwable ex) {
176+
this.ex = ex;
177+
}
178+
179+
@Override
180+
public long totalRetries() {
181+
return 0;
182+
}
183+
184+
@Override
185+
public long totalRetriesInARow() {
186+
return 0;
187+
}
188+
189+
@Override
190+
public Throwable failure() {
191+
return ex;
192+
}
193+
}
180194
}

‎rsocket-core/src/main/java/io/rsocket/resume/ExponentialBackoffResumeStrategy.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,13 @@
2121
import org.reactivestreams.Publisher;
2222
import reactor.core.publisher.Flux;
2323
import reactor.core.publisher.Mono;
24+
import reactor.util.retry.Retry;
2425

26+
/**
27+
* @deprecated as of 1.0 RC7 in favor of passing {@link Retry#backoff(long, Duration)} to {@link
28+
* io.rsocket.core.Resume#retry(Retry)}.
29+
*/
30+
@Deprecated
2531
public class ExponentialBackoffResumeStrategy implements ResumeStrategy {
2632
private volatile Duration next;
2733
private final Duration firstBackoff;

‎rsocket-core/src/main/java/io/rsocket/resume/PeriodicResumeStrategy.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,13 @@
1919
import java.time.Duration;
2020
import org.reactivestreams.Publisher;
2121
import reactor.core.publisher.Mono;
22+
import reactor.util.retry.Retry;
2223

24+
/**
25+
* @deprecated as of 1.0 RC7 in favor of passing {@link Retry#fixedDelay(long, Duration)} to {@link
26+
* io.rsocket.core.Resume#retry(Retry)}.
27+
*/
28+
@Deprecated
2329
public class PeriodicResumeStrategy implements ResumeStrategy {
2430
private final Duration interval;
2531

‎rsocket-core/src/main/java/io/rsocket/resume/ResumeStrategy.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2015-2019 the original author or authors.
2+
* Copyright 2015-2020 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -18,6 +18,12 @@
1818

1919
import java.util.function.BiFunction;
2020
import org.reactivestreams.Publisher;
21+
import reactor.util.retry.Retry;
2122

23+
/**
24+
* @deprecated as of 1.0 RC7 in favor of using {@link io.rsocket.core.Resume#retry(Retry)} via
25+
* {@link io.rsocket.core.RSocketConnector} or {@link io.rsocket.core.RSocketServer}.
26+
*/
27+
@Deprecated
2228
@FunctionalInterface
2329
public interface ResumeStrategy extends BiFunction<ClientResume, Throwable, Publisher<?>> {}

‎rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/resume/ResumeFileTransfer.java‎

Lines changed: 6 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -22,17 +22,14 @@
2222
import io.rsocket.core.RSocketConnector;
2323
import io.rsocket.core.RSocketServer;
2424
import io.rsocket.core.Resume;
25-
import io.rsocket.resume.ClientResume;
26-
import io.rsocket.resume.PeriodicResumeStrategy;
27-
import io.rsocket.resume.ResumeStrategy;
2825
import io.rsocket.transport.netty.client.TcpClientTransport;
2926
import io.rsocket.transport.netty.server.CloseableChannel;
3027
import io.rsocket.transport.netty.server.TcpServerTransport;
3128
import io.rsocket.util.DefaultPayload;
3229
import java.time.Duration;
33-
import org.reactivestreams.Publisher;
3430
import reactor.core.publisher.Flux;
3531
import reactor.core.publisher.Mono;
32+
import reactor.util.retry.Retry;
3633

3734
public class ResumeFileTransfer {
3835
/*amount of file chunks requested by subscriber: n, refilled on n/2 of received items*/
@@ -43,8 +40,11 @@ public static void main(String[] args) {
4340
Resume resume =
4441
new Resume()
4542
.sessionDuration(Duration.ofMinutes(5))
46-
.resumeStrategy(
47-
() -> new VerboseResumeStrategy(new PeriodicResumeStrategy(Duration.ofSeconds(1))));
43+
.retry(
44+
Retry.fixedDelay(Long.MAX_VALUE, Duration.ofSeconds(1))
45+
.doBeforeRetry(
46+
retrySignal ->
47+
System.out.println("Disconnected. Trying to resume connection...")));
4848

4949
CloseableChannel server =
5050
RSocketServer.create((setup, rSocket) -> Mono.just(new FileServer(requestCodec)))
@@ -88,20 +88,6 @@ public Flux<Payload> requestStream(Payload payload) {
8888
}
8989
}
9090

91-
private static class VerboseResumeStrategy implements ResumeStrategy {
92-
private final ResumeStrategy resumeStrategy;
93-
94-
public VerboseResumeStrategy(ResumeStrategy resumeStrategy) {
95-
this.resumeStrategy = resumeStrategy;
96-
}
97-
98-
@Override
99-
public Publisher<?> apply(ClientResume clientResume, Throwable throwable) {
100-
return Flux.from(resumeStrategy.apply(clientResume, throwable))
101-
.doOnNext(v -> System.out.println("Disconnected. Trying to resume connection..."));
102-
}
103-
}
104-
10591
private static class RequestCodec {
10692

10793
public Payload encode(Request request) {

‎rsocket-examples/src/test/java/io/rsocket/resume/ResumeIntegrationTest.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import reactor.core.publisher.Mono;
4343
import reactor.core.scheduler.Schedulers;
4444
import reactor.test.StepVerifier;
45+
import reactor.util.retry.Retry;
4546

4647
@SlowTest
4748
public class ResumeIntegrationTest {
@@ -155,7 +156,7 @@ static ServerTransport<CloseableChannel> serverTransport(String host, int port)
155156
}
156157

157158
private static Flux<Payload> testRequest() {
158-
return Flux.interval(Duration.ofMillis(50))
159+
return Flux.interval(Duration.ofMillis(500))
159160
.map(v -> DefaultPayload.create("client_request"))
160161
.onBackpressureDrop();
161162
}
@@ -183,7 +184,7 @@ private static Mono<RSocket> newClientRSocket(
183184
.sessionDuration(Duration.ofSeconds(sessionDurationSeconds))
184185
.storeFactory(t -> new InMemoryResumableFramesStore("client", 500_000))
185186
.cleanupStoreOnKeepAlive()
186-
.resumeStrategy(() -> new PeriodicResumeStrategy(Duration.ofSeconds(1))))
187+
.retry(Retry.fixedDelay(Long.MAX_VALUE, Duration.ofSeconds(1))))
187188
.keepAlive(Duration.ofSeconds(5), Duration.ofMinutes(5))
188189
.connect(clientTransport);
189190
}

0 commit comments

Comments
 (0)