fixing conflicts in StratosThreadPool

apache · Dec 24, 2015 · 412cb2c · 412cb2c
1 parent f302906
commit 412cb2c
Show file tree

Hide file tree

Showing 13 changed files with 208 additions and 149 deletions.
diff --git a/...atos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java b/...atos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgent.java
@@ -90,10 +90,10 @@ public void run() {
  }
 
  // Start instance notifier listener thread
- registerInstanceNotifierEventListeners();
- if (log.isInfoEnabled()) {
- log.info("Cartridge agent registerInstanceNotifierEventListeners done");
- }
+// registerInstanceNotifierEventListeners();
+// if (log.isInfoEnabled()) {
+// log.info("Cartridge agent registerInstanceNotifierEventListeners done");
+// }
 
  // Start tenant event receiver thread
  /*
@@ -174,17 +174,17 @@ public void run() {
  logPublisherManager.stop();
  }
 
- protected void registerInstanceNotifierEventListeners() {
- if (log.isDebugEnabled()) {
- log.debug("SsubscribeToTopicsAndRegisterListeners before");
- }
-
- eventListenerns.startInstanceNotifierReceiver();
-
- if (log.isDebugEnabled()) {
- log.debug("SsubscribeToTopicsAndRegisterListeners after");
- }
- }
+// protected void registerInstanceNotifierEventListeners() {
+// if (log.isDebugEnabled()) {
+// log.debug("SsubscribeToTopicsAndRegisterListeners before");
+// }
+//
+// eventListenerns.startInstanceNotifierReceiver();
+//
+// if (log.isDebugEnabled()) {
+// log.debug("SsubscribeToTopicsAndRegisterListeners after");
+// }
+// }
 
 // protected void registerTopologyEventListeners() {
 // if (log.isDebugEnabled()) {

diff --git a/....agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgentEventListeners.java b/....agent/src/main/java/org/apache/stratos/cartridge/agent/CartridgeAgentEventListeners.java
@@ -65,8 +65,8 @@ public class CartridgeAgentEventListeners {
  private ApplicationSignUpEventReceiver applicationsEventReceiver;
 
  private ExtensionHandler extensionHandler;
- private static final ExecutorService eventListenerExecutorService =
- StratosThreadPool.getExecutorService("cartridge.agent.event.listener.thread.pool", 10);
+// private static final ExecutorService eventListenerExecutorService =
+// StratosThreadPool.getExecutorService("cartridge.agent.event.listener.thread.pool", 10);
 
  public CartridgeAgentEventListeners() {
  if (log.isDebugEnabled()) {
@@ -77,7 +77,7 @@ public CartridgeAgentEventListeners() {
  this.topologyEventReceiver = TopologyEventReceiver.getInstance();
  //this.topologyEventReceiver.setExecutorService(eventListenerExecutorService);
 
- this.instanceNotifierEventReceiver = new InstanceNotifierEventReceiver();
+ this.instanceNotifierEventReceiver = InstanceNotifierEventReceiver.getInstance();
 
  this.tenantEventReceiver = TenantEventReceiver.getInstance();
 // this.tenantEventReceiver.setExecutorService(eventListenerExecutorService);
@@ -113,23 +113,23 @@ public CartridgeAgentEventListeners() {
 //
 // }
 
- public void startInstanceNotifierReceiver() {
-
- if (log.isDebugEnabled()) {
- log.debug("Starting cartridge agent instance notifier event message receiver");
- }
-
- eventListenerExecutorService.submit(new Runnable() {
- @Override
- public void run() {
- instanceNotifierEventReceiver.execute();
- }
- });
-
- if (log.isDebugEnabled()) {
- log.debug("Cartridge agent Instance notifier event message receiver started, waiting for event messages ...");
- }
- }
+// public void startInstanceNotifierReceiver() {
+//
+// if (log.isDebugEnabled()) {
+// log.debug("Starting cartridge agent instance notifier event message receiver");
+// }
+//
+// eventListenerExecutorService.submit(new Runnable() {
+// @Override
+// public void run() {
+// instanceNotifierEventReceiver.execute();
+// }
+// });
+//
+// if (log.isDebugEnabled()) {
+// log.debug("Cartridge agent Instance notifier event message receiver started, waiting for event messages ...");
+// }
+// }
 
 // public void startTenantEventReceiver() {
 //
@@ -521,9 +521,9 @@ protected void onEvent(Event event) {
  * Terminate load balancer topology receiver thread.
  */
 
- public void terminate() {
- topologyEventReceiver.terminate();
- }
+// public void terminate() {
+// topologyEventReceiver.terminate();
+// }
 
 }
 
