Skip to content
This repository has been archived by the owner on May 12, 2021. It is now read-only.

Messaging model changes and MQTT integration #50

Closed
wants to merge 7 commits into from
Closed

Messaging model changes and MQTT integration #50

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d8f8ebd
messaging model changes and MQTT updates
Sep 22, 2014
2150658
messaging model changes and MQTT updates
Sep 22, 2014
e41b437
updates
Sep 22, 2014
d7cb133
revert the code formatting
Sep 23, 2014
e953c73
revert the code formatting
Sep 23, 2014
53b8a18
revert the code formatting
Sep 23, 2014
25bd46a
puppet master update with python agent
Sep 23, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
325 changes: 164 additions & 161 deletions ...java/org/apache/stratos/cartridge/agent/event/publisher/CartridgeAgentEventPublisher.java
View file
Open in desktop
@@ -1,25 +1,24 @@
package org.apache.stratos.cartridge.agent.event.publisher;

/*
*
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* 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
*
* 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
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/

*/

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
Expand All @@ -32,159 +31,163 @@
import org.apache.stratos.messaging.event.instance.status.InstanceReadyToShutdownEvent;
import org.apache.stratos.messaging.event.instance.status.InstanceStartedEvent;
import org.apache.stratos.messaging.util.Constants;
import org.apache.stratos.messaging.util.Util;

