Draining from queue instead of clearing it (#262)

now instead of first passing all the spans from the queue to the list and then clearing it we're now draining it contents. If in the meantime any spans will arrive they will be rained at next passing

fixes #259

spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/StreamSpanReporter.java

@@ -17,10 +17,10 @@
 package org.springframework.cloud.sleuth.stream;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 import org.springframework.cloud.sleuth.Span;
 import org.springframework.cloud.sleuth.SpanReporter;
@@ -38,7 +38,7 @@ import org.springframework.integration.annotation.MessageEndpoint;
 @MessageEndpoint
 public class StreamSpanReporter implements SpanReporter {
 
-	private Collection<Span> queue = new ConcurrentLinkedQueue<>();
+	private BlockingQueue<Span> queue = new LinkedBlockingQueue<>();
 	private final HostLocator endpointLocator;
 	private final SpanMetricReporter spanMetricReporter;
 
@@ -47,14 +47,14 @@ public class StreamSpanReporter implements SpanReporter {
 		this.spanMetricReporter = spanMetricReporter;
 	}
 
-	public void setQueue(Collection<Span> queue) {
+	public void setQueue(BlockingQueue<Span> queue) {
 		this.queue = queue;
 	}
 
 	@InboundChannelAdapter(value = SleuthSource.OUTPUT)
 	public Spans poll() {
-		List<Span> result = new ArrayList<>(this.queue);
-		this.queue.clear();
+		List<Span> result = new ArrayList<>();
+		this.queue.drainTo(result);
 		for (Iterator<Span> iterator = result.iterator(); iterator.hasNext();) {
 			Span span = iterator.next();
 			if (span.getName() != null && span.getName().equals("message/" + SleuthSource.OUTPUT)) {

spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/StreamSpanListenerTests.java

@@ -24,6 +24,9 @@ import static org.mockito.Mockito.verify;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.LinkedBlockingQueue;
 
 import javax.annotation.PostConstruct;
 
@@ -155,7 +158,7 @@ public class StreamSpanListenerTests {
 	@MessageEndpoint
 	protected static class ZipkinTestConfiguration {
 
-		private List<Span> spans = new ArrayList<>();
+		private BlockingQueue<Span> spans = new LinkedBlockingQueue<>();
 
 		@Autowired
 		StreamSpanReporter listener;