/*
 * 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.geode.management.internal.beans;

import static java.util.Arrays.asList;
import static java.util.Collections.unmodifiableList;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.geode.management.JMXNotificationType.ASYNC_EVENT_QUEUE_CLOSED;
import static org.apache.geode.management.JMXNotificationType.ASYNC_EVENT_QUEUE_CREATED;
import static org.apache.geode.management.JMXNotificationType.CACHE_SERVER_STARTED;
import static org.apache.geode.management.JMXNotificationType.CACHE_SERVER_STOPPED;
import static org.apache.geode.management.JMXNotificationType.CACHE_SERVICE_CREATED;
import static org.apache.geode.management.JMXNotificationType.CLIENT_CRASHED;
import static org.apache.geode.management.JMXNotificationType.CLIENT_JOINED;
import static org.apache.geode.management.JMXNotificationType.CLIENT_LEFT;
import static org.apache.geode.management.JMXNotificationType.DISK_STORE_CLOSED;
import static org.apache.geode.management.JMXNotificationType.DISK_STORE_CREATED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_RECEIVER_CREATED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_RECEIVER_DESTROYED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_RECEIVER_STARTED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_RECEIVER_STOPPED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_SENDER_CREATED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_SENDER_PAUSED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_SENDER_REMOVED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_SENDER_RESUMED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_SENDER_STARTED;
import static org.apache.geode.management.JMXNotificationType.GATEWAY_SENDER_STOPPED;
import static org.apache.geode.management.JMXNotificationType.LOCATOR_STARTED;
import static org.apache.geode.management.JMXNotificationType.LOCK_SERVICE_CLOSED;
import static org.apache.geode.management.JMXNotificationType.LOCK_SERVICE_CREATED;
import static org.apache.geode.management.JMXNotificationType.MANAGER_STARTED;
import static org.apache.geode.management.JMXNotificationType.REGION_CLOSED;
import static org.apache.geode.management.JMXNotificationType.REGION_CREATED;
import static org.apache.geode.management.JMXNotificationType.SYSTEM_ALERT;
import static org.apache.geode.management.JMXNotificationUserData.ALERT_LEVEL;
import static org.apache.geode.management.JMXNotificationUserData.MEMBER;
import static org.apache.geode.management.JMXNotificationUserData.THREAD;
import static org.apache.geode.management.internal.AlertDetails.getAlertLevelAsString;
import static org.apache.geode.management.internal.ManagementConstants.AGGREGATE_MBEAN_PATTERN;
import static org.apache.geode.management.internal.ManagementConstants.ASYNC_EVENT_QUEUE_CLOSED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.ASYNC_EVENT_QUEUE_CREATED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.CACHE_SERVER_STARTED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.CACHE_SERVER_STOPPED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.CACHE_SERVICE_CREATED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.CLIENT_CRASHED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.CLIENT_JOINED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.CLIENT_LEFT_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.DISK_STORE_CLOSED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.DISK_STORE_CREATED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_RECEIVER_CREATED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_RECEIVER_DESTROYED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_RECEIVER_STARTED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_RECEIVER_STOPPED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_SENDER_CREATED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_SENDER_PAUSED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_SENDER_REMOVED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_SENDER_RESUMED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_SENDER_STARTED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.GATEWAY_SENDER_STOPPED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.LOCATOR_STARTED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.LOCK_SERVICE_CLOSED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.LOCK_SERVICE_CREATED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.MANAGER_STARTED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.REGION_CLOSED_PREFIX;
import static org.apache.geode.management.internal.ManagementConstants.REGION_CREATED_PREFIX;

import java.lang.management.ManagementFactory;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.management.InstanceNotFoundException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.Notification;
import javax.management.NotificationBroadcasterSupport;
import javax.management.ObjectInstance;
import javax.management.ObjectName;

import org.apache.logging.log4j.Logger;

import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.wan.GatewayReceiver;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.Locator;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.locks.DLockService;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.ClassLoadUtils;
import org.apache.geode.internal.cache.CacheService;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionedRegionHelper;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.AsyncEventQueueMXBean;
import org.apache.geode.management.CacheServerMXBean;
import org.apache.geode.management.DiskStoreMXBean;
import org.apache.geode.management.GatewayReceiverMXBean;
import org.apache.geode.management.GatewaySenderMXBean;
import org.apache.geode.management.LocatorMXBean;
import org.apache.geode.management.LockServiceMXBean;
import org.apache.geode.management.ManagementException;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.ManagerMXBean;
import org.apache.geode.management.MemberMXBean;
import org.apache.geode.management.RegionMXBean;
import org.apache.geode.management.internal.AlertDetails;
import org.apache.geode.management.internal.FederationComponent;
import org.apache.geode.management.internal.MBeanJMXAdapter;
import org.apache.geode.management.internal.SystemManagementService;
import org.apache.geode.management.membership.ClientMembership;
import org.apache.geode.management.membership.ClientMembershipEvent;
import org.apache.geode.management.membership.ClientMembershipListener;
import org.apache.geode.management.membership.ClientMembershipListenerAdapter;
import org.apache.geode.pdx.internal.PeerTypeRegistration;

/**
 * Acts as an intermediate between MBean layer and Federation Layer. Handles all Call backs from
 * GemFire to instantiate or remove MBeans from GemFire Domain.
 *
 * <p>
 * Even though this class have a lot of utility functions it interacts with the state of the system
 * and contains some state itself.
 */
