Do not send lease frames for clients which do not honor leases.

Today server sends leases even though the client does not honor leases. There are two things here:

- A server should protect itself, even if the client dishonors leases, i.e. leases aren't infinite if the client does not honor. So, the server should still expect requests within capacity from clients and reject all other requests.
- Clients should not receive lease frames as they are not relevant.

Modified `ServerReactiveSocket` to not send leases when the setup is such that client does not honor leases.
Also, added an override in `SetupProvider` to be able to modify the `DisableLeaseSocket` if required.

Expected results and optimizations on the wire w.r.t data transfers.

Author: NiteshKant

reactivesocket-core/src/main/java/io/reactivesocket/ServerReactiveSocket.java

@@ -52,9 +52,8 @@ public class ServerReactiveSocket implements ReactiveSocket {
 
     private final ReactiveSocket requestHandler;
     private Subscription receiversSubscription;
-
     public ServerReactiveSocket(DuplexConnection connection, ReactiveSocket requestHandler,
-                                Consumer<Throwable> errorConsumer) {
+                                boolean clientHonorsLease, Consumer<Throwable> errorConsumer) {
         this.requestHandler = requestHandler;
         this.connection = connection;
         serverInput = connection.receive();
@@ -64,14 +63,22 @@ public ServerReactiveSocket(DuplexConnection connection, ReactiveSocket requestH
         if (requestHandler instanceof LeaseEnforcingSocket) {
             LeaseEnforcingSocket enforcer = (LeaseEnforcingSocket) requestHandler;
             enforcer.acceptLeaseSender(lease -> {
+                if (!clientHonorsLease) {
+                    return;
+                }
                 Frame leaseFrame = Lease.from(lease.getTtl(), lease.getAllowedRequests(), lease.metadata());
                 Px.from(connection.sendOne(leaseFrame))
-                    .doOnError(errorConsumer)
-                    .subscribe();
+                  .doOnError(errorConsumer)
+                  .subscribe();
             });
         }
     }
 
+    public ServerReactiveSocket(DuplexConnection connection, ReactiveSocket requestHandler,
+                                Consumer<Throwable> errorConsumer) {
+        this(connection, requestHandler, true, errorConsumer);
+    }
+
     @Override
     public Publisher<Void> fireAndForget(Payload payload) {
         return requestHandler.fireAndForget(payload);

reactivesocket-core/src/main/java/io/reactivesocket/client/SetupProvider.java

@@ -22,6 +22,7 @@
 import io.reactivesocket.DuplexConnection;
 import io.reactivesocket.Frame;
 import io.reactivesocket.Frame.Setup;
+import io.reactivesocket.lease.DisableLeaseSocket;
 import io.reactivesocket.lease.LeaseHonoringSocket;
 import io.reactivesocket.ReactiveSocket;
 import io.reactivesocket.frame.SetupFrameFlyweight;
@@ -84,6 +85,15 @@ public interface SetupProvider {
      */
     SetupProvider disableLease();
 
+    /**
+     * Creates a new {@code SetupProvider} that does not honor leases.
+     *
+     * @param socketFactory A factory to create {@link DisableLeaseSocket} for each accepted socket.
+     *
+     * @return A new {@code SetupProvider} instance.
+     */
+    SetupProvider disableLease(Function<ReactiveSocket, DisableLeaseSocket> socketFactory);
+
     /**
      * Creates a new {@code SetupProvider} that uses the passed {@code setupPayload} as the payload for the setup frame.
      * Default instances, do not have any payload.

reactivesocket-core/src/main/java/io/reactivesocket/client/SetupProviderImpl.java

@@ -42,11 +42,11 @@
 final class SetupProviderImpl implements SetupProvider {
 
     private final Frame setupFrame;
-    private final Function<ReactiveSocket, LeaseHonoringSocket> leaseDecorator;
+    private final Function<ReactiveSocket, ? extends LeaseHonoringSocket> leaseDecorator;
     private final Consumer<Throwable> errorConsumer;
     private final KeepAliveProvider keepAliveProvider;
 
-    SetupProviderImpl(Frame setupFrame, Function<ReactiveSocket, LeaseHonoringSocket> leaseDecorator,
+    SetupProviderImpl(Frame setupFrame, Function<ReactiveSocket, ? extends LeaseHonoringSocket> leaseDecorator,
                       KeepAliveProvider keepAliveProvider, Consumer<Throwable> errorConsumer) {
         this.keepAliveProvider = keepAliveProvider;
         this.errorConsumer = errorConsumer;
@@ -98,12 +98,16 @@ public SetupProvider honorLease(Function<ReactiveSocket, LeaseHonoringSocket> le
 
     @Override
     public SetupProvider disableLease() {
+        return disableLease(DisableLeaseSocket::new);
+    }
+
+    @Override
+    public SetupProvider disableLease(Function<ReactiveSocket, DisableLeaseSocket> socketFactory) {
         Frame newSetup = from(getFlags(setupFrame) & ~ConnectionSetupPayload.HONOR_LEASE,
                               keepaliveInterval(setupFrame), maxLifetime(setupFrame),
                               Frame.Setup.metadataMimeType(setupFrame), Frame.Setup.dataMimeType(setupFrame),
                               setupFrame);
-        return new SetupProviderImpl(newSetup, reactiveSocket -> new DisableLeaseSocket(reactiveSocket),
-                                     keepAliveProvider, errorConsumer);
+        return new SetupProviderImpl(newSetup, socketFactory, keepAliveProvider, errorConsumer);
     }
 
     @Override

reactivesocket-core/src/main/java/io/reactivesocket/lease/DisableLeaseSocket.java

@@ -19,24 +19,25 @@
 import io.reactivesocket.Payload;
 import io.reactivesocket.ReactiveSocket;
 import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * {@link LeaseHonoringSocket} that does not expect to receive any leases and {@link #accept(Lease)} throws an error.
  */
 public class DisableLeaseSocket implements LeaseHonoringSocket {
 
+    private static final Logger logger = LoggerFactory.getLogger(DisableLeaseSocket.class);
+
     private final ReactiveSocket delegate;
 
     public DisableLeaseSocket(ReactiveSocket delegate) {
         this.delegate = delegate;
     }
 
-    /**
-     * @throws IllegalArgumentException Always thrown.
-     */
     @Override
     public void accept(Lease lease) {
-        throw new IllegalArgumentException("Leases are disabled.");
+        logger.info("Leases are disabled but received a lease from the peer. " + lease);
     }
 
     @Override

reactivesocket-core/src/main/java/io/reactivesocket/lease/LeaseImpl.java

@@ -57,4 +57,13 @@ public long expiry() {
     public ByteBuffer metadata() {
         return metadata;
     }
+
+    @Override
+    public String toString() {
+        return "LeaseImpl{" +
+               "allowedRequests=" + allowedRequests +
+               ", ttl=" + ttl +
+               ", expiry=" + expiry +
+               '}';
+    }
 }

reactivesocket-core/src/main/java/io/reactivesocket/server/ReactiveSocketServer.java

@@ -53,6 +53,7 @@ static ReactiveSocketServer create(TransportServer transportServer) {
                                                                                  KeepAliveProvider.never());
                           LeaseEnforcingSocket handler = acceptor.accept(setupPayload, sender);
                           ServerReactiveSocket receiver = new ServerReactiveSocket(duplexConnection, handler,
+                                                                                   setupPayload.willClientHonorLease(),
                                                                                    Throwable::printStackTrace);
                           receiver.start();
                           return duplexConnection.onClose();

reactivesocket-transport-local/src/main/java/io/reactivesocket/local/internal/PeerConnector.java

@@ -65,13 +65,13 @@ public static PeerConnector connect(String name, int id) {
         return new PeerConnector(uniqueName);
     }
 
-    private final class LocalDuplexConnection implements DuplexConnection, Consumer<Frame> {
+    private final class LocalDuplexConnection implements DuplexConnection {
 
         private volatile ValidatingSubscription<Frame> receiver;
         private volatile boolean connected;
         private final EmptySubject closeNotifier;
         private final boolean client;
-        private volatile Consumer<Frame> peer;
+        private volatile LocalDuplexConnection peer;
 
         private LocalDuplexConnection(EmptySubject closeNotifier, boolean client) {
             this.closeNotifier = closeNotifier;
@@ -91,7 +91,7 @@ public Publisher<Void> send(Publisher<Frame> frames) {
                     subscription.request(Long.MAX_VALUE); // Local transport is not flow controlled.
                 }, frame -> {
                     if (peer != null) {
-                        peer.accept(frame);
+                        peer.receiveFrameFromPeer(frame);
                     } else {
                         logger.warn("Sending a frame but peer not connected. Ignoring frame: " + frame);
                     }
@@ -145,8 +145,7 @@ public String toString() {
             return "[local connection(" + (client ? "client" : "server" + ") - ") + name + "] connected: " + connected;
         }
 
-        @Override
-        public void accept(Frame frame) {
+        public void receiveFrameFromPeer(Frame frame) {
             if (receiver != null) {
                 receiver.safeOnNext(frame);
             } else {

reactivesocket-transport-local/src/test/java/io/reactivesocket/ClientDishonorLeaseTest.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2016 Netflix, Inc.
+ * <p>
+ *  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ *  the License. You may obtain a copy of the License at
+ *  <p>
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  <p>
+ *  Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ *  specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivesocket;
+
+import io.reactivesocket.client.ReactiveSocketClient;
+import io.reactivesocket.lease.DefaultLeaseEnforcingSocket;
+import io.reactivesocket.lease.DefaultLeaseEnforcingSocket.LeaseDistributor;
+import io.reactivesocket.lease.DisableLeaseSocket;
+import io.reactivesocket.lease.Lease;
+import io.reactivesocket.lease.LeaseImpl;
+import io.reactivesocket.local.LocalSendReceiveTest.LocalRule;
+import io.reactivesocket.reactivestreams.extensions.internal.CancellableImpl;
+import io.reactivesocket.server.ReactiveSocketServer;
+import io.reactivesocket.util.PayloadImpl;
+import io.reactivex.Single;
+import io.reactivex.subscribers.TestSubscriber;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+
+import static io.reactivesocket.client.KeepAliveProvider.*;
+import static io.reactivesocket.client.SetupProvider.*;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+import static org.mockito.Matchers.*;
+
+public class ClientDishonorLeaseTest {
+
+    @Rule
+    public final LocalRSRule rule = new LocalRSRule();
+
+    @Test(timeout = 10000)
+    public void testNoLeasesSentToClient() throws Exception {
+        ReactiveSocket socket = rule.connectSocket();
+        rule.sendLease();
+
+        TestSubscriber<Payload> s = TestSubscriber.create();
+        socket.requestResponse(PayloadImpl.EMPTY).subscribe(s);
+        s.awaitTerminalEvent();
+
+        assertThat("Unexpected leases received by the client.", rule.leases, is(empty()));
+    }
+
+    public static class LocalRSRule extends LocalRule {
+
+        private ReactiveSocketServer socketServer;
+        private ReactiveSocketClient socketClient;
+        private LeaseDistributor leaseDistributorMock;
+        private List<Lease> leases;
+
+        @Override
+        public Statement apply(final Statement base, Description description) {
+            return new Statement() {
+                @Override
+                public void evaluate() throws Throwable {
+                    leases = new CopyOnWriteArrayList<>();
+                    leaseDistributorMock = Mockito.mock(LeaseDistributor.class);
+                    Mockito.when(leaseDistributorMock.registerSocket(any())).thenReturn(new CancellableImpl());
+                    init();
+                    socketServer = ReactiveSocketServer.create(localServer);
+                    socketServer.start((setup, sendingSocket) -> {
+                        return new DefaultLeaseEnforcingSocket(new AbstractReactiveSocket() { }, leaseDistributorMock);
+                    });
+                    socketClient = ReactiveSocketClient.create(localClient, keepAlive(never())
+                            .disableLease(reactiveSocket -> new DisableLeaseSocket(reactiveSocket) {
+                                @Override
+                                public void accept(Lease lease) {
+                                    leases.add(lease);
+                                }
+                            }));
+                    base.evaluate();
+                }
+            };
+        }
+
+        public ReactiveSocket connectSocket() {
+            return Single.fromPublisher(socketClient.connect()).blockingGet();
+        }
+
+        @SuppressWarnings({"rawtypes", "unchecked"})
+        public Consumer<Lease> sendLease() {
+            ArgumentCaptor<Consumer> leaseConsumerCaptor = ArgumentCaptor.forClass(Consumer.class);
+            Mockito.verify(leaseDistributorMock).registerSocket(leaseConsumerCaptor.capture());
+            Consumer<Lease> leaseConsumer = leaseConsumerCaptor.getValue();
+            leaseConsumer.accept(new LeaseImpl(1, 1, Frame.NULL_BYTEBUFFER));
+            return leaseConsumer;
+        }
+    }
+}

reactivesocket-transport-local/src/test/java/io/reactivesocket/local/LocalSendReceiveTest.java

@@ -67,26 +67,30 @@ public void testClose() throws Exception {
 
     public static class LocalRule extends ExternalResource {
 
-        private LocalClient localClient;
-        private String name;
+        protected String name;
+        protected LocalServer localServer;
+        protected LocalClient localClient;
 
         @Override
         public Statement apply(final Statement base, Description description) {
             return new Statement() {
                 @Override
                 public void evaluate() throws Throwable {
-                    name = "test-send-receive-server-" + ThreadLocalRandom.current().nextInt();
-                    LocalServer.create(name)
-                               .start(duplexConnection -> {
-                                   return duplexConnection.send(duplexConnection.receive());
-                               });
-
-                    localClient = LocalClient.create(name);
+                    init();
+                    localServer.start(duplexConnection -> {
+                        return duplexConnection.send(duplexConnection.receive());
+                    });
                     base.evaluate();
                 }
             };
         }
 
+        protected void init() {
+            name = "test-send-receive-server-" + ThreadLocalRandom.current().nextInt();
+            localServer = LocalServer.create(name);
+            localClient = LocalClient.create(name);
+        }
+
         public DuplexConnection connect() {
             return Single.fromPublisher(localClient.connect()).blockingGet();
         }