You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stratos.apache.org by la...@apache.org on 2013/10/22 16:25:44 UTC
git commit: Changing topology lock as a global lock and pacckage
refactoring
Updated Branches:
refs/heads/master 0b5d031ef -> e56c1ed24
Changing topology lock as a global lock and pacckage refactoring
Project: http://git-wip-us.apache.org/repos/asf/incubator-stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-stratos/commit/e56c1ed2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-stratos/tree/e56c1ed2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-stratos/diff/e56c1ed2
Branch: refs/heads/master
Commit: e56c1ed243cff734d8673d689d83fed44232ee05
Parents: 0b5d031
Author: Lahiru Sandaruwan <la...@apache.org>
Authored: Tue Oct 22 20:00:38 2013 +0530
Committer: Lahiru Sandaruwan <la...@apache.org>
Committed: Tue Oct 22 20:00:38 2013 +0530
----------------------------------------------------------------------
.../client/TopicSubscriberManager.java | 35 ------
.../health/HealthEventMessageDelegator.java | 78 -------------
.../health/HealthEventMessageReceiver.java | 50 ---------
.../client/health/HealthEventQueue.java | 44 --------
.../client/health/HealthStatManager.java | 70 ------------
.../topology/TopologyEventMessageDelegator.java | 109 ------------------
.../topology/TopologyEventMessageReceiver.java | 50 ---------
.../client/topology/TopologyEventQueue.java | 44 --------
.../client/topology/TopologyManager.java | 58 ----------
.../internal/AutoscalerServerComponent.java | 11 +-
.../AverageRequestInFlightEventProcessor.java | 6 +-
...GradientOfRequestInFlightEventProcessor.java | 7 +-
...rivativeOfRequestInFlightEventProcessor.java | 7 +-
.../receiver/TopicSubscriberManager.java | 70 ++++++++++++
.../health/HealthEventMessageDelegator.java | 79 +++++++++++++
.../health/HealthEventMessageReceiver.java | 50 +++++++++
.../receiver/health/HealthEventQueue.java | 44 ++++++++
.../topology/TopologyEventMessageDelegator.java | 110 +++++++++++++++++++
.../topology/TopologyEventMessageReceiver.java | 50 +++++++++
.../receiver/topology/TopologyEventQueue.java | 44 ++++++++
20 files changed, 465 insertions(+), 551 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/TopicSubscriberManager.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/TopicSubscriberManager.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/TopicSubscriberManager.java
deleted file mode 100644
index 9b10880..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/TopicSubscriberManager.java
+++ /dev/null
@@ -1,35 +0,0 @@
-package org.apache.stratos.autoscaler.client;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.client.topology.TopologyEventMessageReceiver;
-import org.apache.stratos.messaging.broker.subscribe.TopicSubscriber;
-import org.apache.stratos.messaging.util.Constants;
-
-/**
- * This class will initiate all the subscriber to topics
- */
-public class TopicSubscriberManager {
- private static final Log log = LogFactory.getLog(TopicSubscriberManager.class);
-
- public void subscribeAllTopics(){
- TopicSubscriber topologyTopicSubscriber = new TopicSubscriber(Constants.TOPOLOGY_TOPIC);
- topologyTopicSubscriber.setMessageListener(new TopologyEventMessageReceiver());
- Thread topologyTopicSubscriberThread = new Thread(topologyTopicSubscriber);
- topologyTopicSubscriberThread.start();
-
- if (log.isDebugEnabled()) {
- log.debug("Topology event message receiver thread started");
- }
-
- TopicSubscriber healthStatTopicSubscriber = new TopicSubscriber(Constants.HEALTH_STAT_TOPIC);
- healthStatTopicSubscriber.setMessageListener(new TopologyEventMessageReceiver());
- Thread healthStatTopicSubscriberThread = new Thread(healthStatTopicSubscriber);
- healthStatTopicSubscriberThread.start();
-
- if (log.isDebugEnabled()) {
- log.debug("Health Stat event message receiver thread started");
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventMessageDelegator.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventMessageDelegator.java
deleted file mode 100644
index 90f3054..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventMessageDelegator.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-package org.apache.stratos.autoscaler.client.health;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.message.processor.AverageRequestInFlightEventProcessor;
-import org.apache.stratos.autoscaler.message.processor.GradientOfRequestInFlightEventProcessor;
-import org.apache.stratos.autoscaler.message.processor.SecondDerivativeOfRequestInFlightEventProcessor;
-import org.apache.stratos.messaging.util.Constants;
-
-import javax.jms.TextMessage;
-
-
-/**
- * A thread for processing topology messages and updating the topology data structure.
- */
-public class HealthEventMessageDelegator implements Runnable {
-
- private static final Log log = LogFactory.getLog(HealthEventMessageDelegator.class);
-
- @Override
- public void run() {
- log.info("Health stat event message processor started");
-
- // instantiate all the relevant processors
- AverageRequestInFlightEventProcessor processor1 = new AverageRequestInFlightEventProcessor();
- GradientOfRequestInFlightEventProcessor processor2 = new GradientOfRequestInFlightEventProcessor();
- SecondDerivativeOfRequestInFlightEventProcessor processor3 = new SecondDerivativeOfRequestInFlightEventProcessor();
-
- // link all the relevant processors in the required order
- processor1.setNext(processor2);
- processor2.setNext(processor3);
-
- while (true) {
- try {
- TextMessage message = HealthEventQueue.getInstance().take();
-
- // retrieve the header
- String type = message.getStringProperty(Constants.EVENT_CLASS_NAME);
- // retrieve the actual message
- String json = message.getText();
-
- if (log.isDebugEnabled()) {
- log.debug(String.format("Event message received from queue: %s", type));
- }
-
- try {
- HealthStatManager.acquireWriteLock();
- processor1.process(type, json);
- } finally {
- HealthStatManager.releaseWriteLock();
- }
-
- } catch (Exception e) {
- String error = "Failed to retrieve the topology event message.";
- log.error(error, e);
- throw new RuntimeException(error, e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventMessageReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventMessageReceiver.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventMessageReceiver.java
deleted file mode 100644
index bccb57f..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventMessageReceiver.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-package org.apache.stratos.autoscaler.client.health;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.TextMessage;
-
-public class HealthEventMessageReceiver implements MessageListener {
-
- private static final Log log = LogFactory.getLog(HealthEventMessageReceiver.class);
-
- @Override
- public void onMessage(Message message) {
- if (message instanceof TextMessage) {
- TextMessage receivedMessage = (TextMessage) message;
- log.info(receivedMessage);
- try {
- if (log.isDebugEnabled()) {
- log.debug("Message received: " + ((TextMessage) message).getText());
- }
- // Add received message to the queue
- HealthEventQueue.getInstance().add(receivedMessage);
-
- } catch (JMSException e) {
- log.error(e.getMessage(), e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventQueue.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventQueue.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventQueue.java
deleted file mode 100644
index c1d5601..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthEventQueue.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.stratos.autoscaler.client.health;
-
-import javax.jms.TextMessage;
-import java.util.concurrent.LinkedBlockingQueue;
-
-/**
- * Implements topology event queue.
- */
-public class HealthEventQueue extends LinkedBlockingQueue<TextMessage>{
- private static volatile HealthEventQueue instance;
-
- private HealthEventQueue(){
- }
-
- public static synchronized HealthEventQueue getInstance() {
- if (instance == null) {
- synchronized (HealthEventQueue.class){
- if (instance == null) {
- instance = new HealthEventQueue();
- }
- }
- }
- return instance;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthStatManager.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthStatManager.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthStatManager.java
deleted file mode 100644
index 7652f6c..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/health/HealthStatManager.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.stratos.autoscaler.client.health;
-
-//import org.apache.stratos.messaging.domain.topology.Topology;
-
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-public class HealthStatManager {
-// private static volatile Topology topology;
- private static volatile ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
- private static volatile ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
- private static volatile ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
-
- public static void acquireReadLock() {
- readLock.lock();
- }
-
- public static void releaseReadLock() {
- readLock.unlock();
- }
-
- public static void acquireWriteLock() {
- writeLock.lock();
- }
-
- public static void releaseWriteLock() {
- writeLock.unlock();
- }
-
-// public static synchronized Topology getTopology() {
-// if (topology == null) {
-// synchronized (HealthStatManager.class){
-// if (topology == null) {
-// topology = new Topology();
-// }
-// }
-// }
-// return topology;
-// }
-
-
-// public static synchronized Topology getTopology() {
-// if (topology == null) {
-// synchronized (HealthStatManager.class){
-// if (topology == null) {
-// topology = new Topology();
-// }
-// }
-// }
-// return topology;
-// }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventMessageDelegator.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventMessageDelegator.java
deleted file mode 100644
index b6a532a..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventMessageDelegator.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-package org.apache.stratos.autoscaler.client.topology;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.message.processor.*;
-import org.apache.stratos.messaging.util.Constants;
-
-import javax.jms.TextMessage;
-
-
-/**
- * A thread for processing topology messages and updating the topology data structure.
- */
-public class TopologyEventMessageDelegator implements Runnable {
-
- private static final Log log = LogFactory.getLog(TopologyEventMessageDelegator.class);
-
- @Override
- public void run() {
- log.info("Topology event message processor started");
- log.info("Waiting for the complete topology event message...");
- while (true) {
- try {
- // First take the complete topology event
- TextMessage message = TopologyEventQueue.getInstance().take();
-
- // retrieve the header
- String type = message.getStringProperty(Constants.EVENT_CLASS_NAME);
- // retrieve the actual message
- String json = message.getText();
-
- CompleteTopologyEventProcessor completeTopologyEventProcessor = new CompleteTopologyEventProcessor();
- if(completeTopologyEventProcessor.process(type, json, TopologyManager.getTopology())) {
- break;
- }
-
- } catch (Exception e) {
- log.error("Failed to retrieve the full topology.", e);
- throw new RuntimeException("Failed to retrieve the full topology.", e);
- }
- }
-
- // instantiate all the relevant processors
- ServiceCreatedEventProcessor processor1 = new ServiceCreatedEventProcessor();
- ServiceRemovedEventProcessor processor2 = new ServiceRemovedEventProcessor();
- ClusterCreatedEventProcessor processor3 = new ClusterCreatedEventProcessor();
- ClusterRemovedEventProcessor processor4 = new ClusterRemovedEventProcessor();
- InstanceSpawnedEventProcessor processor5 = new InstanceSpawnedEventProcessor();
- MemberStartedEventProcessor processor6 = new MemberStartedEventProcessor();
- MemberActivatedEventProcessor processor7 = new MemberActivatedEventProcessor();
- MemberSuspendedEventProcessor processor8 = new MemberSuspendedEventProcessor();
- MemberTerminatedEventProcessor processor9 = new MemberTerminatedEventProcessor();
-
- // link all the relevant processors in the required order
- processor1.setNext(processor2);
- processor2.setNext(processor3);
- processor3.setNext(processor4);
- processor4.setNext(processor5);
- processor5.setNext(processor6);
- processor6.setNext(processor7);
- processor7.setNext(processor8);
- processor8.setNext(processor9);
-
- while (true) {
- try {
- TextMessage message = TopologyEventQueue.getInstance().take();
-
- // retrieve the header
- String type = message.getStringProperty(Constants.EVENT_CLASS_NAME);
- // retrieve the actual message
- String json = message.getText();
-
- if (log.isDebugEnabled()) {
- log.debug(String.format("Event message received from queue: %s", type));
- }
-
- try {
- TopologyManager.acquireWriteLock();
- processor1.process(type, json, TopologyManager.getTopology());
- } finally {
- TopologyManager.releaseWriteLock();
- }
-
- } catch (Exception e) {
- String error = "Failed to retrieve the topology event message.";
- log.error(error, e);
- throw new RuntimeException(error, e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventMessageReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventMessageReceiver.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventMessageReceiver.java
deleted file mode 100644
index 41330f9..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventMessageReceiver.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-package org.apache.stratos.autoscaler.client.topology;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.TextMessage;
-
-public class TopologyEventMessageReceiver implements MessageListener {
-
- private static final Log log = LogFactory.getLog(TopologyEventMessageReceiver.class);
-
- @Override
- public void onMessage(Message message) {
- if (message instanceof TextMessage) {
- TextMessage receivedMessage = (TextMessage) message;
- log.info(receivedMessage);
- try {
- if (log.isDebugEnabled()) {
- log.debug("Message received: " + ((TextMessage) message).getText());
- }
- // Add received message to the queue
- TopologyEventQueue.getInstance().add(receivedMessage);
-
- } catch (JMSException e) {
- log.error(e.getMessage(), e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventQueue.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventQueue.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventQueue.java
deleted file mode 100644
index 9d094dd..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyEventQueue.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.stratos.autoscaler.client.topology;
-
-import javax.jms.TextMessage;
-import java.util.concurrent.LinkedBlockingQueue;
-
-/**
- * Implements topology event queue.
- */
-public class TopologyEventQueue extends LinkedBlockingQueue<TextMessage>{
- private static volatile TopologyEventQueue instance;
-
- private TopologyEventQueue(){
- }
-
- public static synchronized TopologyEventQueue getInstance() {
- if (instance == null) {
- synchronized (TopologyEventQueue.class){
- if (instance == null) {
- instance = new TopologyEventQueue ();
- }
- }
- }
- return instance;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyManager.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyManager.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyManager.java
deleted file mode 100644
index 92508cf..0000000
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/client/topology/TopologyManager.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-package org.apache.stratos.autoscaler.client.topology;
-
-import org.apache.stratos.messaging.domain.topology.Topology;
-
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-public class TopologyManager {
- private static volatile Topology topology;
- private static volatile ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
- private static volatile ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
- private static volatile ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
-
- public static void acquireReadLock() {
- readLock.lock();
- }
-
- public static void releaseReadLock() {
- readLock.unlock();
- }
-
- public static void acquireWriteLock() {
- writeLock.lock();
- }
-
- public static void releaseWriteLock() {
- writeLock.unlock();
- }
-
- public static synchronized Topology getTopology() {
- if (topology == null) {
- synchronized (TopologyManager.class){
- if (topology == null) {
- topology = new Topology();
- }
- }
- }
- return topology;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java
index e138973..a2ccbce 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/internal/AutoscalerServerComponent.java
@@ -20,7 +20,7 @@ package org.apache.stratos.autoscaler.internal;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.client.TopicSubscriberManager;
+import org.apache.stratos.autoscaler.message.receiver.TopicSubscriberManager;
import org.apache.stratos.autoscaler.rule.ExecutorTaskScheduler;
import org.osgi.service.component.ComponentContext;
@@ -40,18 +40,25 @@ public class AutoscalerServerComponent {
if (log.isInfoEnabled()) {
log.info("Autoscaler Server Component activated");
}
+
+ //Subscribe to all topics
TopicSubscriberManager topicSubscriberManager = new TopicSubscriberManager();
topicSubscriberManager.subscribeAllTopics();
if (log.isDebugEnabled()) {
log.debug("Topology event message receiver thread started");
}
+
+ //Start scheduler for running rules
ExecutorTaskScheduler executor = new ExecutorTaskScheduler();
executor.start();
-//
+
if (log.isInfoEnabled()) {
log.info("Rules executor started");
}
+
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/AverageRequestInFlightEventProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/AverageRequestInFlightEventProcessor.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/AverageRequestInFlightEventProcessor.java
index 733e678..df34a9e 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/AverageRequestInFlightEventProcessor.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/AverageRequestInFlightEventProcessor.java
@@ -20,7 +20,7 @@ package org.apache.stratos.autoscaler.message.processor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.client.topology.TopologyManager;
+import org.apache.stratos.autoscaler.message.receiver.TopicSubscriberManager;
import org.apache.stratos.autoscaler.event.AverageRequestsInFlightEvent;
import org.apache.stratos.messaging.domain.topology.Service;
import org.apache.stratos.messaging.util.Util;
@@ -48,10 +48,10 @@ public class AverageRequestInFlightEventProcessor implements HealthStatEventProc
String clusterId = event.getClusterId();
//Get all values of services Map from topology
- for (Service service : ((Map<String, Service>) TopologyManager.getTopology().getServices()).values())
+ for (Service service : ((Map<String, Service>) TopicSubscriberManager.getTopology().getServices()).values())
{
if(service.clusterExists(clusterId)){
- ((Map<String, Service>) TopologyManager.getTopology().getServices())
+ ((Map<String, Service>) TopicSubscriberManager.getTopology().getServices())
.get(service.getServiceName()).getCluster(clusterId)
.setAverageRequestsInFlight(event.getValue());
}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/GradientOfRequestInFlightEventProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/GradientOfRequestInFlightEventProcessor.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/GradientOfRequestInFlightEventProcessor.java
index 4783dc0..f06082b 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/GradientOfRequestInFlightEventProcessor.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/GradientOfRequestInFlightEventProcessor.java
@@ -20,10 +20,9 @@ package org.apache.stratos.autoscaler.message.processor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.client.topology.TopologyManager;
+import org.apache.stratos.autoscaler.message.receiver.TopicSubscriberManager;
import org.apache.stratos.autoscaler.event.AverageRequestsInFlightEvent;
import org.apache.stratos.autoscaler.event.GradientOfRequestsInFlightEvent;
-import org.apache.stratos.messaging.domain.topology.Cluster;
import org.apache.stratos.messaging.domain.topology.Service;
import org.apache.stratos.messaging.util.Util;
@@ -50,11 +49,11 @@ public class GradientOfRequestInFlightEventProcessor implements HealthStatEventP
String clusterId = event.getClusterId();
//Get all values of services Map from topology
- for (Service service : ((Map<String, Service>) TopologyManager.getTopology().getServices()).values())
+ for (Service service : ((Map<String, Service>) TopicSubscriberManager.getTopology().getServices()).values())
{
if(service.clusterExists(clusterId)){
- ((Map<String, Service>) TopologyManager.getTopology().getServices())
+ ((Map<String, Service>) TopicSubscriberManager.getTopology().getServices())
.get(service.getServiceName()).getCluster(clusterId)
.setRequestsInFlightGradient(event.getValue());
}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/SecondDerivativeOfRequestInFlightEventProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/SecondDerivativeOfRequestInFlightEventProcessor.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/SecondDerivativeOfRequestInFlightEventProcessor.java
index ba417b5..d2321d9 100644
--- a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/SecondDerivativeOfRequestInFlightEventProcessor.java
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/processor/SecondDerivativeOfRequestInFlightEventProcessor.java
@@ -20,10 +20,9 @@ package org.apache.stratos.autoscaler.message.processor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.autoscaler.client.topology.TopologyManager;
+import org.apache.stratos.autoscaler.message.receiver.TopicSubscriberManager;
import org.apache.stratos.autoscaler.event.AverageRequestsInFlightEvent;
import org.apache.stratos.autoscaler.event.SecondDerivativeOfRequestsInFlightEvent;
-import org.apache.stratos.messaging.domain.topology.Cluster;
import org.apache.stratos.messaging.domain.topology.Service;
import org.apache.stratos.messaging.util.Util;
@@ -51,11 +50,11 @@ public class SecondDerivativeOfRequestInFlightEventProcessor implements HealthSt
String clusterId = event.getClusterId();
//Get all values of services Map from topology
- for (Service service : ((Map<String, Service>) TopologyManager.getTopology().getServices()).values())
+ for (Service service : ((Map<String, Service>) TopicSubscriberManager.getTopology().getServices()).values())
{
if(service.clusterExists(clusterId)){
- ((Map<String, Service>) TopologyManager.getTopology().getServices())
+ ((Map<String, Service>) TopicSubscriberManager.getTopology().getServices())
.get(service.getServiceName()).getCluster(clusterId)
.setRequestsInFlightSecondDerivative(event.getValue());
}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/TopicSubscriberManager.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/TopicSubscriberManager.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/TopicSubscriberManager.java
new file mode 100644
index 0000000..6bed22a
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/TopicSubscriberManager.java
@@ -0,0 +1,70 @@
+package org.apache.stratos.autoscaler.message.receiver;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.autoscaler.message.receiver.topology.TopologyEventMessageReceiver;
+import org.apache.stratos.messaging.broker.subscribe.TopicSubscriber;
+import org.apache.stratos.messaging.domain.topology.Topology;
+import org.apache.stratos.messaging.util.Constants;
+
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * This class will initiate all the subscriber to topics
+ */
+public class TopicSubscriberManager {
+ private static final Log log = LogFactory.getLog(TopicSubscriberManager.class);
+ private static volatile Topology topology;
+ private static volatile ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ private static volatile ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
+ private static volatile ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
+
+ public static void acquireReadLock() {
+ readLock.lock();
+ }
+
+ public static void releaseReadLock() {
+ readLock.unlock();
+ }
+
+ public static void acquireWriteLock() {
+ writeLock.lock();
+ }
+
+ public static void releaseWriteLock() {
+ writeLock.unlock();
+ }
+
+ public static synchronized Topology getTopology() {
+ if (topology == null) {
+ synchronized (TopicSubscriberManager.class){
+ if (topology == null) {
+ topology = new Topology();
+ }
+ }
+ }
+ return topology;
+ }
+
+ public void subscribeAllTopics(){
+
+ TopicSubscriber topologyTopicSubscriber = new TopicSubscriber(Constants.TOPOLOGY_TOPIC);
+ topologyTopicSubscriber.setMessageListener(new TopologyEventMessageReceiver());
+ Thread topologyTopicSubscriberThread = new Thread(topologyTopicSubscriber);
+ topologyTopicSubscriberThread.start();
+
+ if (log.isDebugEnabled()) {
+ log.debug("Topology event message receiver thread started");
+ }
+
+ TopicSubscriber healthStatTopicSubscriber = new TopicSubscriber(Constants.HEALTH_STAT_TOPIC);
+ healthStatTopicSubscriber.setMessageListener(new TopologyEventMessageReceiver());
+ Thread healthStatTopicSubscriberThread = new Thread(healthStatTopicSubscriber);
+ healthStatTopicSubscriberThread.start();
+
+ if (log.isDebugEnabled()) {
+ log.debug("Health Stat event message receiver thread started");
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventMessageDelegator.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventMessageDelegator.java
new file mode 100644
index 0000000..d183c14
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventMessageDelegator.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.stratos.autoscaler.message.receiver.health;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.autoscaler.message.processor.AverageRequestInFlightEventProcessor;
+import org.apache.stratos.autoscaler.message.processor.GradientOfRequestInFlightEventProcessor;
+import org.apache.stratos.autoscaler.message.processor.SecondDerivativeOfRequestInFlightEventProcessor;
+import org.apache.stratos.autoscaler.message.receiver.TopicSubscriberManager;
+import org.apache.stratos.messaging.util.Constants;
+
+import javax.jms.TextMessage;
+
+
+/**
+ * A thread for processing topology messages and updating the topology data structure.
+ */
+public class HealthEventMessageDelegator implements Runnable {
+
+ private static final Log log = LogFactory.getLog(HealthEventMessageDelegator.class);
+
+ @Override
+ public void run() {
+ log.info("Health stat event message processor started");
+
+ // instantiate all the relevant processors
+ AverageRequestInFlightEventProcessor processor1 = new AverageRequestInFlightEventProcessor();
+ GradientOfRequestInFlightEventProcessor processor2 = new GradientOfRequestInFlightEventProcessor();
+ SecondDerivativeOfRequestInFlightEventProcessor processor3 = new SecondDerivativeOfRequestInFlightEventProcessor();
+
+ // link all the relevant processors in the required order
+ processor1.setNext(processor2);
+ processor2.setNext(processor3);
+
+ while (true) {
+ try {
+ TextMessage message = HealthEventQueue.getInstance().take();
+
+ // retrieve the header
+ String type = message.getStringProperty(Constants.EVENT_CLASS_NAME);
+ // retrieve the actual message
+ String json = message.getText();
+
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Event message received from queue: %s", type));
+ }
+
+ try {
+ TopicSubscriberManager.acquireWriteLock();
+ processor1.process(type, json);
+ } finally {
+ TopicSubscriberManager.releaseWriteLock();
+ }
+
+ } catch (Exception e) {
+ String error = "Failed to retrieve the topology event message.";
+ log.error(error, e);
+ throw new RuntimeException(error, e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventMessageReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventMessageReceiver.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventMessageReceiver.java
new file mode 100644
index 0000000..85c04dd
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventMessageReceiver.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.stratos.autoscaler.message.receiver.health;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.TextMessage;
+
+public class HealthEventMessageReceiver implements MessageListener {
+
+ private static final Log log = LogFactory.getLog(HealthEventMessageReceiver.class);
+
+ @Override
+ public void onMessage(Message message) {
+ if (message instanceof TextMessage) {
+ TextMessage receivedMessage = (TextMessage) message;
+ log.info(receivedMessage);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("Message received: " + ((TextMessage) message).getText());
+ }
+ // Add received message to the queue
+ HealthEventQueue.getInstance().add(receivedMessage);
+
+ } catch (JMSException e) {
+ log.error(e.getMessage(), e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventQueue.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventQueue.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventQueue.java
new file mode 100644
index 0000000..6544d4f
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/health/HealthEventQueue.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.stratos.autoscaler.message.receiver.health;
+
+import javax.jms.TextMessage;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * Implements topology event queue.
+ */
+public class HealthEventQueue extends LinkedBlockingQueue<TextMessage>{
+ private static volatile HealthEventQueue instance;
+
+ private HealthEventQueue(){
+ }
+
+ public static synchronized HealthEventQueue getInstance() {
+ if (instance == null) {
+ synchronized (HealthEventQueue.class){
+ if (instance == null) {
+ instance = new HealthEventQueue();
+ }
+ }
+ }
+ return instance;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageDelegator.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageDelegator.java
new file mode 100644
index 0000000..febac47
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageDelegator.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.stratos.autoscaler.message.receiver.topology;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.autoscaler.message.receiver.TopicSubscriberManager;
+import org.apache.stratos.messaging.message.processor.*;
+import org.apache.stratos.messaging.util.Constants;
+
+import javax.jms.TextMessage;
+
+
+/**
+ * A thread for processing topology messages and updating the topology data structure.
+ */
+public class TopologyEventMessageDelegator implements Runnable {
+
+ private static final Log log = LogFactory.getLog(TopologyEventMessageDelegator.class);
+
+ @Override
+ public void run() {
+ log.info("Topology event message processor started");
+ log.info("Waiting for the complete topology event message...");
+ while (true) {
+ try {
+ // First take the complete topology event
+ TextMessage message = TopologyEventQueue.getInstance().take();
+
+ // retrieve the header
+ String type = message.getStringProperty(Constants.EVENT_CLASS_NAME);
+ // retrieve the actual message
+ String json = message.getText();
+
+ CompleteTopologyEventProcessor completeTopologyEventProcessor = new CompleteTopologyEventProcessor();
+ if(completeTopologyEventProcessor.process(type, json, TopicSubscriberManager.getTopology())) {
+ break;
+ }
+
+ } catch (Exception e) {
+ log.error("Failed to retrieve the full topology.", e);
+ throw new RuntimeException("Failed to retrieve the full topology.", e);
+ }
+ }
+
+ // instantiate all the relevant processors
+ ServiceCreatedEventProcessor processor1 = new ServiceCreatedEventProcessor();
+ ServiceRemovedEventProcessor processor2 = new ServiceRemovedEventProcessor();
+ ClusterCreatedEventProcessor processor3 = new ClusterCreatedEventProcessor();
+ ClusterRemovedEventProcessor processor4 = new ClusterRemovedEventProcessor();
+ InstanceSpawnedEventProcessor processor5 = new InstanceSpawnedEventProcessor();
+ MemberStartedEventProcessor processor6 = new MemberStartedEventProcessor();
+ MemberActivatedEventProcessor processor7 = new MemberActivatedEventProcessor();
+ MemberSuspendedEventProcessor processor8 = new MemberSuspendedEventProcessor();
+ MemberTerminatedEventProcessor processor9 = new MemberTerminatedEventProcessor();
+
+ // link all the relevant processors in the required order
+ processor1.setNext(processor2);
+ processor2.setNext(processor3);
+ processor3.setNext(processor4);
+ processor4.setNext(processor5);
+ processor5.setNext(processor6);
+ processor6.setNext(processor7);
+ processor7.setNext(processor8);
+ processor8.setNext(processor9);
+
+ while (true) {
+ try {
+ TextMessage message = TopologyEventQueue.getInstance().take();
+
+ // retrieve the header
+ String type = message.getStringProperty(Constants.EVENT_CLASS_NAME);
+ // retrieve the actual message
+ String json = message.getText();
+
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Event message received from queue: %s", type));
+ }
+
+ try {
+ TopicSubscriberManager.acquireWriteLock();
+ processor1.process(type, json, TopicSubscriberManager.getTopology());
+ } finally {
+ TopicSubscriberManager.releaseWriteLock();
+ }
+
+ } catch (Exception e) {
+ String error = "Failed to retrieve the topology event message.";
+ log.error(error, e);
+ throw new RuntimeException(error, e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageReceiver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageReceiver.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageReceiver.java
new file mode 100644
index 0000000..ca5fc5b
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventMessageReceiver.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.stratos.autoscaler.message.receiver.topology;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.TextMessage;
+
+public class TopologyEventMessageReceiver implements MessageListener {
+
+ private static final Log log = LogFactory.getLog(TopologyEventMessageReceiver.class);
+
+ @Override
+ public void onMessage(Message message) {
+ if (message instanceof TextMessage) {
+ TextMessage receivedMessage = (TextMessage) message;
+ log.info(receivedMessage);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("Message received: " + ((TextMessage) message).getText());
+ }
+ // Add received message to the queue
+ TopologyEventQueue.getInstance().add(receivedMessage);
+
+ } catch (JMSException e) {
+ log.error(e.getMessage(), e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/e56c1ed2/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventQueue.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventQueue.java b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventQueue.java
new file mode 100644
index 0000000..2c0d022
--- /dev/null
+++ b/components/org.apache.stratos.autoscaler/src/main/java/org/apache/stratos/autoscaler/message/receiver/topology/TopologyEventQueue.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.stratos.autoscaler.message.receiver.topology;
+
+import javax.jms.TextMessage;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * Implements topology event queue.
+ */
+public class TopologyEventQueue extends LinkedBlockingQueue<TextMessage>{
+ private static volatile TopologyEventQueue instance;
+
+ private TopologyEventQueue(){
+ }
+
+ public static synchronized TopologyEventQueue getInstance() {
+ if (instance == null) {
+ synchronized (TopologyEventQueue.class){
+ if (instance == null) {
+ instance = new TopologyEventQueue ();
+ }
+ }
+ }
+ return instance;
+ }
+}