public class ManagementAdapter {

  private static final Logger logger = LogService.getLogger();

  @Immutable
  private static final List<String> INTERNAL_LOCK_SERVICES =
      unmodifiableList(asList(DLockService.DTLS, DLockService.LTLS,
          PartitionedRegionHelper.PARTITION_LOCK_SERVICE_NAME,
          PeerTypeRegistration.LOCK_SERVICE_NAME));

  private final MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer();
  private final Object regionOpLock = new Object();

  private SystemManagementService service;
  private InternalCache internalCache;
  private String memberSource;
  private NotificationBroadcasterSupport memberLevelNotificationEmitter;
  private MemberMBean memberBean;
  private MBeanAggregator aggregator;
  private MemberMBeanBridge memberMBeanBridge;

  private volatile boolean serviceInitialised;

  /**
   * Adapter life cycle is tied with the Cache. So its better to make all cache level artifacts as
   * instance variable
   */
  void handleCacheCreation(InternalCache cache) throws ManagementException {
    try {
      internalCache = cache;
      service = (SystemManagementService) ManagementService.getManagementService(internalCache);

      memberMBeanBridge = new MemberMBeanBridge(internalCache, service).init();
      memberBean = new MemberMBean(memberMBeanBridge);
      memberLevelNotificationEmitter = memberBean;
      memberSource = MBeanJMXAdapter.getMemberNameOrUniqueId(cache.getMyId());

      // Type casting to MemberMXBean to expose only those methods described in interface
      ObjectName objectName = MBeanJMXAdapter.getMemberMBeanName(cache.getMyId());
      ObjectName federatedName = service.registerInternalMBean(memberBean, objectName);
      service.federate(federatedName, MemberMXBean.class, true);

      serviceInitialised = true;

      // Service initialised is only for ManagementService and not necessarily Manager service.

      // For situations where locator is created before any cache is created
      if (InternalLocator.hasLocator()) {
        handleLocatorStart(InternalLocator.getLocator());
      }

      if (cache.getInternalDistributedSystem().getConfig().getJmxManager()) {
        service.createManager();
        if (cache.getInternalDistributedSystem().getConfig().getJmxManagerStart()) {
          service.startManager();
        }
      }

    } finally {
      if (!serviceInitialised && service != null) {
        service.close();
        if (logger.isDebugEnabled()) {
          logger.debug("Management Service Could not initialise hence closing");
        }

      } else {
        if (logger.isDebugEnabled()) {
          logger.debug("Management Service is initialised and Running");
        }

      }
    }
  }

