/*
 * 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.admin.internal;

import static org.apache.geode.admin.internal.InetAddressUtilsWithLogging.toInetAddress;
import static org.apache.geode.distributed.ConfigurationProperties.DISABLE_TCP;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_ADDRESS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;

import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

import org.apache.logging.log4j.Logger;
import org.jgroups.annotations.GuardedBy;

import org.apache.geode.CancelException;
import org.apache.geode.SystemFailure;
import org.apache.geode.admin.AdminException;
import org.apache.geode.admin.Alert;
import org.apache.geode.admin.AlertLevel;
import org.apache.geode.admin.AlertListener;
import org.apache.geode.admin.BackupStatus;
import org.apache.geode.admin.CacheServer;
import org.apache.geode.admin.CacheServerConfig;
import org.apache.geode.admin.CacheVm;
import org.apache.geode.admin.ConfigurationParameter;
import org.apache.geode.admin.DistributedSystemConfig;
import org.apache.geode.admin.DistributionLocator;
import org.apache.geode.admin.DistributionLocatorConfig;
import org.apache.geode.admin.GemFireHealth;
import org.apache.geode.admin.ManagedEntity;
import org.apache.geode.admin.ManagedEntityConfig;
import org.apache.geode.admin.OperationCancelledException;
import org.apache.geode.admin.RuntimeAdminException;
import org.apache.geode.admin.SystemMember;
import org.apache.geode.admin.SystemMemberCacheListener;
import org.apache.geode.admin.SystemMembershipEvent;
import org.apache.geode.admin.SystemMembershipListener;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.persistence.PersistentID;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.FutureCancelledException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.admin.ApplicationVM;
import org.apache.geode.internal.admin.GemFireVM;
import org.apache.geode.internal.admin.GfManagerAgent;
import org.apache.geode.internal.admin.GfManagerAgentConfig;
import org.apache.geode.internal.admin.GfManagerAgentFactory;
import org.apache.geode.internal.admin.remote.CompactRequest;
import org.apache.geode.internal.admin.remote.DistributionLocatorId;
import org.apache.geode.internal.admin.remote.MissingPersistentIDsRequest;
import org.apache.geode.internal.admin.remote.PrepareRevokePersistentIDRequest;
import org.apache.geode.internal.admin.remote.RemoteApplicationVM;
import org.apache.geode.internal.admin.remote.RemoteTransportConfig;
import org.apache.geode.internal.admin.remote.RevokePersistentIDRequest;
import org.apache.geode.internal.admin.remote.ShutdownAllRequest;
import org.apache.geode.internal.cache.backup.BackupOperation;
import org.apache.geode.internal.cache.persistence.PersistentMemberPattern;
import org.apache.geode.internal.logging.Banner;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.logging.LogWriterFactory;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.net.SSLConfig;
import org.apache.geode.internal.util.concurrent.FutureResult;
import org.apache.geode.logging.internal.LoggingSession;
import org.apache.geode.logging.internal.NullLoggingSession;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * Represents a GemFire distributed system for remote administration/management.
 *
 * @since GemFire 3.5
 */
