Lease support#648
Conversation
6743589 to
dcf6806
Compare
Signed-off-by: Maksym Ostroverkhov <m.ostroverkhov@gmail.com>
1b01b9f to
512080f
Compare
robertroeser
left a comment
There was a problem hiding this comment.
After spending more time looking at this I wonder if this couldn’t be simplified into something like Function<LeaseStats, Flux> that produces a flux of leases
and has a stats object that holds information about leases, one class that handles leases in the requester, and one class that handles releases in the responder. I don’t know if you’d need more than 3 classes. I would just return a boolean instead of null or object if leases are ok on the requestor side. Instead of null Request Lease handler I think it would be better to just have a no-op handler that always returned true. Same thing on the responder side - just a no-op responder lease handler if leases are disabled. The responder side would subscribe to the stream of leases, map over it, update its leases, and send them off with the duplex connection, etc. The responder lease handle needs like two methods, one too check if there are releases, and another that that takes a Function<LeaseStats, Flux>
| if (leaseHandler != null) { | ||
| MissingLeaseException leaseError = leaseHandler.useLease(); | ||
| if (leaseError != null) { | ||
| return Mono.error(leaseError); |
There was a problem hiding this comment.
need to release the payload here
# Conflicts: # gradle.properties # rsocket-core/src/main/java/io/rsocket/RSocketFactory.java # rsocket-core/src/main/java/io/rsocket/RSocketResponder.java # rsocket-core/src/test/java/io/rsocket/KeepAliveTest.java
…ber only wrap requester rsockets in multi subscription rsocket, except lease case pluggable stats
| sendProcessor.onNext(requestFrame); | ||
|
|
||
| return Mono.empty(); | ||
| } |
There was a problem hiding this comment.
The request* methods should probably surrounded by a try/catch that catches a throwable so if there is an exception it can be handled.
Like:
`Throwable err = checkAvailable();
if (err != null) {
return Mono.error(err);
}
try {
int streamId = streamIdSupplier.nextStreamId();
final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
final ByteBuf requestFrame =
RequestResponseFrameFlyweight.encode(
allocator,
streamId,
false,
payload.sliceMetadata().retain(),
payload.sliceData().retain());
payload.release();
UnicastMonoProcessor<Payload> receiver = UnicastMonoProcessor.create();
receivers.put(streamId, receiver);
sendProcessor.onNext(requestFrame);
return receiver
.doOnError(t -> sendProcessor.onNext(ErrorFrameFlyweight.encode(allocator, streamId, t)))
.doFinally(
s -> {
if (s == SignalType.CANCEL) {
sendProcessor.onNext(CancelFrameFlyweight.encode(allocator, streamId));
}
receivers.remove(streamId);
});
} catch (Throwable t) {
throw Exceptions.propogate(t);
}
`
There was a problem hiding this comment.
Agree. There is another question after looking at this approach yesterday and now: if defer is removed completely, and singleSubscriberRequester enabled in RSocketFactory, not only termination and lease check are performed eagerly, but frames are sent on call and not subscription.
There was a problem hiding this comment.
So I think I'll send frames in receiver's onSubscribe?
| Leases.<NoopStats>create() | ||
| .sender(new LeaseSender(SERVER_TAG, 7_000, 5)) | ||
| .receiver(new LeaseReceiver(SERVER_TAG)) | ||
| .stats(new NoopStats())) |
robertroeser
left a comment
There was a problem hiding this comment.
This looks good. I think the last thing to do is to wrap the request* methods on the requester side with try/catch blocks so they return Mono and Flux wrapped exceptions.
…uests are lazy;
make sure above can be subscribed at most once, and send at most 1 request frame
|
@mostroverkhov I'd love to add this to the CLI and the Demo Server (spewing tweets), any suggestions for a useful pattern for the leases? Or do you think it's only relevant with load balanced high availability situations? Great work, amazing to see the progress of the project towards a complete implementation of the Spec. |
|
@yschimke Yes, It mainly intended for load-balanced RSockets. Maybe Lease listener case on client showing incoming leases while making given N requests per second. |
No description provided.