  /**
   * Handles all the distributed mbean creation part when a Manager is started
   */
  void handleManagerStart() throws ManagementException {
    if (!isServiceInitialised("handleManagerStart")) {
      return;
    }

    DistributedSystemBridge distributedSystemBridge =
        new DistributedSystemBridge(service, internalCache);
    aggregator = new MBeanAggregator(distributedSystemBridge);

    // register the aggregator for Federation framework to use
    service.addProxyListener(aggregator);

    // get the local member mbean as it need to be provided to aggregator first
    MemberMXBean localMemberMXBean = service.getMemberMXBean();

    ObjectName memberObjectName = MBeanJMXAdapter.getMemberMBeanName(
        InternalDistributedSystem.getConnectedInstance().getDistributedMember());

    FederationComponent memberFederation =
        service.getLocalManager().getFedComponents().get(memberObjectName);

    service.afterCreateProxy(memberObjectName, MemberMXBean.class, localMemberMXBean,
        memberFederation);

    MBeanJMXAdapter jmxAdapter = service.getJMXAdapter();
    Map<ObjectName, Object> registeredMBeans = jmxAdapter.getLocalGemFireMBean();

    for (ObjectName objectName : registeredMBeans.keySet()) {
      if (objectName.equals(memberObjectName)) {
        continue;
      }
      Object object = registeredMBeans.get(objectName);
      try {
        ObjectInstance instance = mbeanServer.getObjectInstance(objectName);
        String className = instance.getClassName();
        Class clazz = ClassLoadUtils.classFromName(className);
        Type[] interfaceTypes = clazz.getGenericInterfaces();

        FederationComponent federation =
            service.getLocalManager().getFedComponents().get(objectName);

        for (Type interfaceType : interfaceTypes) {
          Class interfaceTypeAsClass = (Class) interfaceType;
          service.afterCreateProxy(objectName, interfaceTypeAsClass, object, federation);

        }
      } catch (InstanceNotFoundException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("Failed in Registering distributed mbean ");
        }
        throw new ManagementException(e);
      } catch (ClassNotFoundException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("Failed in Registering distributed mbean");
        }
        throw new ManagementException(e);
      }
    }
  }

  /**
   * Handles all the clean up activities when a Manager is stopped It clears the distributed mbeans
   * and underlying data structures
   */
  void handleManagerStop() throws ManagementException {
    if (!isServiceInitialised("handleManagerStop")) {
      return;
    }

    MemberMXBean localMemberMXBean = service.getMemberMXBean();

    ObjectName memberObjectName = MBeanJMXAdapter.getMemberMBeanName(
        InternalDistributedSystem.getConnectedInstance().getDistributedMember());

    FederationComponent memberFederation =
        service.getLocalManager().getFedComponents().get(memberObjectName);

    service.afterRemoveProxy(memberObjectName, MemberMXBean.class, localMemberMXBean,
        memberFederation);

    ObjectName aggregateMBeanPattern = aggregateMBeanPattern();
    MBeanJMXAdapter jmxAdapter = service.getJMXAdapter();
    Map<ObjectName, Object> registeredMBeans = jmxAdapter.getLocalGemFireMBean();
    for (ObjectName objectName : registeredMBeans.keySet()) {
      if (objectName.equals(memberObjectName)) {
        continue;
      }
      if (aggregateMBeanPattern.apply(objectName)) {
        continue;
      }
      Object object = registeredMBeans.get(objectName);
      try {
        ObjectInstance instance = mbeanServer.getObjectInstance(objectName);
        String className = instance.getClassName();
        Class clazz = ClassLoadUtils.classFromName(className);
        Type[] interfaceTypes = clazz.getGenericInterfaces();

        FederationComponent federation =
            service.getLocalManager().getFedComponents().get(objectName);

        for (Type interfaceType : interfaceTypes) {
          Class interfaceTypeClass = (Class) interfaceType;
          service.afterRemoveProxy(objectName, interfaceTypeClass, object, federation);
        }
      } catch (InstanceNotFoundException | ClassNotFoundException e) {
        logger.warn("Failed to invoke aggregator for {} with exception {}", objectName,
            e.getMessage(), e);
      }
    }

    service.removeProxyListener(aggregator);
    aggregator = null;
  }

  /**
   * Assumption is always cache and MemberMbean has been will be created first
   */
  void handleManagerCreation() throws ManagementException {
    if (!isServiceInitialised("handleManagerCreation")) {
      return;
    }

    ObjectName objectName = MBeanJMXAdapter.getManagerName();
    ManagerMBeanBridge managerMBeanBridge = new ManagerMBeanBridge(service);
    ManagerMXBean managerMXBean = new ManagerMBean(managerMBeanBridge);

    ObjectName federatedName = service.registerInternalMBean(managerMXBean, objectName);
    service.federate(federatedName, ManagerMXBean.class, true);

    Notification notification = new Notification(MANAGER_STARTED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), MANAGER_STARTED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);
  }

  /**
   * Handles Region Creation. This is the call back which will create the specified RegionMXBean and
   * will send a notification on behalf of Member Mbean
   *
   * @param region the region for which the call back is invoked
   */
  <K, V> void handleRegionCreation(Region<K, V> region) throws ManagementException {
    if (!isServiceInitialised("handleRegionCreation")) {
      return;
    }

    // Moving region creation operation inside a guarded block
    // After getting access to regionOpLock it again checks for region
    // destroy status

    synchronized (regionOpLock) {
      if (region.isDestroyed()) {
        return;
      }

      // Bridge is responsible for extracting data from GemFire Layer
      RegionMBeanBridge<K, V> regionMBeanBridge = RegionMBeanBridge.getInstance(region);
      RegionMXBean regionMXBean = new RegionMBean<>(regionMBeanBridge);
      ObjectName objectName = MBeanJMXAdapter.getRegionMBeanName(
          internalCache.getDistributedSystem().getDistributedMember(), region.getFullPath());
      ObjectName federatedName = service.registerInternalMBean(regionMXBean, objectName);
      service.federate(federatedName, RegionMXBean.class, true);

      Notification notification = new Notification(REGION_CREATED, memberSource,
          SequenceNumber.next(), System.currentTimeMillis(),
          REGION_CREATED_PREFIX + region.getFullPath());
      memberLevelNotificationEmitter.sendNotification(notification);

      memberMBeanBridge.addRegion(region);
    }
  }

  /**
   * Handles Disk Creation. Will create DiskStoreMXBean and will send a notification
   */
  void handleDiskCreation(DiskStore diskStore) throws ManagementException {
    if (!isServiceInitialised("handleDiskCreation")) {
      return;
    }

    DiskStoreMBeanBridge diskStoreMBeanBridge = new DiskStoreMBeanBridge(diskStore);
    DiskStoreMXBean diskStoreMXBean = new DiskStoreMBean(diskStoreMBeanBridge);
    ObjectName objectName = MBeanJMXAdapter.getDiskStoreMBeanName(
        internalCache.getDistributedSystem().getDistributedMember(), diskStore.getName());
    ObjectName federatedName = service.registerInternalMBean(diskStoreMXBean, objectName);
    service.federate(federatedName, DiskStoreMXBean.class, true);

    Notification notification = new Notification(DISK_STORE_CREATED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(),
        DISK_STORE_CREATED_PREFIX + diskStore.getName());
    memberLevelNotificationEmitter.sendNotification(notification);

    memberMBeanBridge.addDiskStore(diskStore);
  }

  /**
   * Handles LockService Creation
   *
   */
  void handleLockServiceCreation(DLockService lockService) throws ManagementException {
    if (!isServiceInitialised("handleLockServiceCreation")) {
      return;
    }
    // Internal Locks Should not be exposed to client for monitoring
    if (INTERNAL_LOCK_SERVICES.contains(lockService.getName())) {
      return;
    }

    LockServiceMBeanBridge lockServiceMBeanBridge = new LockServiceMBeanBridge(lockService);
    LockServiceMXBean lockServiceMXBean = new LockServiceMBean(lockServiceMBeanBridge);
    ObjectName objectName = MBeanJMXAdapter.getLockServiceMBeanName(
        internalCache.getDistributedSystem().getDistributedMember(), lockService.getName());
    ObjectName federatedName =
        service.registerInternalMBean(lockServiceMXBean, objectName);
    service.federate(federatedName, LockServiceMXBean.class, true);

    Notification notification = new Notification(LOCK_SERVICE_CREATED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(),
        LOCK_SERVICE_CREATED_PREFIX + lockService.getName());
    memberLevelNotificationEmitter.sendNotification(notification);

    memberMBeanBridge.addLockServiceStats(lockService);
  }

  /**
   * Handles GatewaySender creation
   *
   * @param sender the specific gateway sender
   */
  void handleGatewaySenderCreation(GatewaySender sender) throws ManagementException {
    if (!isServiceInitialised("handleGatewaySenderCreation")) {
      return;
    }

    GatewaySenderMBeanBridge gatewaySenderMBeanBridge = new GatewaySenderMBeanBridge(sender);
    GatewaySenderMXBean gatewaySenderMXBean = new GatewaySenderMBean(gatewaySenderMBeanBridge);
    ObjectName objectName = MBeanJMXAdapter.getGatewaySenderMBeanName(
        internalCache.getDistributedSystem().getDistributedMember(), sender.getId());
    ObjectName federatedName = service.registerInternalMBean(gatewaySenderMXBean, objectName);
    service.federate(federatedName, GatewaySenderMXBean.class, true);

    Notification notification = new Notification(GATEWAY_SENDER_CREATED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), GATEWAY_SENDER_CREATED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);
  }

  /**
   * Handles Gateway receiver creation
   *
   * @param gatewayReceiver specific gateway receiver
   */
  void handleGatewayReceiverCreate(GatewayReceiver gatewayReceiver) throws ManagementException {
    if (!isServiceInitialised("handleGatewayReceiverCreate")) {
      return;
    }
    if (!gatewayReceiver.isManualStart()) {
      return;
    }

    createGatewayReceiverMBean(gatewayReceiver);
  }

  private void createGatewayReceiverMBean(GatewayReceiver gatewayReceiver) {
    GatewayReceiverMBeanBridge gatewayReceiverMBeanBridge =
        new GatewayReceiverMBeanBridge(gatewayReceiver);
    GatewayReceiverMXBean gatewayReceiverMXBean =
        new GatewayReceiverMBean(gatewayReceiverMBeanBridge);
    ObjectName objectName = MBeanJMXAdapter
        .getGatewayReceiverMBeanName(internalCache.getDistributedSystem().getDistributedMember());
    ObjectName federatedName = service.registerInternalMBean(gatewayReceiverMXBean, objectName);
    service.federate(federatedName, GatewayReceiverMXBean.class, true);

    Notification notification = new Notification(GATEWAY_RECEIVER_CREATED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), GATEWAY_RECEIVER_CREATED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);
  }

  void handleGatewayReceiverDestroy() throws ManagementException {
    if (!isServiceInitialised("handleGatewayReceiverDestroy")) {
      return;
    }

    GatewayReceiverMBean gatewayReceiverMBean =
        (GatewayReceiverMBean) service.getLocalGatewayReceiverMXBean();
    GatewayReceiverMBeanBridge gatewayReceiverMBeanBridge = gatewayReceiverMBean.getBridge();

    gatewayReceiverMBeanBridge.destroyServer();

    ObjectName objectName = MBeanJMXAdapter
        .getGatewayReceiverMBeanName(internalCache.getDistributedSystem().getDistributedMember());
    service.unregisterMBean(objectName);

    Notification notification = new Notification(GATEWAY_RECEIVER_DESTROYED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), GATEWAY_RECEIVER_DESTROYED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);
  }

  /**
   * Handles Gateway receiver creation
   *
   * @param gatewayReceiver specific gateway receiver
   */
  void handleGatewayReceiverStart(GatewayReceiver gatewayReceiver) throws ManagementException {
    if (!isServiceInitialised("handleGatewayReceiverStart")) {
      return;
    }

    if (!gatewayReceiver.isManualStart()) {
      createGatewayReceiverMBean(gatewayReceiver);
    }

    GatewayReceiverMBean gatewayReceiverMBean =
        (GatewayReceiverMBean) service.getLocalGatewayReceiverMXBean();
    GatewayReceiverMBeanBridge gatewayReceiverMBeanBridge = gatewayReceiverMBean.getBridge();

    gatewayReceiverMBeanBridge.startServer();

    Notification notification = new Notification(GATEWAY_RECEIVER_STARTED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), GATEWAY_RECEIVER_STARTED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);
  }

  void handleGatewayReceiverStop() throws ManagementException {
    if (!isServiceInitialised("handleGatewayReceiverStop")) {
      return;
    }

    GatewayReceiverMBean gatewayReceiverMBean =
        (GatewayReceiverMBean) service.getLocalGatewayReceiverMXBean();
    GatewayReceiverMBeanBridge gatewayReceiverMBeanBridge = gatewayReceiverMBean.getBridge();

    gatewayReceiverMBeanBridge.stopServer();

    Notification notification = new Notification(GATEWAY_RECEIVER_STOPPED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), GATEWAY_RECEIVER_STOPPED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);
  }

  void handleAsyncEventQueueCreation(AsyncEventQueue asyncEventQueue) throws ManagementException {
    if (!isServiceInitialised("handleAsyncEventQueueCreation")) {
      return;
    }

    AsyncEventQueueMBeanBridge asyncEventQueueMBeanBridge =
        new AsyncEventQueueMBeanBridge(asyncEventQueue);
    AsyncEventQueueMXBean asyncEventQueueMXBean =
        new AsyncEventQueueMBean(asyncEventQueueMBeanBridge);
    ObjectName objectName = MBeanJMXAdapter.getAsyncEventQueueMBeanName(
        internalCache.getDistributedSystem().getDistributedMember(), asyncEventQueue.getId());
    ObjectName federatedName = service.registerInternalMBean(asyncEventQueueMXBean, objectName);
    service.federate(federatedName, AsyncEventQueueMXBean.class, true);

    Notification notification = new Notification(ASYNC_EVENT_QUEUE_CREATED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), ASYNC_EVENT_QUEUE_CREATED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);
  }

  void handleAsyncEventQueueRemoval(AsyncEventQueue asyncEventQueue) throws ManagementException {
    if (!isServiceInitialised("handleAsyncEventQueueRemoval")) {
      return;
    }

    ObjectName objectName = MBeanJMXAdapter.getAsyncEventQueueMBeanName(
        internalCache.getDistributedSystem().getDistributedMember(), asyncEventQueue.getId());

    try {
      AsyncEventQueueMBean asyncEventQueueMBean =
          (AsyncEventQueueMBean) service.getLocalAsyncEventQueueMXBean(asyncEventQueue.getId());
      if (asyncEventQueueMBean == null) {
        return;
      }

      asyncEventQueueMBean.stopMonitor();

    } catch (ManagementException e) {
      // If no bean found its a NO-OP
      if (logger.isDebugEnabled()) {
        logger.debug(e.getMessage(), e);
      }
      return;
    }

    service.unregisterMBean(objectName);

    Notification notification = new Notification(ASYNC_EVENT_QUEUE_CLOSED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(),
        ASYNC_EVENT_QUEUE_CLOSED_PREFIX + asyncEventQueue.getId());
    memberLevelNotificationEmitter.sendNotification(notification);
  }

  /**
   * Sends the alert with the Object source as member. This notification will get filtered out for
   * particular alert level
   */
  void handleSystemNotification(AlertDetails alertDetails) {
    if (!isServiceInitialised("handleSystemNotification")) {
      return;
    }

    if (service.isManager()) {
      String systemSource = "DistributedSystem("
          + service.getDistributedSystemMXBean().getDistributedSystemId() + ")";
      Notification notification = new Notification(SYSTEM_ALERT, systemSource,
          SequenceNumber.next(), alertDetails.getMsgTime().getTime(), alertDetails.getMsg());
      notification.setUserData(prepareUserData(alertDetails));
      service.handleNotification(notification);
    }
  }

  private Map<String, String> prepareUserData(AlertDetails alertDetails) {
    Map<String, String> userData = new HashMap<>();

    userData.put(ALERT_LEVEL, getAlertLevelAsString(alertDetails.getAlertLevel()));
    userData.put(THREAD, alertDetails.getSource());
    userData.put(MEMBER, getNameOrId(alertDetails.getSender()));

    return userData;
  }

  /**
   * Assumption is its a cache server instance. For Gateway receiver there will be a separate method
   *
   * @param cacheServer cache server instance
   */
  void handleCacheServerStart(CacheServer cacheServer) {
    if (!isServiceInitialised("handleCacheServerStart")) {
      return;
    }

    CacheServerBridge cacheServerBridge = new CacheServerBridge(internalCache, cacheServer);
    cacheServerBridge.setMemberMBeanBridge(memberMBeanBridge);
    CacheServerMBean cacheServerMBean = new CacheServerMBean(cacheServerBridge);
    ObjectName objectName = MBeanJMXAdapter.getClientServiceMBeanName(
        cacheServer.getPort(), internalCache.getDistributedSystem().getDistributedMember());
    ObjectName federatedName = service.registerInternalMBean(cacheServerMBean, objectName);

    ClientMembershipListener managementClientListener = new CacheServerMembershipListenerAdapter(
        cacheServerMBean, memberLevelNotificationEmitter, federatedName);
    ClientMembership.registerClientMembershipListener(managementClientListener);
    cacheServerBridge.setClientMembershipListener(managementClientListener);

    service.federate(federatedName, CacheServerMXBean.class, true);

    Notification notification = new Notification(CACHE_SERVER_STARTED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), CACHE_SERVER_STARTED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);

    memberMBeanBridge.setCacheServer(true);
  }

  /**
   * Assumption is its a cache server instance. For Gateway receiver there will be a separate method
   */
  void handleCacheServerStop(CacheServer server) {
    if (!isServiceInitialised("handleCacheServerStop")) {
      return;
    }

    CacheServerMBean cacheServerMBean =
        (CacheServerMBean) service.getLocalCacheServerMXBean(server.getPort());

    ClientMembershipListener clientMembershipListener =
        cacheServerMBean.getBridge().getClientMembershipListener();
    if (clientMembershipListener != null) {
      ClientMembership.unregisterClientMembershipListener(clientMembershipListener);
    }

    cacheServerMBean.stopMonitor();

    ObjectName objectName = MBeanJMXAdapter.getClientServiceMBeanName(server.getPort(),
        internalCache.getDistributedSystem().getDistributedMember());
    service.unregisterMBean(objectName);

    Notification notification = new Notification(CACHE_SERVER_STOPPED, memberSource,
        SequenceNumber.next(), System.currentTimeMillis(), CACHE_SERVER_STOPPED_PREFIX);
    memberLevelNotificationEmitter.sendNotification(notification);

    memberMBeanBridge.setCacheServer(false);
  }

  void handleCacheRemoval() throws ManagementException {
    if (!isServiceInitialised("handleCacheRemoval")) {
      return;
    }

    serviceInitialised = false;
    try {
      cleanUpMonitors();
      cleanBridgeResources();
    } catch (Exception e) {
      if (logger.isDebugEnabled()) {
        logger.debug(e.getMessage(), e);
      }
    }

    try {
      service.close();
    } catch (Exception e) {
      logger.warn(e.getMessage(), e);
    } finally {
      internalCache = null;
      service = null;
      memberMBeanBridge = null;
      memberBean = null;
      memberLevelNotificationEmitter = null;
    }
  }

  private String getNameOrId(DistributedMember member) {
    if (member == null) {
      return memberSource;
    }
    return isNotBlank(member.getName()) ? member.getName() : member.getId();
  }

  private void cleanUpMonitors() {
