Skip to content

Commit dcf6806

Browse files
mostroverkhovmostroverkhov
committed
lease support
Signed-off-by: Maksym Ostroverkhov <m.ostroverkhov@gmail.com>
1 parent c0c3e90 commit dcf6806

34 files changed

Lines changed: 1949 additions & 106 deletions

‎gradle.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,4 +11,4 @@
1111
# See the License for the specific language governing permissions and
1212
# limitations under the License.
1313
#
14-
version=0.12.2-RC4-SNAPSHOT
14+
version=0.12.2-RC4-LEASE-SNAPSHOT

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

Lines changed: 48 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,9 @@
3131
import io.rsocket.internal.ClientSetup;
3232
import io.rsocket.internal.ServerSetup;
3333
import io.rsocket.keepalive.KeepAliveHandler;
34+
import io.rsocket.lease.LeaseOptions;
35+
import io.rsocket.lease.LeaseSender;
36+
import io.rsocket.lease.LeaseSupport;
3437
import io.rsocket.plugins.DuplexConnectionInterceptor;
3538
import io.rsocket.plugins.PluginRegistry;
3639
import io.rsocket.plugins.Plugins;
@@ -42,10 +45,7 @@
4245
import io.rsocket.util.EmptyPayload;
4346
import java.time.Duration;
4447
import java.util.Objects;
45-
import java.util.function.BiFunction;
46-
import java.util.function.Consumer;
47-
import java.util.function.Function;
48-
import java.util.function.Supplier;
48+
import java.util.function.*;
4949
import reactor.core.publisher.Mono;
5050

5151
/** Factory for creating RSocket clients and servers. */
@@ -123,6 +123,7 @@ public static class ClientRSocketFactory implements ClientTransportAcceptor {
123123
new ExponentialBackoffResumeStrategy(Duration.ofSeconds(1), Duration.ofSeconds(16), 2);
124124

125125
private ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
126+
private BiConsumer<LeaseSender, LeaseOptions> leaseSenderConsumer;
126127

127128
public ClientRSocketFactory byteBufAllocator(ByteBufAllocator allocator) {
128129
Objects.requireNonNull(allocator);
@@ -205,6 +206,11 @@ public ClientRSocketFactory metadataMimeType(String metadataMimeType) {
205206
return this;
206207
}
207208

209+
public ClientRSocketFactory lease(BiConsumer<LeaseSender, LeaseOptions> leaseSenderConsumer) {
210+
this.leaseSenderConsumer = leaseSenderConsumer;
211+
return this;
212+
}
213+
208214
public ClientRSocketFactory resume() {
209215
this.resumeEnabled = true;
210216
return this;
@@ -302,6 +308,10 @@ public Mono<RSocket> start() {
302308
ClientServerInputMultiplexer multiplexer =
303309
new ClientServerInputMultiplexer(wrappedConnection, plugins);
304310

311+
boolean isLeaseEnabled = leaseSenderConsumer != null;
312+
313+
LeaseSupport leaseSupport = new LeaseSupport(isLeaseEnabled, "client", allocator);
314+
305315
RSocketRequester rSocketRequester =
306316
new RSocketRequester(
307317
allocator,
@@ -311,12 +321,13 @@ public Mono<RSocket> start() {
311321
StreamIdSupplier.clientSupplier(),
312322
keepAliveTickPeriod(),
313323
keepAliveTimeout(),
314-
keepAliveHandler);
324+
keepAliveHandler,
325+
leaseSupport.requesterHandler());
315326

316327
ByteBuf setupFrame =
317328
SetupFrameFlyweight.encode(
318329
allocator,
319-
false,
330+
isLeaseEnabled,
320331
keepAliveTickPeriod(),
321332
keepAliveTimeout(),
322333
resumeToken,
@@ -343,9 +354,13 @@ public Mono<RSocket> start() {
343354
multiplexer.asServerConnection(),
344355
wrappedRSocketHandler,
345356
payloadDecoder,
346-
errorConsumer);
357+
errorConsumer,
358+
leaseSupport.responderHandler());
347359

348-
return wrappedConnection.sendOne(setupFrame).thenReturn(wrappedRSocketRequester);
360+
return wrappedConnection
361+
.sendOne(setupFrame)
362+
.thenReturn(wrappedRSocketRequester)
363+
.doOnSuccess(ignored -> leaseSupport.supplyLeaseSender(leaseSenderConsumer));
349364
});
350365
}
351366

@@ -395,6 +410,7 @@ public static class ServerRSocketFactory {
395410

396411
private ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
397412
private boolean resumeCleanupStoreOnKeepAlive;
413+
private BiConsumer<LeaseSender, LeaseOptions> leaseSenderConsumer;
398414

399415
private ServerRSocketFactory() {}
400416

@@ -450,6 +466,11 @@ public ServerRSocketFactory errorConsumer(Consumer<Throwable> errorConsumer) {
450466
return this;
451467
}
452468

469+
public ServerRSocketFactory lease(BiConsumer<LeaseSender, LeaseOptions> leaseSenderConsumer) {
470+
this.leaseSenderConsumer = leaseSenderConsumer;
471+
return this;
472+
}
473+
453474
public ServerRSocketFactory resume() {
454475
this.resumeSupported = true;
455476
return this;
@@ -541,6 +562,20 @@ private Mono<Void> acceptSetup(
541562
multiplexer.dispose();
542563
});
543564
}
565+
566+
boolean isLeaseEnabled = leaseSenderConsumer != null;
567+
568+
if (SetupFrameFlyweight.honorLease(setupFrame) && !isLeaseEnabled) {
569+
return sendError(multiplexer, new InvalidSetupException("lease is not supported"))
570+
.doFinally(
571+
signalType -> {
572+
setupFrame.release();
573+
multiplexer.dispose();
574+
});
575+
}
576+
577+
LeaseSupport leaseSupport = new LeaseSupport(isLeaseEnabled, "server", allocator);
578+
544579
return serverSetup.acceptRSocketSetup(
545580
setupFrame,
546581
multiplexer,
@@ -553,7 +588,8 @@ private Mono<Void> acceptSetup(
553588
wrappedMultiplexer.asServerConnection(),
554589
payloadDecoder,
555590
errorConsumer,
556-
StreamIdSupplier.serverSupplier());
591+
StreamIdSupplier.serverSupplier(),
592+
leaseSupport.requesterHandler());
557593

558594
RSocket wrappedRSocketRequester = plugins.applyRequester(rSocketRequester);
559595

@@ -574,8 +610,10 @@ private Mono<Void> acceptSetup(
574610
errorConsumer,
575611
setupPayload.keepAliveInterval(),
576612
setupPayload.keepAliveMaxLifetime(),
577-
keepAliveHandler);
613+
keepAliveHandler,
614+
leaseSupport.responderHandler());
578615
})
616+
.doOnSuccess(ignored -> leaseSupport.supplyLeaseSender(leaseSenderConsumer))
579617
.doFinally(signalType -> setupPayload.release())
580618
.then();
581619
});

0 commit comments

Comments
 (0)