Remove unnessary threads in messaging model
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/038b72e3 Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/038b72e3 Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/038b72e3 Branch: refs/heads/master Commit: 038b72e39a88f0fbdf30895061f427f0e535666f Parents: 8012f8c Author: gayan <[email protected]> Authored: Mon Dec 1 16:31:27 2014 +0530 Committer: gayan <[email protected]> Committed: Tue Dec 2 16:37:47 2014 +0530 ---------------------------------------------------------------------- .../extension/api/LoadBalancerExtension.java | 254 +++++++++---------- 1 file changed, 127 insertions(+), 127 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/stratos/blob/038b72e3/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java ---------------------------------------------------------------------- diff --git a/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java b/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java index 188b2ac..24fecf3 100644 --- a/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java +++ b/components/org.apache.stratos.load.balancer.extension.api/src/main/java/org/apache/stratos/load/balancer/extension/api/LoadBalancerExtension.java @@ -35,134 +35,134 @@ import java.util.concurrent.ExecutorService; * received from the message broker. */ public class LoadBalancerExtension implements Runnable { - private static final Log log = LogFactory.getLog(LoadBalancerExtension.class); - - private LoadBalancer loadBalancer; - private LoadBalancerStatisticsReader statsReader; - private boolean loadBalancerStarted; - private TopologyEventReceiver topologyEventReceiver; - private LoadBalancerStatisticsNotifier statisticsNotifier; - private boolean terminated; + private static final Log log = LogFactory.getLog(LoadBalancerExtension.class); + + private LoadBalancer loadBalancer; + private LoadBalancerStatisticsReader statsReader; + private boolean loadBalancerStarted; + private TopologyEventReceiver topologyEventReceiver; + private LoadBalancerStatisticsNotifier statisticsNotifier; + private boolean terminated; private ExecutorService executorService; - - /** - * Load balancer extension constructor. - * - * @param loadBalancer Load balancer instance: Mandatory. - * @param statsReader Statistics reader: If null statistics notifier thread will not be started. - */ - public LoadBalancerExtension(LoadBalancer loadBalancer, LoadBalancerStatisticsReader statsReader) { - this.loadBalancer = loadBalancer; - this.statsReader = statsReader; - } - - @Override - public void run() { - try { - if (log.isInfoEnabled()) { - log.info("Load balancer extension started"); - } - - // Start topology receiver thread - topologyEventReceiver = new TopologyEventReceiver(); - addEventListeners(); - topologyEventReceiver.setExecutorService(executorService); - topologyEventReceiver.execute(); - - if (statsReader != null) { - // Start stats notifier thread - statisticsNotifier = new LoadBalancerStatisticsNotifier(statsReader); - Thread statsNotifierThread = new Thread(statisticsNotifier); - statsNotifierThread.start(); - } else { - if (log.isWarnEnabled()) { - log.warn("Load balancer statistics reader not found"); - } - } - - } catch (Exception e) { - if (log.isErrorEnabled()) { - log.error("Could not start load balancer extension", e); - } - } - } - - private void addEventListeners() { - topologyEventReceiver.addEventListener(new CompleteTopologyEventListener() { - - @Override - protected void onEvent(Event event) { - try { - - if (!loadBalancerStarted) { - // Configure load balancer - loadBalancer.configure(TopologyManager.getTopology()); - - // Start load balancer - loadBalancer.start(); - loadBalancerStarted = true; - } - } catch (Exception e) { - if (log.isErrorEnabled()) { - log.error("Could not start load balancer", e); - } - terminate(); - } - } - }); - topologyEventReceiver.addEventListener(new MemberActivatedEventListener() { - @Override - protected void onEvent(Event event) { - reloadConfiguration(); - } - }); - topologyEventReceiver.addEventListener(new MemberSuspendedEventListener() { - @Override - protected void onEvent(Event event) { - reloadConfiguration(); - } - }); - topologyEventReceiver.addEventListener(new MemberTerminatedEventListener() { - @Override - protected void onEvent(Event event) { - reloadConfiguration(); - } - }); - topologyEventReceiver.addEventListener(new ClusterRemovedEventListener() { - @Override - protected void onEvent(Event event) { - reloadConfiguration(); - } - }); - topologyEventReceiver.addEventListener(new ServiceRemovedEventListener() { - @Override - protected void onEvent(Event event) { - reloadConfiguration(); - } - }); - } - - private void reloadConfiguration() { - try { - if (loadBalancerStarted) { - loadBalancer.reload(TopologyManager.getTopology()); - } - } catch (Exception e) { - if (log.isErrorEnabled()) { - log.error("Could not reload load balancer configuration", e); - } - } - } - - public void terminate() { - if (topologyEventReceiver != null) { - topologyEventReceiver.terminate(); - } - if (statisticsNotifier != null) { - statisticsNotifier.terminate(); - } - terminated = true; - } + /** + * Load balancer extension constructor. + * @param loadBalancer Load balancer instance: Mandatory. + * @param statsReader Statistics reader: If null statistics notifier thread will not be started. + */ + public LoadBalancerExtension(LoadBalancer loadBalancer, LoadBalancerStatisticsReader statsReader) { + this.loadBalancer = loadBalancer; + this.statsReader = statsReader; + } + + @Override + public void run() { + try { + if(log.isInfoEnabled()) { + log.info("Load balancer extension started"); + } + + // Start topology receiver thread + topologyEventReceiver = new TopologyEventReceiver(); + addEventListeners(); + topologyEventReceiver.setExecutorService(executorService); + topologyEventReceiver.execute(); + + + if(statsReader != null) { + // Start stats notifier thread + statisticsNotifier = new LoadBalancerStatisticsNotifier(statsReader); + Thread statsNotifierThread = new Thread(statisticsNotifier); + statsNotifierThread.start(); + } + else { + if(log.isWarnEnabled()) { + log.warn("Load balancer statistics reader not found"); + } + } + + } catch (Exception e) { + if (log.isErrorEnabled()) { + log.error("Could not start load balancer extension", e); + } + } + } + + private void addEventListeners() { + topologyEventReceiver.addEventListener(new CompleteTopologyEventListener() { + + @Override + protected void onEvent(Event event) { + try { + + if (!loadBalancerStarted) { + // Configure load balancer + loadBalancer.configure(TopologyManager.getTopology()); + + // Start load balancer + loadBalancer.start(); + loadBalancerStarted = true; + } + } catch (Exception e) { + if (log.isErrorEnabled()) { + log.error("Could not start load balancer", e); + } + terminate(); + } + } + }); + topologyEventReceiver.addEventListener(new MemberActivatedEventListener() { + @Override + protected void onEvent(Event event) { + reloadConfiguration(); + } + }); + topologyEventReceiver.addEventListener(new MemberSuspendedEventListener() { + @Override + protected void onEvent(Event event) { + reloadConfiguration(); + } + }); + topologyEventReceiver.addEventListener(new MemberTerminatedEventListener() { + @Override + protected void onEvent(Event event) { + reloadConfiguration(); + } + }); + topologyEventReceiver.addEventListener(new ClusterRemovedEventListener() { + @Override + protected void onEvent(Event event) { + reloadConfiguration(); + } + }); + topologyEventReceiver.addEventListener(new ServiceRemovedEventListener() { + @Override + protected void onEvent(Event event) { + reloadConfiguration(); + } + }); + } + + private void reloadConfiguration() { + try { + if (loadBalancerStarted) { + loadBalancer.reload(TopologyManager.getTopology()); + } + } catch (Exception e) { + if (log.isErrorEnabled()) { + log.error("Could not reload load balancer configuration", e); + } + } + } + + public void terminate() { + if(topologyEventReceiver != null) { + topologyEventReceiver.terminate(); + } + if(statisticsNotifier != null) { + statisticsNotifier.terminate(); + } + terminated = true; + } public ExecutorService getExecutorService() { return executorService;
