Fix JUnit Race Conditions

This fix now implements the observer pattern for the
PolicyEngineJUnitImpl so that the test cases can be
notified when an event is available to be pulled.
This eliminates the need to sleep the thread for
some arbitrary time to let the rules process events.

Issue-ID: POLICY-291
Change-Id: I884c959c20b95a1b58f357602099d8a01c8d2e3c
Signed-off-by: Daniel Cruz <dc443y@att.com>

2 files changed, 66 insertions, 3 deletions

diff --git a/controlloop/common/eventmanager/src/main/java/org/onap/policy/drools/PolicyEngineListener.java b/controlloop/common/eventmanager/src/main/java/org/onap/policy/drools/PolicyEngineListener.java
new file mode 100644
index 000000000..216f819ca
--- /dev/null
+++ b/controlloop/common/eventmanager/src/main/java/org/onap/policy/drools/PolicyEngineListener.java
@@ -0,0 +1,35 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * policy engine
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.drools;
+
+public interface PolicyEngineListener {
+    
+    /**
+     * Any class that implements this interface will
+     * be notified of a new event on the queue in the
+     * PolicyEngineJUnitImpl 
+     * 
+     * @param topic a key to the queue that contains
+     *        the event
+     */
+    public void newEventNotification(String topic);
+
+}
diff --git a/controlloop/common/eventmanager/src/main/java/org/onap/policy/drools/impl/PolicyEngineJUnitImpl.java b/controlloop/common/eventmanager/src/main/java/org/onap/policy/drools/impl/PolicyEngineJUnitImpl.java
index aed72815c..a20731aaf 100644
--- a/controlloop/common/eventmanager/src/main/java/org/onap/policy/drools/impl/PolicyEngineJUnitImpl.java
+++ b/controlloop/common/eventmanager/src/main/java/org/onap/policy/drools/impl/PolicyEngineJUnitImpl.java
@@ -20,8 +20,10 @@
 
 package org.onap.policy.drools.impl;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.Queue;
 
@@ -33,12 +35,36 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.onap.policy.drools.PolicyEngine;
+import org.onap.policy.drools.PolicyEngineListener;
 
 public class PolicyEngineJUnitImpl implements PolicyEngine {
 
 	private static final Logger logger = LoggerFactory.getLogger(PolicyEngineJUnitImpl.class);
 	private Map<String, Map<String, Queue<Object>>> busMap = new HashMap<String, Map<String, Queue<Object>>>();
-
+	private List<PolicyEngineListener> listeners = new ArrayList<>();
+	
+	/**
+	 * Adds all objects that implement PolicyEngineListener
+     * to the notification list when an event occurs
+     * 
+	 * @param listener an object that is interest in knowing
+	 * about events published to the PolicyEngine
+	 */
+	public void addListener(PolicyEngineListener listener) {
+	    listeners.add(listener);
+	}
+	
+	/**
+	 * Notifies all listeners about a new event
+	 * @param topic the topic in which the notification
+	 * was sent to
+	 */
+	public void notifyListeners(String topic) {
+	    for (PolicyEngineListener listener: listeners) {
+	        listener.newEventNotification(topic);
+	    }
+	}
+	
 	@Override
 	public boolean deliver(String busType, String topic, Object obj) {
 		if (obj instanceof ControlLoopNotification) {
@@ -48,7 +74,7 @@ public class PolicyEngineJUnitImpl implements PolicyEngine {
 		}
 		if (obj instanceof Request) {
 			Request request = (Request) obj;
-			logger.debug("Request: {} subrequst {}", request.Action, request.CommonHeader.SubRequestID);
+			logger.debug("Request: {} subrequest {}", request.Action, request.CommonHeader.SubRequestID);
 		}
 		else if (obj instanceof LCMRequestWrapper) {
 		    LCMRequestWrapper dmaapRequest = (LCMRequestWrapper) obj;
@@ -82,7 +108,9 @@ public class PolicyEngineJUnitImpl implements PolicyEngine {
 		// Get the topic queue
 		//
 		logger.debug("queueing");
-		return topicMap.get(topic).add(obj);
+		boolean res = topicMap.get(topic).add(obj);
+		notifyListeners(topic);
+		return res;
 	}
 	
 	public Object	subscribe(String busType, String topic) {