added waitForItems methods so to not rely on guessing the right time (#621)

* added waitForItems methods so to not rely on guessing the right time

* intellij messed up my imports...fixed, i guess

* last time i mess with the imports

* srsly, this is the last time

Author: littleaj

test/smoke/framework/fakeIngestion/servlet/src/main/java/com/microsoft/applicationinsights/test/fakeingestion/MockedAppInsightsIngestionServlet.java

@@ -9,6 +9,7 @@
 import com.microsoft.applicationinsights.internal.schemav2.Envelope;
 import com.microsoft.applicationinsights.smoketest.JsonHelper;
 
+import javax.annotation.concurrent.GuardedBy;
 import javax.servlet.ServletConfig;
 import javax.servlet.ServletException;
 import javax.servlet.http.HttpServlet;
@@ -18,9 +19,17 @@
 import java.io.InputStreamReader;
 import java.io.StringWriter;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Queue;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.zip.GZIPInputStream;
 
 public class MockedAppInsightsIngestionServlet extends HttpServlet {
@@ -31,12 +40,18 @@ public class MockedAppInsightsIngestionServlet extends HttpServlet {
 
     private final String appid = "DUMMYAPPID";
 
-    private Queue<Envelope> telemetryReceived;
-    private ListMultimap<String, Envelope> type2envelope;
+
+    private final Queue<Envelope> telemetryReceived;
+    @GuardedBy("multimapLock")
+    private final ListMultimap<String, Envelope> type2envelope;
     private List<Predicate<Envelope>> filters;
 
+    private final Object multimapLock = new Object();
+
     private MockedIngestionServletConfig config;
 
+    private final ExecutorService itemExecutor = Executors.newSingleThreadExecutor();
+
     public static final String LOG_PAYLOADS_PARAMETER_KEY = "logPayloads";
     public static final String RETAIN_PAYLOADS_PARAMETER_KEY = "retainPayloads";
 
@@ -96,7 +111,9 @@ public MockedIngestionServletConfig getIngestionConfig() {
     public void resetData() {
         logit("Clearing telemetry accumulator...");
         telemetryReceived.clear();
-        type2envelope.clear();
+        synchronized (multimapLock) {
+            type2envelope.clear();
+        }
     }
 
     public boolean hasData() {
@@ -113,7 +130,33 @@ public Envelope nextItem() {
 
     public List<Envelope> getItemsByType(String type) {
         Preconditions.checkNotNull(type, "type");
-        return type2envelope.get(type);
+        synchronized (multimapLock) {
+            return type2envelope.get(type);
+        }
+    }
+
+    public List<Envelope> waitForItems(final Predicate<Envelope> condition, final int numItems, int timeout, TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
+        final Future<List<Envelope>> future = itemExecutor.submit(new Callable<List<Envelope>>() {
+            @Override
+            public List<Envelope> call() throws Exception {
+                List<Envelope> targetCollection = new ArrayList<>(numItems);
+                while(targetCollection.size() < numItems) {
+                    targetCollection.clear();
+                    final Collection<Envelope> currentValues;
+                    synchronized (multimapLock) {
+                        currentValues = new ArrayList<>(type2envelope.values());
+                    }
+                    for (Envelope val : currentValues) {
+                        if (condition.apply(val)) {
+                            targetCollection.add(val);
+                        }
+                    }
+                    TimeUnit.MILLISECONDS.sleep(75);
+                }
+                return targetCollection;
+            }
+        });
+        return future.get(timeout, timeUnit);
     }
 
     @Override
@@ -157,8 +200,10 @@ protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws S
                                 String baseType = envelope.getData().getBaseType();
                                 if (filtersAllowItem(envelope)) {
                                     logit("Adding telemetry item: "+baseType);
-                                    type2envelope.put(baseType, envelope);
                                     telemetryReceived.offer(envelope);
+                                    synchronized (multimapLock) {
+                                        type2envelope.put(baseType, envelope);
+                                    }
                                 } else {
                                     logit("Rejected telemetry item by filter: "+baseType);
                                 }

test/smoke/framework/fakeIngestion/standalone/src/main/java/com/microsoft/applicationinsights/test/fakeingestion/MockedAppInsightsIngestionServer.java

@@ -1,18 +1,20 @@
 package com.microsoft.applicationinsights.test.fakeingestion;
 
-import java.util.ArrayList;
-import java.util.List;
-
 import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
 import com.microsoft.applicationinsights.internal.schemav2.Data;
 import com.microsoft.applicationinsights.internal.schemav2.Domain;
 import com.microsoft.applicationinsights.internal.schemav2.Envelope;
-
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.servlet.ServletHandler;
 import org.eclipse.jetty.servlet.ServletHolder;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
 public class MockedAppInsightsIngestionServer implements AutoCloseable {
 	public static final int DEFAULT_PORT = 60606;
 
@@ -89,6 +91,32 @@ public <T extends Domain> T getBaseDataForType(int index, String type) {
 		return data.getBaseData();
 	}
 
+	/**
+	 * Waits the given amount of time for this mocked server to recieve one telemetry item matching the given predicate.
+	 *
+	 * @see #waitForItems(Predicate, int, int, TimeUnit)
+	 */
+	public Envelope waitForItem(Predicate<Envelope> condition, int timeout, TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
+		return waitForItems(condition, 1, timeout, timeUnit).get(0);
+	}
+
+
+	/**
+	 * Waits the given amount of time for this mocked server to receive a certain number of items which match the given predicate.
+	 *
+	 * @param condition condition describing what items to wait for.
+	 * @param numItems number of matching items to wait for.
+	 * @param timeout amount of time to wait
+	 * @param timeUnit the unit of time to wait
+	 * @return The items the given condition. This will be at least {@code numItems}, but could be more.
+	 * @throws InterruptedException if the thread is interrupted while waiting
+	 * @throws ExecutionException if an exception is thrown while waiting
+	 * @throws TimeoutException if the timeout is reached
+	 */
+	public List<Envelope> waitForItems(Predicate<Envelope> condition, int numItems, int timeout, TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
+		return this.servlet.waitForItems(condition, numItems, timeout, timeUnit);
+	}
+
 	@Override
 	public void close() throws Exception {
 		stopServer();