/**
* Cartridge agent event publisher.
*/
public class CartridgeAgentEventPublisher {
private static final Log log = LogFactory.getLog(CartridgeAgentEventPublisher.class);
private static boolean started;
private static boolean activated;
private static boolean readyToShutdown;
private static boolean maintenance;

public static void publishInstanceStartedEvent() {
if (!isStarted()) {
if (log.isInfoEnabled()) {
log.info("Publishing instance started event");
}
InstanceStartedEvent event = new InstanceStartedEvent(
CartridgeAgentConfiguration.getInstance().getServiceName(),
CartridgeAgentConfiguration.getInstance().getClusterId(),
CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
CartridgeAgentConfiguration.getInstance().getPartitionId(),
CartridgeAgentConfiguration.getInstance().getMemberId());

EventPublisher eventPublisher = EventPublisherPool.getPublisher(Constants.INSTANCE_STATUS_TOPIC);
eventPublisher.publish(event);
setStarted(true);
if (log.isInfoEnabled()) {
log.info("Instance started event published");
}

} else {
if (log.isWarnEnabled()) {
log.warn("Instance already started");
}
}
}

public static void publishInstanceActivatedEvent() {
if (!isActivated()) {
if (log.isInfoEnabled()) {
log.info("Publishing instance activated event");
}
InstanceActivatedEvent event = new InstanceActivatedEvent(
CartridgeAgentConfiguration.getInstance().getServiceName(),
CartridgeAgentConfiguration.getInstance().getClusterId(),
CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
CartridgeAgentConfiguration.getInstance().getPartitionId(),
CartridgeAgentConfiguration.getInstance().getMemberId()
);

// Event publisher connection will
EventPublisher eventPublisher = EventPublisherPool.getPublisher(Constants.INSTANCE_STATUS_TOPIC);
eventPublisher.publish(event);
if (log.isInfoEnabled()) {
log.info("Instance activated event published");
}

if (log.isInfoEnabled()) {
log.info("Starting health statistics notifier");
}
Thread thread = new Thread(new HealthStatisticsNotifier());
thread.start();
setActivated(true);
if (log.isInfoEnabled()) {
log.info("Health statistics notifier started");
}
} else {
if (log.isWarnEnabled()) {
log.warn("Instance already activated");
}
}
}

public static void publishInstanceReadyToShutdownEvent() {
if (!isReadyToShutdown()) {
if (log.isInfoEnabled()) {
log.info("Publishing instance activated event");
}
InstanceReadyToShutdownEvent event = new InstanceReadyToShutdownEvent(
CartridgeAgentConfiguration.getInstance().getServiceName(),
CartridgeAgentConfiguration.getInstance().getClusterId(),
CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
CartridgeAgentConfiguration.getInstance().getPartitionId(),
CartridgeAgentConfiguration.getInstance().getMemberId());

EventPublisher eventPublisher = EventPublisherPool.getPublisher(Constants.INSTANCE_STATUS_TOPIC);
eventPublisher.publish(event);
setReadyToShutdown(true);
if (log.isInfoEnabled()) {
log.info("Instance ReadyToShutDown event published");
}
} else {
if (log.isWarnEnabled()) {
log.warn("Instance already sent ReadyToShutDown event....");
}
}
}

public static void publishMaintenanceModeEvent() {
if (!isMaintenance()) {
if (log.isInfoEnabled()) {
log.info("Publishing instance maintenance mode event");
}
InstanceMaintenanceModeEvent event = new InstanceMaintenanceModeEvent(
CartridgeAgentConfiguration.getInstance().getServiceName(),
CartridgeAgentConfiguration.getInstance().getClusterId(),
CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
CartridgeAgentConfiguration.getInstance().getPartitionId(),
CartridgeAgentConfiguration.getInstance().getMemberId());

EventPublisher eventPublisher = EventPublisherPool.getPublisher(Constants.INSTANCE_STATUS_TOPIC);
eventPublisher.publish(event);
setMaintenance(true);
if (log.isInfoEnabled()) {
log.info("Instance Maintenance mode event published");
}
} else {
if (log.isWarnEnabled()) {
log.warn("Instance already in a Maintenance mode....");
}
}
}

public static boolean isStarted() {
return started;
}

public static void setStarted(boolean started) {
CartridgeAgentEventPublisher.started = started;
}

public static boolean isActivated() {
return activated;
}

public static void setActivated(boolean activated) {
CartridgeAgentEventPublisher.activated = activated;
}

public static boolean isReadyToShutdown() {
return readyToShutdown;
}

public static void setReadyToShutdown(boolean readyToShutdown) {
CartridgeAgentEventPublisher.readyToShutdown = readyToShutdown;
}

public static boolean isMaintenance() {
return maintenance;
}

public static void setMaintenance(boolean maintenance) {
CartridgeAgentEventPublisher.maintenance = maintenance;
}
private static final Log log = LogFactory
.getLog(CartridgeAgentEventPublisher.class);
private static boolean started;
private static boolean activated;
private static boolean readyToShutdown;
private static boolean maintenance;

public static void publishInstanceStartedEvent() {
if (!isStarted()) {
if (log.isInfoEnabled()) {
log.info("Publishing instance started event");
}
InstanceStartedEvent event = new InstanceStartedEvent(
CartridgeAgentConfiguration.getInstance().getServiceName(),
CartridgeAgentConfiguration.getInstance().getClusterId(),
CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
CartridgeAgentConfiguration.getInstance().getPartitionId(),
CartridgeAgentConfiguration.getInstance().getMemberId());

EventPublisher eventPublisher = EventPublisherPool
.getPublisher(Constants.INSTANCE_STATUS_TOPIC);
eventPublisher.publish(event);
setStarted(true);
if (log.isInfoEnabled()) {
log.info("Instance started event published");
}

} else {
if (log.isWarnEnabled()) {
log.warn("Instance already started");
}
}
}

public static void publishInstanceActivatedEvent() {
if (!isActivated()) {
if (log.isInfoEnabled()) {
log.info("Publishing instance activated event");
}
InstanceActivatedEvent event = new InstanceActivatedEvent(
CartridgeAgentConfiguration.getInstance().getServiceName(),
CartridgeAgentConfiguration.getInstance().getClusterId(),
CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
CartridgeAgentConfiguration.getInstance().getPartitionId(),
CartridgeAgentConfiguration.getInstance().getMemberId());

// Event publisher connection will
EventPublisher eventPublisher = EventPublisherPool
.getPublisher(Constants.INSTANCE_STATUS_TOPIC);
eventPublisher.publish(event);
if (log.isInfoEnabled()) {
log.info("Instance activated event published");
}

if (log.isInfoEnabled()) {
log.info("Starting health statistics notifier");
}
Thread thread = new Thread(new HealthStatisticsNotifier());
thread.start();
setActivated(true);
if (log.isInfoEnabled()) {
log.info("Health statistics notifier started");
}
} else {
if (log.isWarnEnabled()) {
log.warn("Instance already activated");
}
}
}

public static void publishInstanceReadyToShutdownEvent() {
if (!isReadyToShutdown()) {
if (log.isInfoEnabled()) {
log.info("Publishing instance activated event");
}
InstanceReadyToShutdownEvent event = new InstanceReadyToShutdownEvent(
CartridgeAgentConfiguration.getInstance().getServiceName(),
CartridgeAgentConfiguration.getInstance().getClusterId(),
CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
CartridgeAgentConfiguration.getInstance().getPartitionId(),
CartridgeAgentConfiguration.getInstance().getMemberId());

EventPublisher eventPublisher = EventPublisherPool
.getPublisher(Util.getMessageTopicName(event));
eventPublisher.publish(event);
setReadyToShutdown(true);
if (log.isInfoEnabled()) {
log.info("Instance ReadyToShutDown event published");
}
} else {
if (log.isWarnEnabled()) {
log.warn("Instance already sent ReadyToShutDown event....");
}
}
}

public static void publishMaintenanceModeEvent() {
if (!isMaintenance()) {
if (log.isInfoEnabled()) {
log.info("Publishing instance maintenance mode event");
}
InstanceMaintenanceModeEvent event = new InstanceMaintenanceModeEvent(
CartridgeAgentConfiguration.getInstance().getServiceName(),
CartridgeAgentConfiguration.getInstance().getClusterId(),
CartridgeAgentConfiguration.getInstance().getNetworkPartitionId(),
CartridgeAgentConfiguration.getInstance().getPartitionId(),
CartridgeAgentConfiguration.getInstance().getMemberId());

EventPublisher eventPublisher = EventPublisherPool
.getPublisher(Constants.INSTANCE_STATUS_TOPIC);
eventPublisher.publish(event);
setMaintenance(true);
if (log.isInfoEnabled()) {
log.info("Instance Maintenance mode event published");
}
} else {
if (log.isWarnEnabled()) {
log.warn("Instance already in a Maintenance mode....");
}
}
}

public static boolean isStarted() {
return started;
}

public static void setStarted(boolean started) {
CartridgeAgentEventPublisher.started = started;
}

public static boolean isActivated() {
return activated;
}

public static void setActivated(boolean activated) {
CartridgeAgentEventPublisher.activated = activated;
}

public static boolean isReadyToShutdown() {
return readyToShutdown;
}

public static void setReadyToShutdown(boolean readyToShutdown) {
CartridgeAgentEventPublisher.readyToShutdown = readyToShutdown;
}

public static boolean isMaintenance() {
return maintenance;
}

public static void setMaintenance(boolean maintenance) {
CartridgeAgentEventPublisher.maintenance = maintenance;
}
}