Merge pull request #96 from ankgoyalEXP/dev_Ankit

chaning arrayList to linkedList and releasing the semaphore

Author: mchandramouli

core/src/main/java/com/expedia/www/haystack/client/dispatchers/InMemoryDispatcher.java

@@ -17,9 +17,9 @@
 package com.expedia.www.haystack.client.dispatchers;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.Semaphore;
 
@@ -44,9 +44,9 @@ public class InMemoryDispatcher implements Dispatcher {
 
     public InMemoryDispatcher(Metrics metrics, int limit) {
         limiter = new Semaphore(limit);
-        reported = new ArrayList<>();
-        received = new ArrayList<>();
-        flushed = new ArrayList<>();
+        reported = new LinkedList<>();
+        received = new LinkedList<>();
+        flushed = new LinkedList<>();
 
         // held in the registry a reference; but we don't need a local reference
         Gauge.builder("reported", reported, Collection::size).register(metrics);
@@ -75,6 +75,7 @@ public void dispatch(Span span) {
 
                 reported.add(span);
                 received.add(span);
+                limiter.release();
             }
         }
     }
@@ -84,7 +85,7 @@ public void flush() throws IOException {
         synchronized (this) {
             try (Sample timer = flushTimer.start()) {
                 flushed.addAll(reported);
-                reported.clear();
+                reported = new LinkedList<>();
             }
         }
     }

core/src/main/java/com/expedia/www/haystack/client/dispatchers/RemoteDispatcher.java

@@ -232,5 +232,4 @@ public RemoteDispatcher build() {
             return new RemoteDispatcher(metrics, client, acceptQueue, flushInterval, shutdownTimeout, executor);
         }
     }
-
 }

core/src/main/java/com/expedia/www/haystack/client/dispatchers/clients/InMemoryClient.java

@@ -16,10 +16,10 @@
  */
 package com.expedia.www.haystack.client.dispatchers.clients;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.Semaphore;
 
@@ -50,9 +50,9 @@ public class InMemoryClient implements Client<Span> {
 
     public InMemoryClient(Metrics metrics, int limit) {
         limiter = new Semaphore(limit);
-        total = new ArrayList<>();
-        received = new ArrayList<>();
-        flushed = new ArrayList<>();
+        total = new LinkedList<>();
+        received = new LinkedList<>();
+        flushed = new LinkedList<>();
 
         this.sendTimer = Timer.builder("send").register(metrics);
         this.sendExceptionCounter = Counter.builder("send").tag(new Tag("state", "exception")).register(metrics);
@@ -72,6 +72,7 @@ public boolean send(Span span) {
             limiter.acquire();
             total.add(span);
             received.add(span);
+            limiter.release();
             return true;
         } catch (InterruptedException e) {
             sendExceptionCounter.increment();
@@ -92,7 +93,7 @@ public void flush() {
         try (Sample timer = flushTimer.start()) {
             LOGGER.info("Client flushed");
             flushed.addAll(received);
-            received.clear();
+            received = new LinkedList<>();
         }
     }