Fix bug in FairLeaseGovernor (#147)

***Problem***
There is a bug in FairLeaseGovernor, the number of ticket per Responder is
never decreased.

***Solution***
Fix the bug, add two unit-tests.

Author: stevegury

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

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.lease;
 
 import io.reactivesocket.Frame;
@@ -15,11 +30,11 @@
  * Distribute evenly a static number of tickets to all connected clients.
  */
 public class FairLeaseGovernor implements LeaseGovernor {
-    private static ScheduledExecutorService EXECUTOR = Executors.newScheduledThreadPool(1);
-
     private final int tickets;
     private final long period;
     private final TimeUnit unit;
+    private final ScheduledExecutorService executor;
+
     private final Map<Responder, Integer> responders;
     private ScheduledFuture<?> runningTask;
 
@@ -41,19 +56,29 @@ private synchronized void distribute(int ttlMs) {
         }
     }
 
-    public FairLeaseGovernor(int tickets, long period, TimeUnit unit) {
+    public FairLeaseGovernor(int tickets, long period, TimeUnit unit, ScheduledExecutorService executor) {
         this.tickets = tickets;
         this.period = period;
         this.unit = unit;
+        this.executor = executor;
         responders = new HashMap<>();
     }
 
+    public FairLeaseGovernor(int tickets, long period, TimeUnit unit) {
+        this(tickets, period, unit, Executors.newScheduledThreadPool(2, runnable -> {
+            Thread thread = new Thread(runnable);
+            thread.setDaemon(true);
+            thread.setName("FairLeaseGovernor");
+            return thread;
+        }));
+    }
+
     @Override
     public synchronized void register(Responder responder) {
         responders.put(responder, 0);
         if (runningTask == null) {
             final int ttl = (int)TimeUnit.NANOSECONDS.convert(period, unit);
-            runningTask = EXECUTOR.scheduleAtFixedRate(() -> distribute(ttl), 0, period, unit);
+            runningTask = executor.scheduleAtFixedRate(() -> distribute(ttl), 0, period, unit);
         }
     }
 
@@ -68,8 +93,13 @@ public synchronized void unregister(Responder responder) {
 
     @Override
     public synchronized boolean accept(Responder responder, Frame frame) {
-        boolean valid;
-        final Integer remainingTickets = responders.get(responder);
-        return remainingTickets == null || remainingTickets > 0;
+        Integer remainingTickets = responders.get(responder);
+        if (remainingTickets != null) {
+            remainingTickets--;
+        } else {
+            remainingTickets = -1;
+        }
+        responders.put(responder, remainingTickets);
+        return remainingTickets >= 0;
     }
 }

reactivesocket-examples/src/main/java/io/reactivesocket/examples/StressTest.java

@@ -21,6 +21,7 @@
 import io.reactivesocket.ReactiveSocket;
 import io.reactivesocket.RequestHandler;
 import io.reactivesocket.client.ClientBuilder;
+import io.reactivesocket.lease.FairLeaseGovernor;
 import io.reactivesocket.transport.tcp.client.TcpReactiveSocketConnector;
 import io.reactivesocket.util.Unsafe;
 import io.reactivesocket.test.TestUtil;
@@ -80,13 +81,14 @@ public void cancel() {}
                 .build();
 
         SocketAddress addr = new InetSocketAddress("127.0.0.1", 0);
-        TcpReactiveSocketServer.StartedServer server = TcpReactiveSocketServer.create(addr).start(setupHandler);
-        SocketAddress serverAddress = server.getServerAddress();
-        return serverAddress;
+        FairLeaseGovernor leaseGovernor = new FairLeaseGovernor(5000, 100, TimeUnit.MILLISECONDS);
+        TcpReactiveSocketServer.StartedServer server =
+            TcpReactiveSocketServer.create(addr).start(setupHandler, leaseGovernor);
+        return server.getServerAddress();
     }
 
     private static Publisher<Collection<SocketAddress>> getServersList() {
-        Observable<Collection<SocketAddress>> serverAddresses = Observable.interval(2, TimeUnit.SECONDS)
+        Observable<Collection<SocketAddress>> serverAddresses = Observable.interval(0, 2, TimeUnit.SECONDS)
             .map(new Func1<Long, Collection<SocketAddress>>() {
                 List<SocketAddress> addresses = new ArrayList<>();