diff --git a/...e.agent/src/test/java/org/apache/stratos/cartridge/agent/test/JavaCartridgeAgentTest.java b/...e.agent/src/test/java/org/apache/stratos/cartridge/agent/test/JavaCartridgeAgentTest.java
@@ -111,9 +111,9 @@ public void setup() {
  //topologyEventReceiver.setExecutorService(executorService);
  //topologyEventReceiver.execute();
 
- instanceStatusEventReceiver = new InstanceStatusEventReceiver();
- instanceStatusEventReceiver.setExecutorService(executorService);
- instanceStatusEventReceiver.execute();
+ instanceStatusEventReceiver = InstanceStatusEventReceiver.getInstance();
+// instanceStatusEventReceiver.setExecutorService(executorService);
+// instanceStatusEventReceiver.execute();
 
  instanceStarted = false;
  instanceStatusEventReceiver.addEventListener(new InstanceStartedEventListener() {
@@ -176,8 +176,8 @@ public void tearDown() {
  } catch (Exception ignore) {
  }
 
- this.instanceStatusEventReceiver.terminate();
- this.topologyEventReceiver.terminate();
+ //this.instanceStatusEventReceiver.terminate();
+ // this.topologyEventReceiver.terminate();
 
  this.instanceActivated = false;
  this.instanceStarted = false;

diff --git a/...in/java/org/apache/stratos/cloud/controller/internal/CloudControllerServiceComponent.java b/...in/java/org/apache/stratos/cloud/controller/internal/CloudControllerServiceComponent.java
@@ -162,8 +162,8 @@ private void executeCoordinatorTasks() {
  }
 
  instanceStatusTopicReceiver = new InstanceStatusTopicReceiver();
- instanceStatusTopicReceiver.setExecutorService(executorService);
- instanceStatusTopicReceiver.execute();
+// instanceStatusTopicReceiver.setExecutorService(executorService);
+// instanceStatusTopicReceiver.execute();
 
  if (log.isInfoEnabled()) {
  log.info("Instance status event receiver thread started");

diff --git a/...atos/cloud/controller/messaging/receiver/instance/status/InstanceStatusTopicReceiver.java b/...atos/cloud/controller/messaging/receiver/instance/status/InstanceStatusTopicReceiver.java
@@ -45,21 +45,21 @@ public class InstanceStatusTopicReceiver {
  private ExecutorService executorService;
 
  public InstanceStatusTopicReceiver() {
- this.statusEventReceiver = new InstanceStatusEventReceiver();
+ this.statusEventReceiver = InstanceStatusEventReceiver.getInstance();
  addEventListeners();
  }
 
- public void execute() {
- statusEventReceiver.setExecutorService(executorService);
- statusEventReceiver.execute();
- if (log.isInfoEnabled()) {
- log.info("Cloud controller application status thread started");
- }
-
- if (log.isInfoEnabled()) {
- log.info("Cloud controller application status thread terminated");
- }
- }
+// public void execute() {
+// statusEventReceiver.setExecutorService(executorService);
+// statusEventReceiver.execute();
+// if (log.isInfoEnabled()) {
+// log.info("Cloud controller application status thread started");
+// }
+//
+// if (log.isInfoEnabled()) {
+// log.info("Cloud controller application status thread terminated");
+// }
+// }
 
  private void addEventListeners() {
  statusEventReceiver.addEventListener(new InstanceActivatedEventListener() {

diff --git a/...e.stratos.common/src/main/java/org/apache/stratos/common/threading/StratosThreadPool.java b/...e.stratos.common/src/main/java/org/apache/stratos/common/threading/StratosThreadPool.java
@@ -36,8 +36,8 @@ public class StratosThreadPool {
 
  private static final Log log = LogFactory.getLog(StratosThreadPool.class);
 
- private static Map<String, ExecutorService> executorServiceMap = new ConcurrentHashMap<String, ExecutorService>();
- private static Map<String, ScheduledExecutorService> scheduledServiceMap = new ConcurrentHashMap<String, ScheduledExecutorService>();
+ private static volatile Map<String, ExecutorService> executorServiceMap = new ConcurrentHashMap<>();
+ private static volatile Map<String, ScheduledExecutorService> scheduledServiceMap = new ConcurrentHashMap<String, ScheduledExecutorService>();
  private static Object executorServiceMapLock = new Object();
  private static Object scheduledServiceMapLock = new Object();
 

diff --git a/...ger/src/main/java/org/apache/stratos/manager/internal/StratosManagerServiceComponent.java b/...ger/src/main/java/org/apache/stratos/manager/internal/StratosManagerServiceComponent.java
@@ -202,8 +202,8 @@ private void initializeInitializerEventReceiver() {
  */
  private void initializeInstanceStatusEventReceiver() {
  instanceStatusEventReceiver = new StratosManagerInstanceStatusEventReceiver();
- instanceStatusEventReceiver.setExecutorService(executorService);
- instanceStatusEventReceiver.execute();
+// instanceStatusEventReceiver.setExecutorService(executorService);
+// instanceStatusEventReceiver.execute();
  }
 
  /**

diff --git a/.../apache/stratos/manager/messaging/receiver/StratosManagerInstanceStatusEventReceiver.java b/.../apache/stratos/manager/messaging/receiver/StratosManagerInstanceStatusEventReceiver.java
@@ -37,32 +37,34 @@
 /**
  * Stratos manager instance status event receiver.
  */
-public class StratosManagerInstanceStatusEventReceiver extends InstanceStatusEventReceiver {
+public class StratosManagerInstanceStatusEventReceiver {
 
  private static final Log log = LogFactory.getLog(StratosManagerInstanceStatusEventReceiver.class);
 
  private ApplicationSignUpHandler signUpManager;
  private ArtifactDistributionCoordinator artifactDistributionCoordinator;
+ private InstanceStatusEventReceiver instanceStatusEventReceiver;
 
 
  public StratosManagerInstanceStatusEventReceiver() {
  signUpManager = new ApplicationSignUpHandler();
  artifactDistributionCoordinator = new ArtifactDistributionCoordinator();
+ instanceStatusEventReceiver = InstanceStatusEventReceiver.getInstance();
 
  addEventListeners();
  }
 
- @Override
- public void execute() {
- super.execute();
-
- if (log.isInfoEnabled()) {
- log.info("Stratos manager instance status event receiver thread started");
- }
- }
+// @Override
+// public void execute() {
+// super.execute();
+//
+// if (log.isInfoEnabled()) {
+// log.info("Stratos manager instance status event receiver thread started");
+// }
+// }
 
  private void addEventListeners() {
- addEventListener(new InstanceStartedEventListener() {
+ instanceStatusEventReceiver.addEventListener(new InstanceStartedEventListener() {
  @Override
  protected void onEvent(Event event) {
  InstanceStartedEvent instanceStartedEvent = (InstanceStartedEvent) event;