Enhance ASDC listener bundler start and stop

Store ASDC listner start up initialization thread, to provide reference
for ASDC listener stop to terminate the thread if it is still running at
stopping time.
Add thread interrupt checking in initialization thread, to allow earlier
termination of the thread during stopping process.

Issue-Id: APPC-166
Change-Id: Ibc13afce251ce9344a4ad807d3927fc277bcbe6c
Signed-off-by: beili.zhou <beili.zhou@amdocs.com>

2 files changed, 300 insertions, 208 deletions

diff --git a/appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcConfig.java b/appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcConfig.java
index 448a7f365..d0d5319f9 100644
--- a/appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcConfig.java
+++ b/appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcConfig.java
@@ -29,161 +29,169 @@ import com.att.eelf.configuration.EELFManager;
 import org.openecomp.sdc.api.consumer.IConfiguration;
 
 import java.net.URI;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 
 public class AsdcConfig implements IConfiguration {
 
-	private String host;
-	private String consumer;
-	private String consumerId;
-	private String env;
-	private String keystorePath;
-	private String keystorePass;
-	private int pollingInterval; // Time between listening sessions
-	private int pollingTimeout; // Time to listen for (dmaap timeout url param)/1000
-	private List<String> types = new ArrayList<>();
-	private String user;
-	private String pass;
-
-	private URI storeOp;
-
-	Properties props;
-
-	private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcConfig.class);
-
-	public AsdcConfig(Properties props) throws Exception {
-		this.props = props;
-		init();
-	}
-
-	private void init() throws Exception {
-		if (props != null) {
-			// Keystore for ca cert
-			keystorePath = props.getProperty("appc.asdc.keystore.path");
-			keystorePass = props.getProperty("appc.asdc.keystore.pass");
-
-			// ASDC host
-			host = props.getProperty("appc.asdc.host");
-			env = props.getProperty("appc.asdc.env");
-			user = props.getProperty("appc.asdc.user");
-			pass = props.getProperty("appc.asdc.pass");
-
-			// DMaaP properties
-			consumer = props.getProperty("appc.asdc.consumer");
-			consumerId = props.getProperty("appc.asdc.consumer.id");
-
-			pollingInterval = Integer.valueOf(props.getProperty("interval", "60"));
-
-			// Client uses cambriaClient-0.2.4 which throws non relevant (wrong)
-			// exceptions with times > 30s
-			pollingTimeout = Integer.valueOf(props.getProperty("timeout", "25"));
-
-			// Anything less than 60 and we risk 429 Too Many Requests
-			if (pollingInterval < 60) {
-				pollingInterval = 60;
-			}
-
-			if (pollingInterval > pollingInterval) {
-				logger.warn(String.format(
-						"Message acknowledgement may be delayed by %ds in the ADSC listener. [Listening Time: %s, Poll Period: %s]",
-						pollingInterval - pollingTimeout, pollingTimeout, pollingInterval));
-			}
-
-			logParams();
-
-			// Download type
-			types.add("APPC_CONFIG");
-			types.add("VF_LICENSE");
-			types.add("TOSCA_CSAR");
-			/*
-			This types seems redundant, as it looks from the code that they are not being used anywhere
-			 */
-
-			storeOp = new URI(props.getProperty("appc.asdc.provider.url"));
-		}
-	}
-
-	@Override
-	public boolean activateServerTLSAuth() {
-		return false;
-	}
-
-	//@Override
-	public boolean isFilterInEmptyResources() {
-		return false;
-	}
-
-	@Override
-	public String getAsdcAddress() {
-		return host;
-	}
-
-	@Override
-	public String getConsumerGroup() {
-		return consumer;
-	}
-
-	@Override
-	public String getConsumerID() {
-		return consumerId;
-	}
-
-	@Override
-	public String getEnvironmentName() {
-		return env;
-	}
-
-	@Override
-	public String getKeyStorePassword() {
-		return keystorePass;
-	}
-
-	@Override
-	public String getKeyStorePath() {
-		return keystorePath;
-	}
-
-	@Override
-	public String getPassword() {
-		return pass;
-	}
-
-	@Override
-	public int getPollingInterval() {
-		return pollingInterval;
-	}
-
-	@Override
-	public int getPollingTimeout() {
-		return pollingTimeout;
-	}
-
-	@Override
-	public List<String> getRelevantArtifactTypes() {
-		return types;
-	}
-
-	@Override
-	public String getUser() {
-		return user;
-	}
-
-	public URI getStoreOpURI() {
-		return storeOp;
-	}
-
-	/**
-	 * Logs the relevant parameters
-	 */
-	public void logParams() {
-		Map<String, String> params = new HashMap<String, String>();
-		params.put("ASDC Host", getAsdcAddress());
-		params.put("ASDC Environment", getEnvironmentName());
-		params.put("Consumer Name", getConsumerGroup());
-		params.put("Consumer ID", getConsumerID());
-		params.put("Poll Active Wait", String.valueOf(getPollingInterval()));
-		params.put("Poll Timeout", String.valueOf(getPollingTimeout()));
-
-		logger.info(String.format("ASDC Params: %s", params));
-	}
+    private String host;
+    private String consumer;
+    private String consumerId;
+    private String env;
+    private String keystorePath;
+    private String keystorePass;
+    /** Polling internal is time between listening sessions */
+    private int pollingInterval;
+    /** Polling timeout is the time to listen for (dmaap timeout url param)/1000 */
+    private int pollingTimeout;
+    private List<String> types = new ArrayList<>();
+    private String user;
+    private String pass;
+
+    private URI storeOp;
+
+    private Properties props;
+
+    private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcConfig.class);
+
+    AsdcConfig(Properties props) throws Exception {
+        this.props = props;
+        init();
+    }
+
+    private void init() throws Exception {
+        if (props == null) {
+            logger.error("SdcConfig init is skipped due to properties is null");
+            return;
+        }
+
+        // Keystore for ca cert
+        keystorePath = props.getProperty("appc.asdc.keystore.path");
+        keystorePass = props.getProperty("appc.asdc.keystore.pass");
+
+        // ASDC host
+        host = props.getProperty("appc.asdc.host");
+        env = props.getProperty("appc.asdc.env");
+        user = props.getProperty("appc.asdc.user");
+        pass = props.getProperty("appc.asdc.pass");
+
+        // DMaaP properties
+        consumer = props.getProperty("appc.asdc.consumer");
+        consumerId = props.getProperty("appc.asdc.consumer.id");
+
+        pollingInterval = Integer.valueOf(props.getProperty("interval", "60"));
+
+        // Client uses cambriaClient-0.2.4 which throws non relevant (wrong)
+        // exceptions with times > 30s
+        pollingTimeout = Integer.valueOf(props.getProperty("timeout", "25"));
+
+        // Anything less than 60 and we risk 429 Too Many Requests
+        if (pollingInterval < 60) {
+            pollingInterval = 60;
+        }
+
+        if (pollingInterval > pollingTimeout) {
+            logger.warn(String.format(
+                    "Message acknowledgement may be delayed by %ds in the ADSC listener. [Listening Time: %s, Poll Period: %s]",
+                    pollingInterval - pollingTimeout, pollingTimeout, pollingInterval));
+        }
+
+        logParams();
+
+        // Download type
+        /*
+          This types seems redundant, as it looks from the code that they are not being used anywhere
+        */
+        types.add("APPC_CONFIG");
+        types.add("VF_LICENSE");
+        types.add("TOSCA_CSAR");
+
+        storeOp = new URI(props.getProperty("appc.asdc.provider.url"));
+    }
+
+    @Override
+    public boolean activateServerTLSAuth() {
+        return false;
+    }
+
+    public boolean isFilterInEmptyResources() {
+        return false;
+    }
+
+    @Override
+    public String getAsdcAddress() {
+        return host;
+    }
+
+    @Override
+    public String getConsumerGroup() {
+        return consumer;
+    }
+
+    @Override
+    public String getConsumerID() {
+        return consumerId;
+    }
+
+    @Override
+    public String getEnvironmentName() {
+        return env;
+    }
+
+    @Override
+    public String getKeyStorePassword() {
+        return keystorePass;
+    }
+
+    @Override
+    public String getKeyStorePath() {
+        return keystorePath;
+    }
+
+    @Override
+    public String getPassword() {
+        return pass;
+    }
+
+    @Override
+    public int getPollingInterval() {
+        return pollingInterval;
+    }
+
+    @Override
+    public int getPollingTimeout() {
+        return pollingTimeout;
+    }
+
+    @Override
+    public List<String> getRelevantArtifactTypes() {
+        return types;
+    }
+
+    @Override
+    public String getUser() {
+        return user;
+    }
+
+    URI getStoreOpURI() {
+        return storeOp;
+    }
+
+    /**
+     * Logs the relevant parameters
+     */
+    private void logParams() {
+        Map<String, String> params = new HashMap<>();
+        params.put("ASDC Host", getAsdcAddress());
+        params.put("ASDC Environment", getEnvironmentName());
+        params.put("Consumer Name", getConsumerGroup());
+        params.put("Consumer ID", getConsumerID());
+        params.put("Poll Active Wait", String.valueOf(getPollingInterval()));
+        params.put("Poll Timeout", String.valueOf(getPollingTimeout()));
+
+        logger.info(String.format("ASDC Params: %s", params));
+    }
 }
