You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stratos.apache.org by im...@apache.org on 2015/01/09 18:04:39 UTC
[3/7] stratos git commit: Renaming applications package to
application in messaging component
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceActivatedProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceActivatedProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceActivatedProcessor.java
new file mode 100644
index 0000000..09625bb
--- /dev/null
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceActivatedProcessor.java
@@ -0,0 +1,120 @@
+/*
+ * 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.messaging.message.processor.application;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.messaging.domain.application.Application;
+import org.apache.stratos.messaging.domain.application.Applications;
+import org.apache.stratos.messaging.domain.application.Group;
+import org.apache.stratos.messaging.domain.application.GroupStatus;
+import org.apache.stratos.messaging.domain.instance.GroupInstance;
+import org.apache.stratos.messaging.event.application.GroupInstanceActivatedEvent;
+import org.apache.stratos.messaging.message.processor.MessageProcessor;
+import org.apache.stratos.messaging.message.processor.application.updater.ApplicationsUpdater;
+import org.apache.stratos.messaging.util.Util;
+
+/**
+ * This processor will act upon the Group activation events
+ */
+public class GroupInstanceActivatedProcessor extends MessageProcessor {
+ private static final Log log = LogFactory.getLog(GroupInstanceActivatedProcessor.class);
+ private MessageProcessor nextProcessor;
+
+ @Override
+ public void setNext(MessageProcessor nextProcessor) {
+ this.nextProcessor = nextProcessor;
+ }
+
+ @Override
+ public boolean process(String type, String message, Object object) {
+ Applications applications = (Applications) object;
+
+ if (GroupInstanceActivatedEvent.class.getName().equals(type)) {
+ // Return if applications has not been initialized
+ if (!applications.isInitialized())
+ return false;
+
+ // Parse complete message and build event
+ GroupInstanceActivatedEvent event = (GroupInstanceActivatedEvent) Util.
+ jsonToObject(message, GroupInstanceActivatedEvent.class);
+
+ ApplicationsUpdater.acquireWriteLockForApplication(event.getAppId());
+
+ try {
+ return doProcess(event, applications);
+
+ } finally {
+ ApplicationsUpdater.releaseWriteLockForApplication(event.getAppId());
+ }
+
+ } else {
+ if (nextProcessor != null) {
+ // ask the next processor to take care of the message.
+ return nextProcessor.process(type, message, applications);
+ } else {
+ throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
+ }
+ }
+ }
+
+ private boolean doProcess(GroupInstanceActivatedEvent event, Applications applications) {
+
+ // Validate event against the existing topology
+ Application application = applications.getApplication(event.getAppId());
+ if (application == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Application does not exist: [service] %s",
+ event.getAppId()));
+ }
+ return false;
+ }
+ Group group = application.getGroupRecursively(event.getGroupId());
+
+ if (group == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group not exists in service: [AppId] %s [groupId] %s", event.getAppId(),
+ event.getGroupId()));
+ return false;
+ }
+ } else {
+ GroupInstance context = group.getInstanceContexts(event.getInstanceId());
+ if(context == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group Instance not exists in Group: [AppId] %s [groupId] %s " +
+ "[instanceId] %s", event.getAppId(), event.getGroupId(),
+ event.getInstanceId()));
+ return false;
+ }
+ }
+ // Apply changes to the topology
+ GroupStatus status = GroupStatus.Active;
+ if (!context.isStateTransitionValid(status)) {
+ log.error("Invalid State Transition from " + context.getStatus() + " to " +
+ status);
+ return false;
+ }
+ context.setStatus(status);
+ }
+
+ // Notify event listeners
+ notifyEventListeners(event);
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceCreatedProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceCreatedProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceCreatedProcessor.java
new file mode 100644
index 0000000..aa44db6
--- /dev/null
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceCreatedProcessor.java
@@ -0,0 +1,115 @@
+/*
+ * 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.messaging.message.processor.application;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.messaging.domain.application.Application;
+import org.apache.stratos.messaging.domain.application.Applications;
+import org.apache.stratos.messaging.domain.application.Group;
+import org.apache.stratos.messaging.domain.instance.GroupInstance;
+import org.apache.stratos.messaging.event.application.GroupInstanceCreatedEvent;
+import org.apache.stratos.messaging.message.processor.MessageProcessor;
+import org.apache.stratos.messaging.message.processor.application.updater.ApplicationsUpdater;
+import org.apache.stratos.messaging.util.Util;
+
+/**
+ * This processor will act upon the Group activation events
+ */
+public class GroupInstanceCreatedProcessor extends MessageProcessor {
+ private static final Log log = LogFactory.getLog(GroupInstanceCreatedProcessor.class);
+ private MessageProcessor nextProcessor;
+
+ @Override
+ public void setNext(MessageProcessor nextProcessor) {
+ this.nextProcessor = nextProcessor;
+ }
+
+ @Override
+ public boolean process(String type, String message, Object object) {
+ Applications applications = (Applications) object;
+
+ if (GroupInstanceCreatedEvent.class.getName().equals(type)) {
+ // Return if applications has not been initialized
+ if (!applications.isInitialized())
+ return false;
+
+ // Parse complete message and build event
+ GroupInstanceCreatedEvent event = (GroupInstanceCreatedEvent) Util.
+ jsonToObject(message, GroupInstanceCreatedEvent.class);
+
+ ApplicationsUpdater.acquireWriteLockForApplication(event.getAppId());
+
+ try {
+ return doProcess(event, applications);
+
+ } finally {
+ ApplicationsUpdater.releaseWriteLockForApplication(event.getAppId());
+ }
+
+ } else {
+ if (nextProcessor != null) {
+ // ask the next processor to take care of the message.
+ return nextProcessor.process(type, message, applications);
+ } else {
+ throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
+ }
+ }
+ }
+
+ private boolean doProcess(GroupInstanceCreatedEvent event, Applications applications) {
+
+ // Validate event against the existing applications
+ Application application = applications.getApplication(event.getAppId());
+ if (application == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Application does not exist: [service] %s",
+ event.getAppId()));
+ }
+ return false;
+ }
+ Group group = application.getGroupRecursively(event.getGroupId());
+
+ if (group == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group not exists in service: [AppId] %s [groupId] %s", event.getAppId(),
+ event.getGroupId()));
+ return false;
+ }
+ } else {
+ // Apply changes to the applications
+ String instanceId = event.getGroupInstance().getInstanceId();
+ GroupInstance context = group.getInstanceContexts(instanceId);
+ if(context != null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group Instance already exists in Group: [AppId] %s [groupId] %s " +
+ "[instanceId] %s", event.getAppId(), event.getGroupId(),
+ instanceId));
+ return false;
+ }
+ }
+ // Apply changes to the topology
+ group.addInstance(instanceId, event.getGroupInstance());
+ }
+
+ // Notify event listeners
+ notifyEventListeners(event);
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceInactivateProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceInactivateProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceInactivateProcessor.java
new file mode 100644
index 0000000..041f6eb
--- /dev/null
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceInactivateProcessor.java
@@ -0,0 +1,120 @@
+/*
+ * 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.messaging.message.processor.application;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.messaging.domain.application.Application;
+import org.apache.stratos.messaging.domain.application.Applications;
+import org.apache.stratos.messaging.domain.application.Group;
+import org.apache.stratos.messaging.domain.application.GroupStatus;
+import org.apache.stratos.messaging.domain.instance.GroupInstance;
+import org.apache.stratos.messaging.event.application.GroupInstanceInactivatedEvent;
+import org.apache.stratos.messaging.message.processor.MessageProcessor;
+import org.apache.stratos.messaging.message.processor.application.updater.ApplicationsUpdater;
+import org.apache.stratos.messaging.util.Util;
+
+/**
+ * This processor will act upon the Group activation events
+ */
+public class GroupInstanceInactivateProcessor extends MessageProcessor {
+ private static final Log log = LogFactory.getLog(GroupInstanceInactivateProcessor.class);
+ private MessageProcessor nextProcessor;
+
+ @Override
+ public void setNext(MessageProcessor nextProcessor) {
+ this.nextProcessor = nextProcessor;
+ }
+
+ @Override
+ public boolean process(String type, String message, Object object) {
+ Applications applications = (Applications) object;
+
+ if (GroupInstanceInactivatedEvent.class.getName().equals(type)) {
+ // Return if applications has not been initialized
+ if (!applications.isInitialized())
+ return false;
+
+ // Parse complete message and build event
+ GroupInstanceInactivatedEvent event = (GroupInstanceInactivatedEvent) Util.
+ jsonToObject(message, GroupInstanceInactivatedEvent.class);
+
+ ApplicationsUpdater.acquireWriteLockForApplication(event.getAppId());
+
+ try {
+ return doProcess(event, applications);
+
+ } finally {
+ ApplicationsUpdater.releaseWriteLockForApplication(event.getAppId());
+ }
+
+ } else {
+ if (nextProcessor != null) {
+ // ask the next processor to take care of the message.
+ return nextProcessor.process(type, message, applications);
+ } else {
+ throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
+ }
+ }
+ }
+
+ private boolean doProcess(GroupInstanceInactivatedEvent event, Applications applications) {
+
+ // Validate event against the existing applications
+ Application application = applications.getApplication(event.getAppId());
+ if (application == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Application does not exist: [service] %s",
+ event.getAppId()));
+ }
+ return false;
+ }
+ Group group = application.getGroupRecursively(event.getGroupId());
+
+ if (group == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group not exists in service: [AppId] %s [groupId] %s", event.getAppId(),
+ event.getGroupId()));
+ return false;
+ }
+ } else {
+ GroupInstance context = group.getInstanceContexts(event.getInstanceId());
+ if(context == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group Instance not exists in Group: [AppId] %s [groupId] %s " +
+ "[instanceId] %s", event.getAppId(), event.getGroupId(),
+ event.getInstanceId()));
+ return false;
+ }
+ }
+ // Apply changes to the topology
+ GroupStatus status = GroupStatus.Inactive;
+ if (!context.isStateTransitionValid(status)) {
+ log.error("Invalid State Transition from " + context.getStatus() + " to " +
+ status);
+ return false;
+ }
+ context.setStatus(status);
+ }
+
+ // Notify event listeners
+ notifyEventListeners(event);
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceTerminatedProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceTerminatedProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceTerminatedProcessor.java
new file mode 100644
index 0000000..1f7605d
--- /dev/null
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceTerminatedProcessor.java
@@ -0,0 +1,121 @@
+/*
+ * 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.messaging.message.processor.application;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.messaging.domain.application.Application;
+import org.apache.stratos.messaging.domain.application.Applications;
+import org.apache.stratos.messaging.domain.application.Group;
+import org.apache.stratos.messaging.domain.application.GroupStatus;
+import org.apache.stratos.messaging.domain.instance.GroupInstance;
+import org.apache.stratos.messaging.event.application.GroupInstanceTerminatedEvent;
+import org.apache.stratos.messaging.message.processor.MessageProcessor;
+import org.apache.stratos.messaging.message.processor.application.updater.ApplicationsUpdater;
+import org.apache.stratos.messaging.util.Util;
+
+/**
+ * This processor will act upon the Group activation events
+ */
+public class GroupInstanceTerminatedProcessor extends MessageProcessor {
+ private static final Log log = LogFactory.getLog(GroupInstanceTerminatedProcessor.class);
+ private MessageProcessor nextProcessor;
+
+ @Override
+ public void setNext(MessageProcessor nextProcessor) {
+ this.nextProcessor = nextProcessor;
+ }
+
+ @Override
+ public boolean process(String type, String message, Object object) {
+ Applications applications = (Applications) object;
+
+ if (GroupInstanceTerminatedEvent.class.getName().equals(type)) {
+ // Return if applications has not been initialized
+ if (!applications.isInitialized())
+ return false;
+
+ // Parse complete message and build event
+ GroupInstanceTerminatedEvent event = (GroupInstanceTerminatedEvent) Util.
+ jsonToObject(message, GroupInstanceTerminatedEvent.class);
+
+ ApplicationsUpdater.acquireWriteLockForApplication(event.getAppId());
+
+ try {
+ return doProcess(event, applications);
+
+ } finally {
+ ApplicationsUpdater.releaseWriteLockForApplication(event.getAppId());
+ }
+
+ } else {
+ if (nextProcessor != null) {
+ // ask the next processor to take care of the message.
+ return nextProcessor.process(type, message, applications);
+ } else {
+ throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
+ }
+ }
+ }
+
+ private boolean doProcess(GroupInstanceTerminatedEvent event, Applications applications) {
+
+ // Validate event against the existing applications
+ Application application = applications.getApplication(event.getAppId());
+ if (application == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Application does not exist: [service] %s",
+ event.getAppId()));
+ }
+ return false;
+ }
+ Group group = application.getGroupRecursively(event.getGroupId());
+
+ if (group == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group not exists in service: [AppId] %s [groupId] %s", event.getAppId(),
+ event.getGroupId()));
+ return false;
+ }
+ } else {
+ GroupInstance context = group.getInstanceContexts(event.getInstanceId());
+ if(context == null) {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Group Instance alread removed in Group: [AppId] %s [groupId] %s " +
+ "[instanceId] %s", event.getAppId(), event.getGroupId(),
+ event.getInstanceId()));
+ }
+ } else {
+ // Apply changes to the topology
+ GroupStatus status = GroupStatus.Terminated;
+ if (!context.isStateTransitionValid(status)) {
+ log.error("Invalid State Transition from " + context.getStatus() + " to " +
+ status);
+ return false;
+ }
+ context.setStatus(status);
+ group.removeInstance(event.getInstanceId());
+ }
+ }
+
+ // Notify event listeners
+ notifyEventListeners(event);
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceTerminatingProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceTerminatingProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceTerminatingProcessor.java
new file mode 100644
index 0000000..2548340
--- /dev/null
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/GroupInstanceTerminatingProcessor.java
@@ -0,0 +1,122 @@
+/*
+ * 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.messaging.message.processor.application;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.messaging.domain.application.Application;
+import org.apache.stratos.messaging.domain.application.Applications;
+import org.apache.stratos.messaging.domain.application.Group;
+import org.apache.stratos.messaging.domain.application.GroupStatus;
+import org.apache.stratos.messaging.domain.instance.GroupInstance;
+import org.apache.stratos.messaging.event.application.GroupInstanceTerminatingEvent;
+import org.apache.stratos.messaging.message.processor.MessageProcessor;
+import org.apache.stratos.messaging.message.processor.application.updater.ApplicationsUpdater;
+import org.apache.stratos.messaging.util.Util;
+
+/**
+ * This processor will act upon the Group activation events
+ */
+public class GroupInstanceTerminatingProcessor extends MessageProcessor {
+ private static final Log log = LogFactory.getLog(GroupInstanceTerminatingProcessor.class);
+ private MessageProcessor nextProcessor;
+
+ @Override
+ public void setNext(MessageProcessor nextProcessor) {
+ this.nextProcessor = nextProcessor;
+ }
+
+ @Override
+ public boolean process(String type, String message, Object object) {
+ Applications applications = (Applications) object;
+
+ if (GroupInstanceTerminatingEvent.class.getName().equals(type)) {
+ // Return if applications has not been initialized
+ if (!applications.isInitialized())
+ return false;
+
+ // Parse complete message and build event
+ GroupInstanceTerminatingEvent event = (GroupInstanceTerminatingEvent) Util.
+ jsonToObject(message, GroupInstanceTerminatingEvent.class);
+
+ ApplicationsUpdater.acquireWriteLockForApplication(event.getAppId());
+
+ try {
+ return doProcess(event, applications);
+
+ } finally {
+ ApplicationsUpdater.releaseWriteLockForApplication(event.getAppId());
+ }
+
+ } else {
+ if (nextProcessor != null) {
+ // ask the next processor to take care of the message.
+ return nextProcessor.process(type, message, applications);
+ } else {
+ throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
+ }
+ }
+ }
+
+ private boolean doProcess(GroupInstanceTerminatingEvent event, Applications applications) {
+
+ // Validate event against the existing applications
+ Application application = applications.getApplication(event.getAppId());
+ if (application == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Application does not exist: [service] %s",
+ event.getAppId()));
+ }
+ return false;
+ }
+ Group group = application.getGroupRecursively(event.getGroupId());
+
+ if (group == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group not exists in service: [AppId] %s [groupId] %s", event.getAppId(),
+ event.getGroupId()));
+ return false;
+ }
+ } else {
+ // Apply changes to the applications
+ GroupInstance context = group.getInstanceContexts(event.getInstanceId());
+ if(context == null) {
+ if (log.isWarnEnabled()) {
+ log.warn(String.format("Group Instance not exists in Group: [AppId] %s [groupId] %s " +
+ "[instanceId] %s", event.getAppId(), event.getGroupId(),
+ event.getInstanceId()));
+ return false;
+ }
+ }
+ // Apply changes to the topology
+ GroupStatus status = GroupStatus.Terminating;
+ if (!context.isStateTransitionValid(status)) {
+ log.error("Invalid State Transition from " + context.getStatus() + " to " +
+ status);
+ return false;
+ }
+ context.setStatus(status);
+
+ }
+
+ // Notify event listeners
+ notifyEventListeners(event);
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/updater/ApplicationsUpdater.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/updater/ApplicationsUpdater.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/updater/ApplicationsUpdater.java
new file mode 100644
index 0000000..47cacd4
--- /dev/null
+++ b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/application/updater/ApplicationsUpdater.java
@@ -0,0 +1,138 @@
+/*
+ * 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.messaging.message.processor.application.updater;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.messaging.domain.application.locking.ApplicationLock;
+import org.apache.stratos.messaging.domain.application.locking.ApplicationLockHierarchy;
+import org.apache.stratos.messaging.message.receiver.application.ApplicationManager;
+
+/**
+ * Used to lock the Topology for writes by messaging component
+ * <p/>
+ * Acquire a write lock:
+ * <p/>
+ * From root level, acquire read lock, and acquire a write lock only for the
+ * relevant sub tree.
+ * <p/>
+ * Example 1: Acquiring write lock for a Cluster to modify the Cluster object -
+ * acquiring:
+ * public static void acquireWriteLockForCluster (String serviceName, String clusterId)
+ * <p/>
+ * releasing:
+ * public static void releaseWriteLockForCluster (String serviceName, String clusterId)
+ * <p/>
+ * Example 2: Acquiring write lock to add a new Cluster object -
+ * acquiring:
+ * public static void acquireWriteLockForService (String serviceName)
+ * <p/>
+ * releasing:
+ * public static void releaseWriteLockForService (String serviceName)
+ * <p/>
+ * Example 3: Acquiring the write lock to add a deploy a Cartridge (add a new Service)
+ * acquire:
+ * public static void acquireWriteLockForServices()
+ * <p/>
+ * release:
+ * public static void releaseWriteLockForServices()
+ */
+
+public class ApplicationsUpdater {
+
+ private static final Log log = LogFactory.getLog(ApplicationsUpdater.class);
+
+ private static volatile ApplicationLockHierarchy applicationLockHierarchy =
+ ApplicationLockHierarchy.getInstance();
+
+ // Top level locks - should be used to lock the entire Topology
+
+ /**
+ * Acquires write lock for all Applications
+ */
+ public static void acquireWriteLockForApplications() {
+ if (log.isDebugEnabled()) {
+ log.debug("Write lock acquired for Applications");
+ }
+ applicationLockHierarchy.getApplicationLock().acquireWriteLock();
+ }
+
+ /**
+ * Releases write lock for all Applications
+ */
+ public static void releaseWriteLockForApplications() {
+ if (log.isDebugEnabled()) {
+ log.debug("Write lock released for Applications");
+ }
+ applicationLockHierarchy.getApplicationLock().releaseWritelock();
+ }
+
+ /**
+ * Acquires write lock for the Application
+ *
+ * @param appId Application id
+ */
+ public static void acquireWriteLockForApplication(String appId) {
+
+ // acquire read lock for all Applications
+ ApplicationManager.acquireReadLockForApplications();
+
+ ApplicationLock applicationLock = applicationLockHierarchy.getLockForApplication(appId);
+ if (applicationLock == null) {
+ handleLockNotFound("Topology lock not found for Application " + appId);
+
+ } else {
+ // now, lock Application
+ applicationLock.acquireWriteLock();
+ if (log.isDebugEnabled()) {
+ log.debug("Write lock acquired for Application " + appId);
+ }
+ }
+ }
+
+ /**
+ * Releases write lock for the Application
+ *
+ * @param appId Application id
+ */
+ public static void releaseWriteLockForApplication(String appId) {
+
+ ApplicationLock applicationLock = applicationLockHierarchy.getLockForApplication(appId);
+ if (applicationLock == null) {
+ handleLockNotFound("Topology lock not found for Application " + appId);
+
+ } else {
+ // release App lock
+ applicationLock.releaseWritelock();
+ if (log.isDebugEnabled()) {
+ log.debug("Write lock released for Application " + appId);
+ }
+ }
+
+ // release read lock for all Applications
+ ApplicationManager.releaseReadLockForApplications();
+ }
+
+ private static void handleLockNotFound(String errorMsg) {
+ log.warn(errorMsg);
+ //throw new RuntimeException(errorMsg);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationCreatedMessageProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationCreatedMessageProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationCreatedMessageProcessor.java
deleted file mode 100644
index a945cd8..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationCreatedMessageProcessor.java
+++ /dev/null
@@ -1,105 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.applications.Applications;
-import org.apache.stratos.messaging.event.applications.ApplicationCreatedEvent;
-import org.apache.stratos.messaging.message.processor.MessageProcessor;
-import org.apache.stratos.messaging.message.processor.applications.updater.ApplicationsUpdater;
-import org.apache.stratos.messaging.util.Util;
-
-public class ApplicationCreatedMessageProcessor extends MessageProcessor {
-
- private static final Log log = LogFactory.getLog(ApplicationCreatedMessageProcessor.class);
- private MessageProcessor nextProcessor;
-
- @Override
- public void setNext(MessageProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
- @Override
- public boolean process(String type, String message, Object object) {
-
- Applications applications = (Applications) object;
-
- if (ApplicationCreatedEvent.class.getName().equals(type)) {
- if (!applications.isInitialized()) {
- return false;
- }
-
- ApplicationCreatedEvent event = (ApplicationCreatedEvent) Util.jsonToObject(message, ApplicationCreatedEvent.class);
- if (event == null) {
- log.error("Unable to convert the JSON message to ApplicationCreatedEvent");
- return false;
- }
-
- ApplicationsUpdater.acquireWriteLockForApplications();
- try {
- return doProcess(event, applications);
-
- } finally {
- ApplicationsUpdater.releaseWriteLockForApplications();
- }
-
- } else {
- if (nextProcessor != null) {
- // ask the next processor to take care of the message.
- return nextProcessor.process(type, message, applications);
- } else {
- throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
- }
- }
- }
-
- private boolean doProcess(ApplicationCreatedEvent event, Applications applications) {
-
- // check if required properties are available
- if (event.getApplication() == null) {
- String errorMsg = "Application object of application created event is invalid";
- log.error(errorMsg);
- throw new RuntimeException(errorMsg);
- }
-
- if (event.getApplication().getUniqueIdentifier() == null || event.getApplication().getUniqueIdentifier().isEmpty()) {
- String errorMsg = "App id of application created event is invalid: [ " + event.getApplication().getUniqueIdentifier() + " ]";
- log.error(errorMsg);
- throw new RuntimeException(errorMsg);
- }
-
- // check if an Application with same name exists in applications
- if (applications.applicationExists(event.getApplication().getUniqueIdentifier())) {
- if(log.isDebugEnabled()) {
- log.debug("Application with id [ " + event.getApplication().getUniqueIdentifier() + " ] already exists");
- }
- } else {
- // add application and the clusters to Topology
- applications.addApplication(event.getApplication());
- if(log.isInfoEnabled()) {
- log.info("Application with id [ " + event.getApplication().getUniqueIdentifier() + " ] created");
- }
- }
-
- notifyEventListeners(event);
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationDeletedMessageProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationDeletedMessageProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationDeletedMessageProcessor.java
deleted file mode 100644
index 51bf741..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationDeletedMessageProcessor.java
+++ /dev/null
@@ -1,94 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.applications.Applications;
-import org.apache.stratos.messaging.event.applications.ApplicationDeletedEvent;
-import org.apache.stratos.messaging.message.processor.MessageProcessor;
-import org.apache.stratos.messaging.message.processor.applications.updater.ApplicationsUpdater;
-import org.apache.stratos.messaging.util.Util;
-
-public class ApplicationDeletedMessageProcessor extends MessageProcessor {
-
- private static final Log log = LogFactory.getLog(ApplicationDeletedMessageProcessor.class);
- private MessageProcessor nextProcessor;
-
- @Override
- public void setNext(MessageProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
- @Override
- public boolean process(String type, String message, Object object) {
-
- Applications applications = (Applications) object;
-
- if (ApplicationDeletedEvent.class.getName().equals(type)) {
- if (!applications.isInitialized()) {
- return false;
- }
-
- ApplicationDeletedEvent event = (ApplicationDeletedEvent) Util.jsonToObject(message, ApplicationDeletedEvent.class);
- if (event == null) {
- log.error("Unable to convert the JSON message to ApplicationDeletedEvent");
- return false;
- }
-
- ApplicationsUpdater.acquireWriteLockForApplications();
- try {
- return doProcess(event, applications);
-
- } finally {
- ApplicationsUpdater.releaseWriteLockForApplications();
- }
-
- } else {
- if (nextProcessor != null) {
- // ask the next processor to take care of the message.
- return nextProcessor.process(type, message, applications);
- } else {
- throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
- }
- }
- }
-
- private boolean doProcess(ApplicationDeletedEvent event, Applications applications) {
-
- // check if required properties are available
- if (event.getAppId() == null || event.getAppId().isEmpty()) {
- String errorMsg = "App id of application deleted event is invalid: [ " + event.getAppId() + " ]";
- log.error(errorMsg);
- throw new RuntimeException(errorMsg);
- }
-
- // Remove application and clusters from topology
- applications.removeApplication(event.getAppId());
-
- notifyEventListeners(event);
-
- if (log.isInfoEnabled()) {
- log.info("[Application] " + event.getAppId() + " has been successfully removed");
- }
-
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceActivatedMessageProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceActivatedMessageProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceActivatedMessageProcessor.java
deleted file mode 100644
index bd83e10..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceActivatedMessageProcessor.java
+++ /dev/null
@@ -1,114 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.applications.Application;
-import org.apache.stratos.messaging.domain.applications.ApplicationStatus;
-import org.apache.stratos.messaging.domain.applications.Applications;
-import org.apache.stratos.messaging.domain.instance.ApplicationInstance;
-import org.apache.stratos.messaging.event.applications.ApplicationInstanceActivatedEvent;
-import org.apache.stratos.messaging.message.processor.MessageProcessor;
-import org.apache.stratos.messaging.message.processor.applications.updater.ApplicationsUpdater;
-import org.apache.stratos.messaging.util.Util;
-
-/**
- * This processor responsible to process the application activation even and update the Topology.
- */
-public class ApplicationInstanceActivatedMessageProcessor extends MessageProcessor {
- private static final Log log =
- LogFactory.getLog(ApplicationInstanceActivatedMessageProcessor.class);
-
-
- private MessageProcessor nextProcessor;
-
- @Override
- public void setNext(MessageProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
-
- @Override
- public boolean process(String type, String message, Object object) {
- Applications applications = (Applications) object;
-
- if (ApplicationInstanceActivatedEvent.class.getName().equals(type)) {
- // Return if applications has not been initialized
- if (!applications.isInitialized())
- return false;
-
- // Parse complete message and build event
- ApplicationInstanceActivatedEvent event = (ApplicationInstanceActivatedEvent) Util.
- jsonToObject(message, ApplicationInstanceActivatedEvent.class);
-
- ApplicationsUpdater.acquireWriteLockForApplication(event.getAppId());
-
- try {
- return doProcess(event, applications);
-
- } finally {
- ApplicationsUpdater.releaseWriteLockForApplication(event.getAppId());
- }
-
- } else {
- if (nextProcessor != null) {
- // ask the next processor to take care of the message.
- return nextProcessor.process(type, message, applications);
- } else {
- throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
- }
- }
- }
-
- private boolean doProcess(ApplicationInstanceActivatedEvent event, Applications applications) {
-
- // Validate event against the existing applications
- Application application = applications.getApplication(event.getAppId());
- if (application == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Application does not exist: [service] %s",
- event.getAppId()));
- }
- return false;
- } else {
- // Apply changes to the applications
- ApplicationInstance context = application.getInstanceContexts(event.getInstanceId());
- if(context == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Application Instance not exists in Group: [AppId] %s" +
- "[instanceId] %s", event.getAppId(), event.getInstanceId()));
- return false;
- }
- }
- ApplicationStatus status = ApplicationStatus.Active;
- if (!context.isStateTransitionValid(status)) {
- log.error("Invalid State transfer from [ " + context.getStatus() +
- " ] to [ " + status + " ]");
- }
- context.setStatus(status);
-
- }
-
- // Notify event listeners
- notifyEventListeners(event);
- return true;
-
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceCreatedMessageProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceCreatedMessageProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceCreatedMessageProcessor.java
deleted file mode 100644
index fe6a3f9..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceCreatedMessageProcessor.java
+++ /dev/null
@@ -1,111 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.applications.Applications;
-import org.apache.stratos.messaging.domain.instance.ApplicationInstance;
-import org.apache.stratos.messaging.event.applications.ApplicationInstanceCreatedEvent;
-import org.apache.stratos.messaging.message.processor.MessageProcessor;
-import org.apache.stratos.messaging.message.processor.applications.updater.ApplicationsUpdater;
-import org.apache.stratos.messaging.util.Util;
-
-public class ApplicationInstanceCreatedMessageProcessor extends MessageProcessor {
-
- private static final Log log = LogFactory.getLog(ApplicationInstanceCreatedMessageProcessor.class);
- private MessageProcessor nextProcessor;
-
- @Override
- public void setNext(MessageProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
- @Override
- public boolean process(String type, String message, Object object) {
-
- Applications applications = (Applications) object;
-
- if (ApplicationInstanceCreatedEvent.class.getName().equals(type)) {
-
- if (!applications.isInitialized()) {
- return false;
- }
-
-
- ApplicationInstanceCreatedEvent event = (ApplicationInstanceCreatedEvent) Util.jsonToObject(message,
- ApplicationInstanceCreatedEvent.class);
- if (event == null) {
- log.error("Unable to convert the JSON message to ApplicationInstanceCreatedEvent");
- return false;
- }
-
- ApplicationsUpdater.acquireWriteLockForApplications();
- try {
- return doProcess(event, applications);
-
- } finally {
- ApplicationsUpdater.releaseWriteLockForApplications();
- }
-
- } else {
- if (nextProcessor != null) {
- // ask the next processor to take care of the message.
- return nextProcessor.process(type, message, applications);
- } else {
- throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
- }
- }
- }
-
- private boolean doProcess(ApplicationInstanceCreatedEvent event, Applications applications) {
-
- // check if required properties are available
- if (event.getApplicationInstance() == null) {
- String errorMsg = "Application instance object of application instance created event is invalid";
- log.error(errorMsg);
- throw new RuntimeException(errorMsg);
-
- }
-
- ApplicationInstance applicationInstance = event.getApplicationInstance();
-
- if (applicationInstance.getInstanceId() == null || applicationInstance.getInstanceId().isEmpty()) {
- String errorMsg = "App instance id of application instance created event is invalid: [ "
- + applicationInstance.getInstanceId() + " ]";
- log.error(errorMsg);
- throw new RuntimeException(errorMsg);
- }
-
- // check if an Application instance with same name exists in applications instance
- if (null != applications.getApplication(event.getApplicationId()).
- getInstanceByNetworkPartitionId(applicationInstance.getNetworkPartitionId())) {
-
- log.warn("Application instance with id [ " + applicationInstance.getInstanceId() + " ] already exists");
-
- } else {
- // add application instance to Application Topology
- applications.getApplication(event.getApplicationId()).addInstance(applicationInstance.getInstanceId(), applicationInstance);
- }
-
- notifyEventListeners(event);
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceInactivatedMessageProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceInactivatedMessageProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceInactivatedMessageProcessor.java
deleted file mode 100644
index e41905b..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceInactivatedMessageProcessor.java
+++ /dev/null
@@ -1,114 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.applications.Application;
-import org.apache.stratos.messaging.domain.applications.ApplicationStatus;
-import org.apache.stratos.messaging.domain.applications.Applications;
-import org.apache.stratos.messaging.domain.instance.ApplicationInstance;
-import org.apache.stratos.messaging.event.applications.ApplicationInstanceInactivatedEvent;
-import org.apache.stratos.messaging.message.processor.MessageProcessor;
-import org.apache.stratos.messaging.message.processor.applications.updater.ApplicationsUpdater;
-import org.apache.stratos.messaging.util.Util;
-
-/**
- * This processor responsible to process the application Inactivation even and update the Topology.
- */
-public class ApplicationInstanceInactivatedMessageProcessor extends MessageProcessor {
- private static final Log log =
- LogFactory.getLog(ApplicationInstanceInactivatedMessageProcessor.class);
-
-
- private MessageProcessor nextProcessor;
-
- @Override
- public void setNext(MessageProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
-
- @Override
- public boolean process(String type, String message, Object object) {
- Applications applications = (Applications) object;
-
- if (ApplicationInstanceInactivatedEvent.class.getName().equals(type)) {
- // Return if applications has not been initialized
- if (!applications.isInitialized())
- return false;
-
- // Parse complete message and build event
- ApplicationInstanceInactivatedEvent event = (ApplicationInstanceInactivatedEvent) Util.
- jsonToObject(message, ApplicationInstanceInactivatedEvent.class);
-
- ApplicationsUpdater.acquireWriteLockForApplication(event.getAppId());
-
- try {
- return doProcess(event, applications);
-
- } finally {
- ApplicationsUpdater.releaseWriteLockForApplication(event.getAppId());
- }
-
- } else {
- if (nextProcessor != null) {
- // ask the next processor to take care of the message.
- return nextProcessor.process(type, message, applications);
- } else {
- throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
- }
- }
- }
-
- private boolean doProcess(ApplicationInstanceInactivatedEvent event, Applications applications) {
-
- // Validate event against the existing applications
- Application application = applications.getApplication(event.getAppId());
- if (application == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Application does not exist: [service] %s",
- event.getAppId()));
- }
- return false;
- } else {
- // Apply changes to the applications
- ApplicationInstance context = application.getInstanceContexts(event.getInstanceId());
- if(context == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Application Instance not exists in Group: [AppId] %s" +
- "[instanceId] %s", event.getAppId(), event.getInstanceId()));
- return false;
- }
- }
- ApplicationStatus status = ApplicationStatus.Inactive;
- if (!context.isStateTransitionValid(status)) {
- log.error("Invalid State transfer from [ " + context.getStatus() +
- " ] to [ " + status + " ]");
- }
- context.setStatus(status);
-
- }
-
- // Notify event listeners
- notifyEventListeners(event);
- return true;
-
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceTerminatedMessageProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceTerminatedMessageProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceTerminatedMessageProcessor.java
deleted file mode 100644
index 5479050..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceTerminatedMessageProcessor.java
+++ /dev/null
@@ -1,110 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.applications.ApplicationStatus;
-import org.apache.stratos.messaging.domain.applications.Applications;
-import org.apache.stratos.messaging.domain.instance.ApplicationInstance;
-import org.apache.stratos.messaging.event.applications.ApplicationInstanceTerminatedEvent;
-import org.apache.stratos.messaging.message.processor.MessageProcessor;
-import org.apache.stratos.messaging.message.processor.applications.updater.ApplicationsUpdater;
-import org.apache.stratos.messaging.util.Util;
-
-/**
- * This processor responsible to process the application Inactivation even and update the Topology.
- */
-public class ApplicationInstanceTerminatedMessageProcessor extends MessageProcessor {
- private static final Log log =
- LogFactory.getLog(ApplicationInstanceTerminatedMessageProcessor.class);
-
-
- private MessageProcessor nextProcessor;
-
- @Override
- public void setNext(MessageProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
-
- @Override
- public boolean process(String type, String message, Object object) {
- Applications applications = (Applications) object;
-
- if (ApplicationInstanceTerminatedEvent.class.getName().equals(type)) {
- // Return if applications has not been initialized
- if (!applications.isInitialized())
- return false;
-
- // Parse complete message and build event
- ApplicationInstanceTerminatedEvent event = (ApplicationInstanceTerminatedEvent) Util.
- jsonToObject(message, ApplicationInstanceTerminatedEvent.class);
-
- ApplicationsUpdater.acquireWriteLockForApplications();
-
- try {
- return doProcess(event, applications);
-
- } finally {
- ApplicationsUpdater.releaseWriteLockForApplications();
- }
-
- } else {
- if (nextProcessor != null) {
- // ask the next processor to take care of the message.
- return nextProcessor.process(type, message, applications);
- } else {
- throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
- }
- }
- }
-
- private boolean doProcess(ApplicationInstanceTerminatedEvent event, Applications applications) {
-
- // check if required properties are available
- if (event.getAppId() == null) {
- String errorMsg = "Application Id of application removed event is invalid";
- log.error(errorMsg);
- throw new RuntimeException(errorMsg);
- }
-
- // check if an Application with same name exists in applications
- String appId = event.getAppId();
- String instanceId = event.getInstanceId();
- if (applications.applicationExists(appId)) {
- log.warn("Application with id [ " + appId + " ] still exists in Applications, removing it");
- ApplicationInstance instance = applications.getApplication(appId).
- getInstanceContexts(instanceId);
- if(instance == null) {
- if(log.isDebugEnabled()) {
- log.debug("Application [Instance] " + instanceId + " has already been removed");
- }
- } else {
- instance.setStatus(ApplicationStatus.Terminated);
- applications.getApplication(appId).removeInstance(instanceId);
- }
-
- }
-
- notifyEventListeners(event);
- return true;
-
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceTerminatingMessageProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceTerminatingMessageProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceTerminatingMessageProcessor.java
deleted file mode 100644
index 7679091..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationInstanceTerminatingMessageProcessor.java
+++ /dev/null
@@ -1,113 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.applications.Application;
-import org.apache.stratos.messaging.domain.applications.ApplicationStatus;
-import org.apache.stratos.messaging.domain.applications.Applications;
-import org.apache.stratos.messaging.domain.instance.ApplicationInstance;
-import org.apache.stratos.messaging.event.applications.ApplicationInstanceTerminatingEvent;
-import org.apache.stratos.messaging.message.processor.MessageProcessor;
-import org.apache.stratos.messaging.message.processor.applications.updater.ApplicationsUpdater;
-import org.apache.stratos.messaging.util.Util;
-
-/**
- * This processor responsible to process the application Inactivation even and update the Topology.
- */
-public class ApplicationInstanceTerminatingMessageProcessor extends MessageProcessor {
- private static final Log log =
- LogFactory.getLog(ApplicationInstanceTerminatingMessageProcessor.class);
-
-
- private MessageProcessor nextProcessor;
-
- @Override
- public void setNext(MessageProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
-
- @Override
- public boolean process(String type, String message, Object object) {
- Applications applications = (Applications) object;
-
- if (ApplicationInstanceTerminatingEvent.class.getName().equals(type)) {
- // Return if applications has not been initialized
- if (!applications.isInitialized())
- return false;
-
- // Parse complete message and build event
- ApplicationInstanceTerminatingEvent event = (ApplicationInstanceTerminatingEvent) Util.
- jsonToObject(message, ApplicationInstanceTerminatingEvent.class);
-
- ApplicationsUpdater.acquireWriteLockForApplication(event.getAppId());
-
- try {
- return doProcess(event, applications);
-
- } finally {
- ApplicationsUpdater.releaseWriteLockForApplication(event.getAppId());
- }
-
- } else {
- if (nextProcessor != null) {
- // ask the next processor to take care of the message.
- return nextProcessor.process(type, message, applications);
- } else {
- throw new RuntimeException(String.format("Failed to process message using available message processors: [type] %s [body] %s", type, message));
- }
- }
- }
-
- private boolean doProcess(ApplicationInstanceTerminatingEvent event, Applications applications) {
-
- // Validate event against the existing applications
- Application application = applications.getApplication(event.getAppId());
- if (application == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Application does not exist: [service] %s",
- event.getAppId()));
- }
- return false;
- } else {
- // Apply changes to the applications
- ApplicationInstance context = application.getInstanceContexts(event.getInstanceId());
- if(context == null) {
- if (log.isWarnEnabled()) {
- log.warn(String.format("Application Instance not exists in Group: [AppId] %s" +
- "[instanceId] %s", event.getAppId(), event.getInstanceId()));
- return false;
- }
- }
- ApplicationStatus status = ApplicationStatus.Terminating;
- if (!context.isStateTransitionValid(status)) {
- log.error("Invalid State transfer from [ " + context.getStatus() +
- " ] to [ " + status + " ]");
- }
- context.setStatus(status);
- }
-
- // Notify event listeners
- notifyEventListeners(event);
- return true;
-
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationsMessageProcessorChain.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationsMessageProcessorChain.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationsMessageProcessorChain.java
deleted file mode 100644
index 2318c13..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/ApplicationsMessageProcessorChain.java
+++ /dev/null
@@ -1,127 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.listener.EventListener;
-import org.apache.stratos.messaging.listener.applications.*;
-import org.apache.stratos.messaging.message.processor.MessageProcessorChain;
-
-/**
- * Application Status processor chain is to handle the list processors to parse the application
- * status.
- */
-public class ApplicationsMessageProcessorChain extends MessageProcessorChain {
- private static final Log log = LogFactory.getLog(ApplicationsMessageProcessorChain.class);
-
- private GroupInstanceCreatedProcessor groupCreatedMessageProcessor;
- private GroupInstanceActivatedProcessor groupActivatedMessageProcessor;
- private GroupInstanceInactivateProcessor groupInactivateMessageProcessor;
- private GroupInstanceTerminatedProcessor groupTerminatedProcessor;
- private GroupInstanceTerminatingProcessor groupTerminatingProcessor;
- private ApplicationInstanceCreatedMessageProcessor applicationInstanceCreatedMessageProcessor;
- private ApplicationInstanceActivatedMessageProcessor applicationActivatedMessageProcessor;
- private ApplicationCreatedMessageProcessor applicationCreatedMessageProcessor;
- private ApplicationDeletedMessageProcessor applicationDeletedMessageProcessor;
- private ApplicationInstanceInactivatedMessageProcessor applicationInactivatedMessageProcessor;
- private ApplicationInstanceTerminatedMessageProcessor applicationTerminatedMessageProcessor;
- private ApplicationInstanceTerminatingMessageProcessor applicationTerminatingMessageProcessor;
- private CompleteApplicationsMessageProcessor completeApplicationsMessageProcessor;
-
- public void initialize() {
- // Add instance notifier event processors
-
- groupCreatedMessageProcessor = new GroupInstanceCreatedProcessor();
- add(groupCreatedMessageProcessor);
-
- groupActivatedMessageProcessor = new GroupInstanceActivatedProcessor();
- add(groupActivatedMessageProcessor);
-
- groupInactivateMessageProcessor = new GroupInstanceInactivateProcessor();
- add(groupInactivateMessageProcessor);
-
- groupTerminatedProcessor = new GroupInstanceTerminatedProcessor();
- add(groupTerminatedProcessor);
-
- groupTerminatingProcessor = new GroupInstanceTerminatingProcessor();
- add(groupTerminatingProcessor);
-
- applicationInstanceCreatedMessageProcessor = new ApplicationInstanceCreatedMessageProcessor();
- add(applicationInstanceCreatedMessageProcessor);
-
- applicationActivatedMessageProcessor = new ApplicationInstanceActivatedMessageProcessor();
- add(applicationActivatedMessageProcessor);
-
- applicationCreatedMessageProcessor = new ApplicationCreatedMessageProcessor();
- add(applicationCreatedMessageProcessor);
-
- applicationDeletedMessageProcessor = new ApplicationDeletedMessageProcessor();
- add(applicationDeletedMessageProcessor);
-
- applicationInactivatedMessageProcessor = new ApplicationInstanceInactivatedMessageProcessor();
- add(applicationInactivatedMessageProcessor);
-
- applicationTerminatingMessageProcessor = new ApplicationInstanceTerminatingMessageProcessor();
- add(applicationTerminatingMessageProcessor);
-
- completeApplicationsMessageProcessor = new CompleteApplicationsMessageProcessor();
- add(completeApplicationsMessageProcessor);
-
- applicationTerminatedMessageProcessor = new ApplicationInstanceTerminatedMessageProcessor();
- add(applicationTerminatedMessageProcessor);
-
- if (log.isDebugEnabled()) {
- log.debug("Instance notifier message processor chain initialized");
- }
- }
-
- public void addEventListener(EventListener eventListener) {
-
- if (eventListener instanceof GroupInstanceCreatedEventListener) {
- groupCreatedMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof GroupInstanceInactivateEventListener) {
- groupInactivateMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof GroupInstanceActivatedEventListener) {
- groupActivatedMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof GroupInstanceTerminatingEventListener) {
- groupTerminatingProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof GroupInstanceTerminatedEventListener) {
- groupTerminatedProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof ApplicationInstanceCreatedEventListener) {
- applicationInstanceCreatedMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof ApplicationCreatedEventListener) {
- applicationCreatedMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof ApplicationDeletedEventListener) {
- applicationDeletedMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof ApplicationInstanceActivatedEventListener) {
- applicationActivatedMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof ApplicationInstanceInactivatedEventListener) {
- applicationInactivatedMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof ApplicationInstanceTerminatingEventListener) {
- applicationTerminatingMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof ApplicationInstanceTerminatedEventListener) {
- applicationTerminatedMessageProcessor.addEventListener(eventListener);
- } else if (eventListener instanceof CompleteApplicationsEventListener) {
- completeApplicationsMessageProcessor.addEventListener(eventListener);
- } else {
- throw new RuntimeException("Unknown event listener " + eventListener.toString());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/c65438a3/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/CompleteApplicationsMessageProcessor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/CompleteApplicationsMessageProcessor.java b/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/CompleteApplicationsMessageProcessor.java
deleted file mode 100644
index bd9d5f1..0000000
--- a/components/org.apache.stratos.messaging/src/main/java/org/apache/stratos/messaging/message/processor/applications/CompleteApplicationsMessageProcessor.java
+++ /dev/null
@@ -1,94 +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.messaging.message.processor.applications;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.messaging.domain.applications.Application;
-import org.apache.stratos.messaging.domain.applications.Applications;
-import org.apache.stratos.messaging.event.applications.CompleteApplicationsEvent;
-import org.apache.stratos.messaging.message.processor.MessageProcessor;
-import org.apache.stratos.messaging.message.processor.applications.updater.ApplicationsUpdater;
-import org.apache.stratos.messaging.util.Util;
-
-import java.util.Collection;
-
-public class CompleteApplicationsMessageProcessor extends MessageProcessor {
-
- private static final Log log = LogFactory.getLog(CompleteApplicationsMessageProcessor.class);
- private MessageProcessor nextProcessor;
-
- @Override
- public void setNext(MessageProcessor nextProcessor) {
- this.nextProcessor = nextProcessor;
- }
-
- @Override
- public boolean process(String type, String message, Object object) {
- Applications applications = (Applications) object;
-
- if (CompleteApplicationsEvent.class.getName().equals(type)) {
- // Parse complete message and build event
- CompleteApplicationsEvent event = (CompleteApplicationsEvent) Util.
- jsonToObject(message, CompleteApplicationsEvent.class);
-
- if (!applications.isInitialized()) {
- ApplicationsUpdater.acquireWriteLockForApplications();
-
- try {
- doProcess(event, applications);
-
- } finally {
- ApplicationsUpdater.releaseWriteLockForApplications();
- }
- }
-
- // Notify event listeners
- notifyEventListeners(event);
- return true;
-
- } else {
- if (nextProcessor != null) {
- // ask the next processor to take care of the message.
- return nextProcessor.process(type, message, applications);
- }
- return false;
- }
- }
-
- private void doProcess(CompleteApplicationsEvent event, Applications applications) {
- // add existing Applications to Topology
- Collection<Application> applicationsList = event.getApplications().getApplications().values();
- if (applicationsList != null && !applicationsList.isEmpty()) {
- for (Application application : applicationsList) {
- applications.addApplication(application);
- if (log.isDebugEnabled()) {
- log.debug("Application with id [ " + application.getUniqueIdentifier() + " ] added to Applications");
- }
- }
- } else {
- if (log.isDebugEnabled()) {
- log.debug("No Application information found in Complete Applications event");
- }
- }
-
- // Set topology initialized
- applications.setInitialized(true);
- }
-}