@Deprecated
public class AdminDistributedSystemImpl implements org.apache.geode.admin.AdminDistributedSystem,
    org.apache.geode.internal.admin.JoinLeaveListener,
    org.apache.geode.internal.admin.AlertListener,
    org.apache.geode.distributed.internal.InternalDistributedSystem.DisconnectListener {

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

  /** String identity of this distributed system */
  private String id;

  /** Latest alert broadcast by any system members */
  private Alert latestAlert;

  // -------------------------------------------------------------------------

  /** Internal admin agent to delegate low-level work to */
  private volatile GfManagerAgent gfManagerAgent;

  /** Monitors the health of this distributed system */
  private GemFireHealth health;

  /** Set of non-Manager members in this system */
  private final Set applicationSet = new HashSet();

  /** Set of DistributionLocators for this system */
  private final Set locatorSet = new HashSet();

  /** Set of dedicated CacheServer members in this system */
  private final Set cacheServerSet = new HashSet();

  /** Configuration defining this distributed system */
  private final DistributedSystemConfigImpl config;

  /** Controller for starting and stopping managed entities */
  private final ManagedEntityController controller;

  /** Log file collator for gathering and merging system member logs */
  private final LogCollator logCollator = new LogCollator();

  /**
   * The level above which alerts will be delivered to the alert listeners
   */
  private AlertLevel alertLevel = AlertLevel.WARNING;

  /** The alert listeners registered on this distributed system. */
  private volatile Set<AlertListener> alertListeners = Collections.emptySet();
  private final Object alertLock = new Object();

  private final InternalLogWriter logWriter;

  /** The membership listeners registered on this distributed system */
  private volatile Set membershipListeners = Collections.EMPTY_SET;
  private final Object membershipLock = new Object();

  /* The region listeners registered on this distributed system */
  // for feature requests #32887
  private volatile List cacheListeners = Collections.EMPTY_LIST;
  private final Object cacheListLock = new Object();

  private final LoggingSession loggingSession;

  /**
   * reference to AdminDistributedSystemImpl instance for feature requests #32887.
   * <p>
   * Guarded by {@link #CONNECTION_SYNC}.
   * <p>
   * TODO: reimplement this change and SystemMemberCacheEventProcessor to avoid using this static.
   * SystemMemberCacheEvents should only be sent to Admin VMs that express interest.
   * <p>
   * This is volatile to allow SystemFailure to deliver fatal poison-pill to thisAdminDS without
   * waiting on synchronization.
   *
   */
  @GuardedBy("CONNECTION_SYNC")
  @MakeNotStatic
  private static volatile AdminDistributedSystemImpl thisAdminDS;

  /**
   * Provides synchronization for {@link #connect()} and {@link #disconnect()}. {@link #thisAdminDS}
   * is also now protected by CONNECTION_SYNC and has its lifecycle properly tied to
   * connect/disconnect.
   */
  private static final Object CONNECTION_SYNC = new Object();


  // -------------------------------------------------------------------------
  // Constructor(s)
  // -------------------------------------------------------------------------

  private static LoggingSession createLoggingSession() {
    return NullLoggingSession.create();
  }

  /**
   * Constructs new DistributedSystemImpl with the given configuration.
   *
   * @param config configuration defining this distributed system
   */
  public AdminDistributedSystemImpl(DistributedSystemConfigImpl config) {
    loggingSession = createLoggingSession();

    // init from config...
    this.config = config;

    String systemId = this.config.getSystemId();
    if (systemId != null && systemId.length() > 0) {
      id = systemId;

    }
    if (getLocators() != null && getLocators().length() > 0) {
      id = getLocators();

    } else {
      id = getMcastAddress() + "[" + getMcastPort()
          + "]";
    }

    // LOG: create LogWriterAppender unless one already exists
    loggingSession.startSession();

    // LOG: look in DistributedSystemConfigImpl for existing LogWriter to use
    InternalLogWriter existingLogWriter = this.config.getInternalLogWriter();
    if (existingLogWriter != null) {
      logWriter = existingLogWriter;
    } else {
      // LOG: create LogWriterLogger
      logWriter = LogWriterFactory.createLogWriterLogger(this.config.createLogConfig(), false);
      if (!Boolean.getBoolean(InternalLocator.INHIBIT_DM_BANNER)) {
        // LOG: changed statement from config to info
        logWriter.info(new Banner().getString());
      } else {
        logger.debug("skipping banner - " + InternalLocator.INHIBIT_DM_BANNER + " is set to true");
      }
      // Set this log writer in DistributedSystemConfigImpl
      this.config.setInternalLogWriter(logWriter);
    }

    // set up other details that depend on config attrs...
    controller = ManagedEntityControllerFactory.createManagedEntityController(this);
    initializeDistributionLocators();
    initializeCacheServers();
  }

  // -------------------------------------------------------------------------
  // Initialization
  // -------------------------------------------------------------------------

  /**
   * Creates DistributionLocator instances for every locator entry in the
   * {@link org.apache.geode.admin.DistributedSystemConfig}
   */
  private void initializeDistributionLocators() {
    DistributionLocatorConfig[] configs = config.getDistributionLocatorConfigs();
    if (configs.length == 0) {
      // No work to do
      return;
    }

    for (DistributionLocatorConfig conf : configs) {
      // the Locator impl may vary in this class from the config...
      DistributionLocator locator = createDistributionLocatorImpl(conf);
      locatorSet.add(new FutureResult(locator));
    }
    // update locators string...
    setLocators(parseLocatorSet());
  }

  /**
   * Creates <code>CacheServer</code> instances for every cache server entry in the
   * {@link org.apache.geode.admin.DistributedSystemConfig}
   */
  private void initializeCacheServers() {
    CacheServerConfig[] cacheServerConfigs = config.getCacheServerConfigs();
    for (final CacheServerConfig cacheServerConfig : cacheServerConfigs) {
      try {
        CacheServerConfigImpl copy = new CacheServerConfigImpl(cacheServerConfig);
        cacheServerSet.add(new FutureResult(createCacheServer(copy)));
      } catch (Exception e) {
        logger.warn(e.getMessage(), e);
        continue;
      } catch (VirtualMachineError err) {
        SystemFailure.initiateFailure(err);
        // If this ever returns, rethrow the error. We're poisoned
        // now, so don't let this thread continue.
        throw err;
      } catch (Error e) {
        // Whenever you catch Error or Throwable, you must also
        // catch VirtualMachineError (see above). However, there is
        // _still_ a possibility that you are dealing with a cascading
        // error condition, so you also need to check to see if the JVM
        // is still usable:
        SystemFailure.checkFailure();
        logger.error(e.getMessage(), e);
        continue;
      }
    }
  }

  /**
   * Checks to make sure that {@link #connect()} has been called.
   *
   * @throws IllegalStateException If {@link #connect()} has not been called.
   */
  private void checkConnectCalled() {
    if (gfManagerAgent == null) {
      throw new IllegalStateException(
          "connect() has not been invoked on this AdminDistributedSystem.");
    }
  }

  // -------------------------------------------------------------------------
  // Attributes of this DistributedSystem
  // -------------------------------------------------------------------------

  public GfManagerAgent getGfManagerAgent() {
    return gfManagerAgent;
  }

  @Override
  public boolean isConnected() {
    return gfManagerAgent != null && gfManagerAgent.isConnected();
  }

  @Override
  public String getId() {
    return id;
  }

  @Override
  public String getName() {
    String name = config.getSystemName();
    if (name != null && name.length() > 0) {
      return name;

    } else {
      return getId();
    }
  }

  public String getSystemName() {
    return config.getSystemName();
  }

  @Override
  public String getRemoteCommand() {
    return config.getRemoteCommand();
  }

  @Override
  public void setRemoteCommand(String remoteCommand) {
    config.setRemoteCommand(remoteCommand);
  }

  @Override
  public void setAlertLevel(AlertLevel level) {
    if (isConnected()) {
      gfManagerAgent.setAlertLevel(level.getSeverity());
    }

    alertLevel = level;
  }

  @Override
  public AlertLevel getAlertLevel() {
    return alertLevel;
  }

  @Override
  public void addAlertListener(AlertListener listener) {
    synchronized (alertLock) {
      Set<AlertListener> oldListeners = alertListeners;
      if (!oldListeners.contains(listener)) {
        Set<AlertListener> newListeners = new HashSet<>(oldListeners);
        newListeners.add(listener);
        alertListeners = newListeners;
      }
    }
  }

  public int getAlertListenerCount() {
    synchronized (alertLock) {
      return alertListeners.size();
    }
  }

  @Override
  public void removeAlertListener(AlertListener listener) {
    synchronized (alertLock) {
      Set<AlertListener> oldListeners = alertListeners;
      if (oldListeners.contains(listener)) { // fixed bug 34687
        Set<AlertListener> newListeners = new HashSet<>(oldListeners);
        if (newListeners.remove(listener)) {
          alertListeners = newListeners;
        }
      }
    }
  }

  @Override
  public void addMembershipListener(SystemMembershipListener listener) {
    synchronized (membershipLock) {
      Set oldListeners = membershipListeners;
      if (!oldListeners.contains(listener)) {
        Set newListeners = new HashSet(oldListeners);
        newListeners.add(listener);
        membershipListeners = newListeners;
      }
    }
  }

  @Override
  public void removeMembershipListener(SystemMembershipListener listener) {
    synchronized (membershipLock) {
      Set oldListeners = membershipListeners;
      if (oldListeners.contains(listener)) { // fixed bug 34687
        Set newListeners = new HashSet(oldListeners);
        if (newListeners.remove(listener)) {
          membershipListeners = newListeners;
        }
      }
    }
  }

  @Override
  public String getMcastAddress() {
    return config.getMcastAddress();
  }

  @Override
  public int getMcastPort() {
    return config.getMcastPort();
  }

  public boolean getDisableTcp() {
    return config.getDisableTcp();
  }

  public boolean getDisableAutoReconnect() {
    return config.getDisableAutoReconnect();
  }

  @Override
  public String getLocators() {
    return config.getLocators();
  }

  protected void setLocators(String locators) {
    config.setLocators(locators);
  }

  public String getMembershipPortRange() {
    return getConfig().getMembershipPortRange();
  }

  /** get the direct-channel port to use, or zero if not set */
  public int getTcpPort() {
    return getConfig().getTcpPort();
  }

  public void setTcpPort(int port) {
    getConfig().setTcpPort(port);
  }

  public void setMembershipPortRange(String membershipPortRange) {
    getConfig().setMembershipPortRange(membershipPortRange);
  }

  @Override
  public DistributedSystemConfig getConfig() {
    return config;
  }

  /**
   * Returns true if any members of this system are currently running.
   */
  @Override
  public boolean isRunning() {
    if (gfManagerAgent == null) {
      return false;
    }
    // is there a better way??
    // this.gfManagerAgent.isConnected() ... this.gfManagerAgent.isListening()

    return isAnyMemberRunning();
  }

  /** Returns true if this system can use multicast for communications */
  @Override
  public boolean isMcastEnabled() {
    return getMcastPort() > 0;
  }

  ManagedEntityController getEntityController() {
    return controller;
  }

  private static final String TIMEOUT_MS_NAME = "AdminDistributedSystemImpl.TIMEOUT_MS";
  private static final int TIMEOUT_MS_DEFAULT = 60000; // 30000 -- see bug36470
  private static final int TIMEOUT_MS =
      Integer.getInteger(TIMEOUT_MS_NAME, TIMEOUT_MS_DEFAULT);


  // -------------------------------------------------------------------------
  // Operations of this DistributedSystem
  // -------------------------------------------------------------------------

  /**
   * Starts all managed entities in this system.
   */
  @Override
  public void start() throws AdminException {
    // Wait for each managed entity to start (see bug 32569)
    DistributionLocator[] locs = getDistributionLocators();
    for (final DistributionLocator distributionLocator : locs) {
      distributionLocator.start();
    }
    for (final DistributionLocator loc : locs) {
      try {
        if (!loc.waitToStart(TIMEOUT_MS)) {
          throw new AdminException(
              String.format("%s did not start after %s ms",
                  loc, TIMEOUT_MS));
        }

      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        throw new AdminException(
            String.format("Interrupted while waiting for %s to start.",
                loc),
            ex);
      }
    }

    CacheServer[] servers = getCacheServers();
    for (final CacheServer cacheServer : servers) {
      cacheServer.start();
    }
    for (final CacheServer server : servers) {
      try {
        if (!server.waitToStart(TIMEOUT_MS)) {
          throw new AdminException(
              String.format("%s did not start after %s ms",
                  server, TIMEOUT_MS));
        }

      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        throw new AdminException(
            String.format("Interrupted while waiting for %s to start.",
                server),
            ex);
      }
    }
  }

  /**
   * Stops all GemFire managers that are members of this system.
   */
  @Override
  public void stop() throws AdminException {
    // Stop cache server before GemFire managers because the cache
    // server might host a cache proxy that is dependent on the
    // manager. See bug 32569.

    // Wait for each managed entity to stop (see bug 32569)
    long timeout = 30;

    CacheServer[] servers = getCacheServers();
    for (final CacheServer cacheServer : servers) {
      cacheServer.stop();
    }
    for (final CacheServer server : servers) {
      try {
        if (!server.waitToStop(timeout * 1000)) {
          throw new AdminException(
              String.format("%s did not stop after %s seconds.",
                  server, timeout));
        }

      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        throw new AdminException(
            String.format("Interrupted while waiting for %s to stop.",
                server),
            ex);
      }
    }

    DistributionLocator[] locs = getDistributionLocators();
    for (final DistributionLocator distributionLocator : locs) {
      distributionLocator.stop();
    }
    for (final DistributionLocator loc : locs) {
      try {
        if (!loc.waitToStop(timeout * 1000)) {
          throw new AdminException(
              String.format("%s did not stop after %s seconds.",
                  loc, timeout));
        }

      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        throw new AdminException(
            String.format("Interrupted while waiting for %s to stop.",
                loc),
            ex);
      }
    }
  }

  /** Display merged system member logs */
  @Override
  public String displayMergedLogs() {
    return logCollator.collateLogs(gfManagerAgent);
  }

  /**
   * Returns the license for this GemFire product; else null if unable to retrieve license
   * information
   *
   * @return license for this GemFire product
   */
  @Override
  public java.util.Properties getLicense() {
    SystemMember member = findFirstRunningMember();
    if (member != null) {
      return new Properties();
    } else {
      return null;
    }
  }

  /**
   * Sets the distribution-related portion of the given managed entity's configuration so that the
   * entity is part of this distributed system.
   *
   * @throws AdminException TODO-javadocs
   */
  private void setDistributionParameters(SystemMember member) throws AdminException {

    Assert.assertTrue(member instanceof ManagedSystemMemberImpl);

    // set some config parms to match this system...
    ConfigurationParameter[] configParms = new ConfigurationParameter[] {
        new ConfigurationParameterImpl(MCAST_PORT, config.getMcastPort()),
        new ConfigurationParameterImpl(LOCATORS, config.getLocators()),
        new ConfigurationParameterImpl(MCAST_ADDRESS, toInetAddress(config.getMcastAddress())),
        new ConfigurationParameterImpl(DISABLE_TCP, config.getDisableTcp()),};
    member.setConfiguration(configParms);
  }

  /**
   * Handles an <code>ExecutionException</code> by examining its cause and throwing an appropriate
   * runtime exception.
   */
  private static void handle(ExecutionException ex) {
    Throwable cause = ex.getCause();

    if (cause instanceof OperationCancelledException) {
      // Operation was cancelled, we don't necessary want to propagate
      // this up to the user.
      return;
    }
    if (cause instanceof CancelException) { // bug 37285
      throw new FutureCancelledException(
          "Future cancelled due to shutdown",
          ex);
    }

    // Don't just throw the cause because the stack trace can be
    // misleading. For instance, the cause might have occurred in a
    // different thread. In addition to the cause, we also want to
    // know which code was waiting for the Future.
    throw new RuntimeAdminException(
        "While waiting for Future",
        ex);
  }

  protected void checkCancellation() {
    DistributionManager dm = getDistributionManager();
    // TODO does dm == null mean we're dead?
    if (dm != null) {
      dm.getCancelCriterion().checkCancelInProgress(null);
    }
  }

  /**
   * Returns a list of manageable SystemMember instances for each member of this distributed system.
   *
   * @return array of system members for each non-manager member
   */
  @Override
  public SystemMember[] getSystemMemberApplications() throws org.apache.geode.admin.AdminException {
    synchronized (applicationSet) {
      Collection coll = new ArrayList(applicationSet.size());
      APPS: for (final Object o : applicationSet) {
        Future future = (Future) o;
        for (;;) {
          checkCancellation();
          boolean interrupted = Thread.interrupted();
          try {
            coll.add(future.get());
            break;
          } catch (InterruptedException ex) {
            interrupted = true;
            continue; // keep trying
          } catch (CancellationException ex) {
            continue APPS;
          } catch (ExecutionException ex) {
            handle(ex);
            continue APPS;
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
          }
        } // for
      } // APPS
      SystemMember[] array = new SystemMember[coll.size()];
      coll.toArray(array);
      return array;
    }
  }

  /**
   * Display in readable format the latest Alert in this distributed system.
   *
   * TODO: create an external admin api object for Alert
   */
  @Override
  public String getLatestAlert() {
    if (latestAlert == null) {
      return "";
    }
    return latestAlert.toString();
  }

  /**
   * Connects to the currently configured system.
   */
  @Override
  public void connect() {
    connect(logWriter);
  }

  /**
   * Connects to the currently configured system. This method is public for internal use only
   * (testing, for example).
   *
   * <p>
   *
   * See {@link org.apache.geode.distributed.DistributedSystem#connect} for a list of exceptions
   * that may be thrown.
   *
   * @param logWriter the InternalLogWriter to use for any logging
   */
  public void connect(InternalLogWriter logWriter) {
    synchronized (CONNECTION_SYNC) {
      // Check if the gfManagerAgent is NOT null.
      // If it is already listening, then just return since the connection is already established OR
      // in process.
      // Otherwise cleanup the state of AdminDistributedSystemImpl. This needs to happen
      // automatically.
      if (gfManagerAgent != null) {
        if (gfManagerAgent.isListening()) {
          if (logger.isDebugEnabled()) {
            logger.debug(
                "The RemoteGfManagerAgent is already listening for this AdminDistributedSystem.");
          }
          return;
        }
        disconnect();
      }

      if (thisAdminDS != null) { // TODO: beef up toString and add thisAdminDS
        throw new IllegalStateException(
            "Only one AdminDistributedSystem connection can be made at once.");
      }

      thisAdminDS = this; // added for feature requests #32887

      if (getLocators().length() == 0) {
        id = getMcastAddress() + "[" + getMcastPort() + "]";

      } else {
        id = getLocators();
      }

      if (config instanceof DistributedSystemConfigImpl) {
        config.validate();
        config.setDistributedSystem(this);
      }

      // LOG: passes the AdminDistributedSystemImpl LogWriterLogger into GfManagerAgentConfig for
      // RemoteGfManagerAgent
      gfManagerAgent = GfManagerAgentFactory.getManagerAgent(buildAgentConfig(logWriter));

      // sync to prevent bug 33341 Admin API can double-represent system members
      synchronized (membershipListenerLock) {
        // build the list of applications...
        ApplicationVM[] apps = gfManagerAgent.listApplications();
        for (final ApplicationVM app : apps) {
          try {
            nodeJoined(null, app);
          } catch (RuntimeAdminException e) {
            this.logWriter.warning("encountered a problem processing member " + app);
          }
        }
      }

      // Build admin objects for all locators (see bug 31959)
      String locators = getLocators();
      StringTokenizer st = new StringTokenizer(locators, ",");
      NEXT: while (st.hasMoreTokens()) {
        String locator = st.nextToken();
        int first = locator.indexOf("[");
        int last = locator.indexOf("]");
        String host = locator.substring(0, first);
        int colidx = host.lastIndexOf('@');
        if (colidx < 0) {
          colidx = host.lastIndexOf(':');
        }
        String bindAddr = null;
        if (colidx > 0 && colidx < (host.length() - 1)) {
          String orig = host;
          bindAddr = host.substring(colidx + 1);
          host = host.substring(0, colidx);
          // if the host contains a colon and there's no '@', we probably
          // parsed an ipv6 address incorrectly - try again
          if (host.indexOf(':') >= 0) {
            int bindidx = orig.lastIndexOf('@');
            if (bindidx >= 0) {
              host = orig.substring(0, bindidx);
              bindAddr = orig.substring(bindidx + 1);
            } else {
              host = orig;
              bindAddr = null;
            }
          }
        }
        int port = Integer.parseInt(locator.substring(first + 1, last));

        synchronized (locatorSet) {
          LOCATORS: for (final Object o : locatorSet) {
            Future future = (Future) o;
            DistributionLocatorImpl impl = null;
            for (;;) {
              checkCancellation();
              boolean interrupted = Thread.interrupted();
              try {
                impl = (DistributionLocatorImpl) future.get();
                break; // success
              } catch (InterruptedException ex) {
                interrupted = true;
                continue; // keep trying
              } catch (CancellationException ex) {
                continue LOCATORS;
              } catch (ExecutionException ex) {
                handle(ex);
                continue LOCATORS;
              } finally {
                if (interrupted) {
                  Thread.currentThread().interrupt();
                }
              }
            } // for

            DistributionLocatorConfig conf = impl.getConfig();

            InetAddress host1 = toInetAddress(host);
            InetAddress host2 = toInetAddress(conf.getHost());
            if (port == conf.getPort() && host1.equals(host2)) {
              // Already have an admin object for this locator
              continue NEXT;
            }
          }
        }

        // None of the existing locators matches the locator in the
        // string. Contact the locator to get information and create
        // an admin object for it.
        InetAddress bindAddress = null;
        if (bindAddr != null) {
          bindAddress = toInetAddress(bindAddr);
        }
        DistributionLocatorConfig conf =
            DistributionLocatorConfigImpl.createConfigFor(host, port, bindAddress);
        if (conf != null) {
          DistributionLocator impl = createDistributionLocatorImpl(conf);
          synchronized (locatorSet) {
            locatorSet.add(new FutureResult(impl));
          }
        }
      }
    }
  }

  /**
   * Polls to determine whether or not the connection to the distributed system has been made.
   */
  @Override
  public boolean waitToBeConnected(long timeout) throws InterruptedException {

    if (Thread.interrupted()) {
      throw new InterruptedException();
    }

    checkConnectCalled();

    long start = System.currentTimeMillis();
    while (System.currentTimeMillis() - start < timeout) {
      if (gfManagerAgent.isInitialized()) {
        return true;

      } else {
        Thread.sleep(100);
      }
    }

    return isConnected();
  }

  /**
   * Closes all connections and resources to the connected distributed system.
   *
   * @see org.apache.geode.distributed.DistributedSystem#disconnect()
   */
  @Override
  public void disconnect() {
    synchronized (CONNECTION_SYNC) {
      loggingSession.stopSession();
      try {
        if (thisAdminDS == this) {
          thisAdminDS = null;
        }
        if (gfManagerAgent != null && gfManagerAgent.isListening()) {
          synchronized (this) {
            if (health != null) {
              health.close();
            }
          }
          gfManagerAgent.removeJoinLeaveListener(this);
          gfManagerAgent.disconnect();
        }
        gfManagerAgent = null;
        if (config instanceof DistributedSystemConfigImpl) {
          config.setDistributedSystem(null);
        }
      } finally {
        loggingSession.shutdown();
      }
    }
  }

  /**
   * Returns the DistributionManager this implementation is using to connect to the distributed
   * system.
   */
  public DistributionManager getDistributionManager() {
    GfManagerAgent agent = gfManagerAgent;
    if (agent == null) {
      return null;
    }
    return agent.getDM();
  }

  /**
   * Returns the internal admin API's agent used for administering this
   * <code>AdminDistributedSystem</code>.
   *
   * @since GemFire 4.0
   */
  public GfManagerAgent getAdminAgent() {
    return gfManagerAgent;
  }