diff --git a/appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcListener.java b/appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcListener.java
index d9755ef46..a580e4077 100644
--- a/appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcListener.java
+++ b/appc-asdc-listener/appc-asdc-listener-bundle/src/main/java/org/openecomp/appc/sdc/listener/AsdcListener.java
@@ -23,6 +23,13 @@
  */
 
 package org.openecomp.appc.sdc.listener;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import org.openecomp.appc.configuration.Configuration;
+import org.openecomp.appc.configuration.ConfigurationFactory;
+import org.openecomp.sdc.api.IDistributionClient;
+import org.openecomp.sdc.api.results.IDistributionClientResult;
 import org.openecomp.sdc.impl.DistributionClientFactory;
 import org.openecomp.sdc.utils.DistributionActionResultEnum;
 
@@ -33,16 +40,13 @@ import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.openecomp.appc.configuration.Configuration;
-import org.openecomp.appc.configuration.ConfigurationFactory;
-import org.openecomp.sdc.api.IDistributionClient;
-import org.openecomp.sdc.api.results.IDistributionClientResult;
-import org.openecomp.sdc.impl.DistributionClientFactory;
-import org.openecomp.sdc.utils.DistributionActionResultEnum;
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-
+/**
+ * SDC listener handles bundle start and stop through start and stop method. <p>
+ * Register connection with SDC server based on properties file configuration when start,
+ * and disconnect with SDC server when stop.
+ */
 public class AsdcListener {
+    private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcListener.class);
 
     /**
      * The bundle context
@@ -52,83 +56,163 @@ public class AsdcListener {
     private AsdcConfig config;
     private CountDownLatch latch;
 
-
-    private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcListener.class);
+    private Thread startThread = null;
 
     @SuppressWarnings("unused")
     public void start() throws Exception {
-        logger.info("Starting bundle ASDC Listener");
+        // Add timestamp to the log to differentiate the jmeter run testing calls.
+        final long timeStamp = System.currentTimeMillis();
+        logger.info(String.format("[%d] Starting SDC Listener", timeStamp));
+
         Configuration configuration = ConfigurationFactory.getConfiguration();
         Properties props = configuration.getProperties();
-
         config = new AsdcConfig(props);
+        logger.debug(String.format("[%d] created SDC config", timeStamp));
 
         client = DistributionClientFactory.createDistributionClient();
+        logger.debug(String.format("[%d] created SDC client", timeStamp));
+
         callback = new AsdcCallback(config.getStoreOpURI(), client);
+        logger.debug(String.format("[%d] created SDC callback", timeStamp));
 
         latch = new CountDownLatch(1);
 
-        new Thread(new Runnable() {
-            @Override
-            public void run() {
-                initialRegistration(config);
-
-                IDistributionClientResult result = client.init(config, callback);
-
-                if (result.getDistributionActionResult() == DistributionActionResultEnum.SUCCESS) {
-                    client.start();
-                } else {
-                    logger.error(String.format("Could not register ASDC client. %s - %s", result.getDistributionActionResult(),
-                                    result.getDistributionMessageResult()));
-                }
-
-                latch.countDown();
-            }
-        }).start();
+        startThread = new Thread(new StartRunnable(timeStamp));
+        startThread.setName(String.format("[%d] sdcListener start", timeStamp));
+        logger.debug(String.format("[%d] created SDC initialization thread", timeStamp));
+        startThread.start();
     }
 
     @SuppressWarnings("unused")
     public void stop() throws InterruptedException {
-        logger.info("Stopping ASDC Listener");
-        latch.await(10, TimeUnit.SECONDS);
+        // Add timestamp to the log to differentiate the jmeter run testing calls.
+        final long timeStamp = System.currentTimeMillis();
+        logger.info(String.format("[%d] Stopping ASDC Listener", timeStamp));
+
+        stopStartThread(timeStamp);
+
+        if (latch != null) {
+            logger.debug(String.format("[%d] waiting ASDC latch count to 0 for 10 seconds", timeStamp));
+            latch.await(10, TimeUnit.SECONDS);
+            latch = null;
+        }
 
         if (callback != null) {
+            logger.debug(String.format("[%d] stopping ASDC callback", timeStamp));
             callback.stop();
+            callback = null;
         }
         if (client != null) {
+            logger.debug(String.format("[%d] stopping ASDC client", timeStamp));
             client.stop();
+            client = null;
 
         }
-        logger.info("ASDC Listener stopped successfully");
+        logger.info(String.format("[%d] ASDC Listener stopped successfully", timeStamp));
     }
 
-    private boolean initialRegistration(AsdcConfig config) {
-        try {
-            final String jsonTemplate = "{\"consumerName\": \"%s\",\"consumerSalt\": \"%s\",\"consumerPassword\":\"%s\"}";
-            String saltedPassStr = org.openecomp.tlv.sdc.security.Passwords.hashPassword(config.getPassword());
-            if (saltedPassStr == null || !saltedPassStr.contains(":")) {
-                return false;
+    void stopStartThread(long timeStamp) throws InterruptedException {
+        if (startThread == null) {
+            return;
+        }
+
+        if (startThread.getState() == Thread.State.TERMINATED) {
+            logger.debug(String.format("[%d] ASDC thread(%s) is already terminated.",
+                    timeStamp, startThread.getName()));
+        } else {
+            logger.debug(String.format("[%d] ASDC thread(%s) is to be interrupted with state(%s)",
+                    timeStamp, startThread.getName(), startThread.getState().toString()));
+
+            startThread.interrupt();
+
+            logger.debug(String.format("[%d] ASDC thread(%s) has been interrupted(%s) with state(%s)",
+                    timeStamp, startThread.getName(), startThread.isInterrupted(),
+                    startThread.getState().toString()));
+        }
+        startThread = null;
+    }
+
+    /**
+     * Runnable implementation for actual initialization during ASDC listener start
+     */
+    class StartRunnable implements Runnable {
+        private final long timeStamp;
+
+        StartRunnable(long theTimeStamp) {
+            timeStamp = theTimeStamp;
+        }
+
+        /**
+         * This run method calls ASDC client for init and start which are synchronized calls along with stop.
+         * To interrupt this thread at stop time, we added thread interrupted checking in each step
+         * for earlier interruption.
+         */
+        @Override
+        public void run() {
+            if (!initialRegistration()) {
+                logger.warn(String.format("[%d] ASDC thread initial registration failed.", timeStamp));
+            }
+
+            if (isThreadInterrupted("after initial registration")) {
+                return;
+            }
+
+            IDistributionClientResult result = client.init(config, callback);
+
+            if (isThreadInterrupted("after client init")) {
+                return;
             }
 
-            String[] saltedPass = saltedPassStr.split(":");
-            String json = String.format(jsonTemplate, config.getUser(), saltedPass[0], saltedPass[1]);
+            if (result.getDistributionActionResult() == DistributionActionResultEnum.SUCCESS) {
+                client.start();
+            } else {
+                logger.error(String.format("[%d] Could not register ASDC client. %s - %s",
+                        timeStamp, result.getDistributionActionResult(), result.getDistributionMessageResult()));
+            }
+
+            latch.countDown();
+        }
 
-            Map<String, String> headers = new HashMap<>();
-            // TODO - Replace the header below to sdc's requirements. What should the new value be
-            headers.put("USER_ID", "test");
+        private boolean initialRegistration() {
+            try {
+                final String jsonTemplate =
+                        "{\"consumerName\": \"%s\",\"consumerSalt\": \"%s\",\"consumerPassword\":\"%s\"}";
+                String saltedPassStr = org.openecomp.tlv.sdc.security.Passwords.hashPassword(config.getPassword());
+                if (saltedPassStr == null || !saltedPassStr.contains(":")) {
+                    return false;
+                }
 
-            // TODO - How to format the url. Always same endpoint or ports?
-            String host = config.getAsdcAddress();
-            URL url = new URL(
-                    String.format("http%s://%s/sdc2/rest/v1/consumers", host.contains("443") ? "s" : "", host));
+                String[] saltedPass = saltedPassStr.split(":");
+                String json = String.format(jsonTemplate, config.getUser(), saltedPass[0], saltedPass[1]);
 
-            logger.info(String.format("Attempting to register user %s on %s with salted pass of %s", config.getUser(),
-                    url, saltedPass[1]));
+                Map<String, String> headers = new HashMap<>();
+                // TODO - Replace the header below to sdc's requirements. What should the new value be
+                headers.put("USER_ID", "test");
 
-            ProviderResponse result = ProviderOperations.post(url, json, headers);
-            return result.getStatus() == 200;
-        } catch (Exception e) {
-            logger.error("Error performing initial registration with ASDC server. User may not be able to connect", e);
+                // TODO - How to format the url. Always same endpoint or ports?
+                String host = config.getAsdcAddress();
+                URL url = new URL(String.format("http%s://%s/sdc2/rest/v1/consumers",
+                        host.contains("443") ? "s" : "", host));
+
+                logger.info(String.format("Attempting to register user %s on %s with salted pass of %s",
+                        config.getUser(), url, saltedPass[1]));
+
+                ProviderOperations providerOperations = new ProviderOperations();
+                ProviderResponse result = providerOperations.post(url, json, headers);
+                return result.getStatus() == 200;
+            } catch (Exception e) {
+                logger.error(
+                        "Error performing initial registration with ASDC server. User may not be able to connect",
+                        e);
+                return false;
+            }
+        }
+
+        private boolean isThreadInterrupted(String details) {
+            if (Thread.currentThread().isInterrupted()) {
+                logger.info(String.format("[%d] ASDC thread interrupted %s.", timeStamp, details));
+                return true;
+            }
             return false;
         }
     }