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

import java.io.IOException;
import java.io.NotSerializableException;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Semaphore;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.SystemFailure;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.DMStats;
import org.apache.geode.distributed.internal.DirectReplyProcessor;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.MembershipManager;
import org.apache.geode.internal.alerting.AlertingAction;
import org.apache.geode.internal.cache.DirectReplyMessage;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.net.SocketCreator;
import org.apache.geode.internal.tcp.BaseMsgStreamer;
import org.apache.geode.internal.tcp.ConnectExceptions;
import org.apache.geode.internal.tcp.Connection;
import org.apache.geode.internal.tcp.ConnectionException;
import org.apache.geode.internal.tcp.MemberShunnedException;
import org.apache.geode.internal.tcp.MsgStreamer;
import org.apache.geode.internal.tcp.TCPConduit;
import org.apache.geode.internal.util.Breadcrumbs;
import org.apache.geode.internal.util.concurrent.ReentrantSemaphore;

/**
 * DirectChannel is used to interact directly with other Direct servers to distribute GemFire
 * messages to other nodes. It is held by a
 * org.apache.geode.internal.cache.distribution.DistributionChannel, which is used by the
 * DistributionManager to send and receive asynchronous messages.
 */
public class DirectChannel {

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

  /** this is the conduit used for communications */
  private final transient TCPConduit conduit;

  private volatile boolean disconnected = true;

  /** This is set to true when completely disconnected (all connections are closed) */
  private volatile boolean disconnectCompleted = true;

  /** this is the DistributionManager, most of the time */
  private final DirectChannelListener receiver;

  private final InetAddress address;

  InternalDistributedMember localAddr;

  /**
   * Callback to set the local address, must be done before this channel is used.
   *
   * @throws ConnectionException if the conduit has stopped
   */
  public void setLocalAddr(InternalDistributedMember localAddr) {
    this.localAddr = localAddr;
    conduit.setMemberId(localAddr);
    if (disconnected) {
      disconnected = false;
      disconnectCompleted = false;
      this.groupOrderedSenderSem = new ReentrantSemaphore(MAX_GROUP_SENDERS);
      this.groupUnorderedSenderSem = new ReentrantSemaphore(MAX_GROUP_SENDERS);
    }
  }

  /**
   * Returns the cancel criterion for the channel, which will note if the channel is abnormally
   * closing
   */
  public CancelCriterion getCancelCriterion() {
    return conduit.getCancelCriterion();
  }

  public DirectChannel(MembershipManager mgr, DirectChannelListener listener, DistributionConfig dc)
      throws ConnectionException {
    this.receiver = listener;

    this.address = initAddress(dc);
    boolean isBindAddress = dc.getBindAddress() != null;
    try {
      int port = Integer.getInteger("tcpServerPort", 0).intValue();
      if (port == 0) {
        port = dc.getTcpPort();
      }
      Properties props = System.getProperties();
      if (props.getProperty("p2p.shareSockets") == null) {
        props.setProperty("p2p.shareSockets", String.valueOf(dc.getConserveSockets()));
      }
      if (dc.getSocketBufferSize() != DistributionConfig.DEFAULT_SOCKET_BUFFER_SIZE) {
        // Note that the system property "p2p.tcpBufferSize" will be
        // overridden by the new "socket-buffer-size".
        props.setProperty("p2p.tcpBufferSize", String.valueOf(dc.getSocketBufferSize()));
      }
      if (props.getProperty("p2p.idleConnectionTimeout") == null) {
        props.setProperty("p2p.idleConnectionTimeout", String.valueOf(dc.getSocketLeaseTime()));
      }
      int[] range = dc.getMembershipPortRange();
      props.setProperty("membership_port_range_start", "" + range[0]);
      props.setProperty("membership_port_range_end", "" + range[1]);

      this.conduit = new TCPConduit(mgr, port, address, isBindAddress, this, props);
      disconnected = false;
      disconnectCompleted = false;
      this.groupOrderedSenderSem = new ReentrantSemaphore(MAX_GROUP_SENDERS);
      this.groupUnorderedSenderSem = new ReentrantSemaphore(MAX_GROUP_SENDERS);
      logger.info("GemFire P2P Listener started on {}",
          conduit.getSocketId());

    } catch (ConnectionException ce) {
      logger.fatal(String.format("Unable to initialize direct channel because: %s",
          new Object[] {ce.getMessage()}),
          ce);
      throw ce; // fix for bug 31973
    }
  }


  /**
   * Return how many concurrent operations should be allowed by default. since 6.6, this has been
   * raised to Integer.MAX value from the number of available processors. Setting this to a lower
   * value raises the possibility of a deadlock when serializing a message with PDX objects, because
   * the PDX serialization can trigger further distribution.
   */
  public static final int DEFAULT_CONCURRENCY_LEVEL =
      Integer.getInteger("p2p.defaultConcurrencyLevel", Integer.MAX_VALUE / 2).intValue();

  /**
   * The maximum number of concurrent senders sending a message to a group of recipients.
   */
  private static final int MAX_GROUP_SENDERS =
      Integer.getInteger("p2p.maxGroupSenders", DEFAULT_CONCURRENCY_LEVEL).intValue();
  private Semaphore groupUnorderedSenderSem;
  private Semaphore groupOrderedSenderSem;

  private Semaphore getGroupSem(boolean ordered) {
    if (ordered) {
      return this.groupOrderedSenderSem;
    } else {
      return this.groupUnorderedSenderSem;
    }
  }

  private void acquireGroupSendPermission(boolean ordered) {
    if (this.disconnected) {
      throw new org.apache.geode.distributed.DistributedSystemDisconnectedException(
          "Direct channel has been stopped");
    }
    // @todo darrel: add some stats
    final Semaphore s = getGroupSem(ordered);
    for (;;) {
      this.conduit.getCancelCriterion().checkCancelInProgress(null);
      boolean interrupted = Thread.interrupted();
      try {
        s.acquire();
        break;
      } catch (InterruptedException ex) {
        interrupted = true;
      } finally {
        if (interrupted) {
          Thread.currentThread().interrupt();
        }
      }
    } // for
    if (this.disconnected) {
      s.release();
      throw new DistributedSystemDisconnectedException(
          "communications disconnected");
    }
  }

  private void releaseGroupSendPermission(boolean ordered) {
    final Semaphore s = getGroupSem(ordered);
    s.release();
  }

  /**
   * Returns true if calling thread owns its own communication resources.
   */
  boolean threadOwnsResources() {
    DistributionManager d = getDM();
    if (d != null) {
      return d.getSystem().threadOwnsResources() && !AlertingAction.isThreadAlerting();
    }
    return false;

  }

  /**
   * This is basically just sendToMany, giving us a way to see on the stack whether we are sending
   * to a single member or multiple members, in which case the group-send lock will be held during
   * distribution.
   *
   * @param mgr - the membership manager
   * @param p_destinations - the list of addresses to send the message to.
   * @param msg - the message to send
   * @param ackSAThreshold the severe alert threshold
   * @return number of bytes sent
   * @throws ConnectExceptions if message could not be send to its <code>destination</code>
   * @throws NotSerializableException If the msg cannot be serialized
   */
  private int sendToOne(final MembershipManager mgr, InternalDistributedMember[] p_destinations,
      final DistributionMessage msg, long ackWaitThreshold, long ackSAThreshold)
      throws ConnectExceptions, NotSerializableException {
    return sendToMany(mgr, p_destinations, msg, ackWaitThreshold, ackSAThreshold);
  }


  /**
   * Sends a msg to a list of destinations. This code does some special optimizations to stream
   * large messages
   *
   * @param mgr - the membership manager
   * @param p_destinations - the list of addresses to send the message to.
   * @param msg - the message to send
   * @param ackSAThreshold the severe alert threshold
   * @return number of bytes sent
   * @throws ConnectExceptions if message could not be send to its <code>destination</code>
   * @throws NotSerializableException If the msg cannot be serialized
   */
  private int sendToMany(final MembershipManager mgr, InternalDistributedMember[] p_destinations,
      final DistributionMessage msg, long ackWaitThreshold, long ackSAThreshold)
      throws ConnectExceptions, NotSerializableException {
    InternalDistributedMember destinations[] = p_destinations;

    // Collects connect exceptions that happened during previous attempts to send.
    // These represent members we are not able to distribute to.
    ConnectExceptions failedCe = null;
    // Describes the destinations that we need to retry the send to.
    ConnectExceptions retryInfo = null;
    int bytesWritten = 0;
    boolean retry = false;
    final boolean orderedMsg = msg.orderedDelivery() || Connection.isDominoThread();
    // Connections we actually sent messages to.
    final List totalSentCons = new ArrayList(destinations.length);
    boolean interrupted = false;

    long ackTimeout = 0;
    long ackSDTimeout = 0;
    long startTime = 0;
    final DirectReplyMessage directMsg;
    if (msg instanceof DirectReplyMessage) {
      directMsg = (DirectReplyMessage) msg;
    } else {
      directMsg = null;
    }
    if (directMsg != null || msg.getProcessorId() > 0) {
      ackTimeout = (int) (ackWaitThreshold * 1000);
      if (msg.isSevereAlertCompatible() || ReplyProcessor21.isSevereAlertProcessingForced()) {
        ackSDTimeout = (int) (ackSAThreshold * 1000);
        if (ReplyProcessor21.getShortSevereAlertProcessing()) {
          ackSDTimeout = (int) (ReplyProcessor21.PR_SEVERE_ALERT_RATIO * ackSDTimeout);
        }
      }
    }

    boolean directReply = false;
    if (directMsg != null && directMsg.supportsDirectAck() && threadOwnsResources()) {
      directReply = true;
    }

    // If this is a direct reply message, but we are sending it
    // over the shared socket, tell the message it needs to
    // use a regular reply processor.
    if (!directReply && directMsg != null) {
      directMsg.registerProcessor();
    }
    if (logger.isDebugEnabled()) {
      logger.debug("Sending ({}) to {} peers ({}) via tcp/ip",
          msg, p_destinations.length, Arrays.toString(p_destinations));
    }

    try {
      do {
        interrupted = Thread.interrupted() || interrupted;
        /**
         * Exceptions that happened during one attempt to send
         */
        if (retryInfo != null) {
          // need to retry to each of the members in the exception
          List retryMembers = retryInfo.getMembers();
          InternalDistributedMember[] retryDest =
              new InternalDistributedMember[retryMembers.size()];
          retryDest = (InternalDistributedMember[]) retryMembers.toArray(retryDest);
          destinations = retryDest;
          retryInfo = null;
          retry = true;
        }
        final List cons = new ArrayList(destinations.length);
        ConnectExceptions ce = getConnections(mgr, msg, destinations, orderedMsg, retry, ackTimeout,
            ackSDTimeout, cons);
        if (directReply && msg.getProcessorId() > 0) { // no longer a direct-reply message?
          directReply = false;
        }
        if (ce != null) {
          if (failedCe != null) {
            failedCe.getMembers().addAll(ce.getMembers());
            failedCe.getCauses().addAll(ce.getCauses());
          } else {
            failedCe = ce;
          }
          ce = null;
        }
        if (cons.isEmpty()) {
          if (failedCe != null) {
            throw failedCe;
          }
          return bytesWritten;
        }

        boolean sendingToGroup = cons.size() > 1;
        Connection permissionCon = null;
        if (sendingToGroup) {
          acquireGroupSendPermission(orderedMsg);
        } else {
          // sending over just one connection
          permissionCon = (Connection) cons.get(0);
          if (permissionCon != null) {
            try {
              permissionCon.acquireSendPermission();
            } catch (ConnectionException conEx) {
              // Set retryInfo and then retry.
              // We want to keep calling TCPConduit.getConnection until it doesn't
              // return a connection.
              retryInfo = new ConnectExceptions();
              retryInfo.addFailure(permissionCon.getRemoteAddress(), conEx);
              continue;
            }
          }
        }

        try {
          if (retry && logger.isDebugEnabled()) {
            logger.debug("Retrying send ({}{}) to {} peers ({}) via tcp/ip",
                msg, cons.size(), cons);
          }
          DMStats stats = getDMStats();
          List<?> sentCons; // used for cons we sent to this time

          final BaseMsgStreamer ms =
              MsgStreamer.create(cons, msg, directReply, stats, getConduit().getBufferPool());
          try {
            startTime = 0;
            if (ackTimeout > 0) {
              startTime = System.currentTimeMillis();
            }
            ms.reserveConnections(startTime, ackTimeout, ackSDTimeout);

            int result = ms.writeMessage();
            if (bytesWritten == 0) {
              // bytesWritten only needs to be set once.
              // if we have to do a retry we don't want to count
              // each one's bytes.
              bytesWritten = result;
            }
            ce = ms.getConnectExceptions();
            sentCons = ms.getSentConnections();

            totalSentCons.addAll(sentCons);
          } catch (NotSerializableException e) {
            throw e;
          } catch (IOException ex) {
            throw new InternalGemFireException(
                "Unknown error serializing message",
                ex);
          } finally {
            try {
              ms.close();
            } catch (IOException e) {
              throw new InternalGemFireException("Unknown error serializing message", e);
            }
          }

          if (ce != null) {
            retryInfo = ce;
            ce = null;
          }

          if (directReply && !sentCons.isEmpty()) {
            long readAckStart = 0;
            if (stats != null) {
              readAckStart = stats.startReplyWait();
            }
            try {
              ce = readAcks(sentCons, startTime, ackTimeout, ackSDTimeout, ce,
                  directMsg.getDirectReplyProcessor());
            } finally {
              if (stats != null) {
                stats.endReplyWait(readAckStart, startTime);
              }
            }
          }
        } finally {
          if (sendingToGroup) {
            releaseGroupSendPermission(orderedMsg);
          } else if (permissionCon != null) {
            permissionCon.releaseSendPermission();
          }
        }
        if (ce != null) {
          if (retryInfo != null) {
            retryInfo.getMembers().addAll(ce.getMembers());
            retryInfo.getCauses().addAll(ce.getCauses());
          } else {
            retryInfo = ce;
          }
          ce = null;
        }
        if (retryInfo != null) {
          this.conduit.getCancelCriterion().checkCancelInProgress(null);
        }
      } while (retryInfo != null);
    } finally {
      if (interrupted) {
        Thread.currentThread().interrupt();
      }
      for (Iterator it = totalSentCons.iterator(); it.hasNext();) {
        Connection con = (Connection) it.next();
        con.setInUse(false, 0, 0, 0, null);
      }
    }
    if (failedCe != null) {
      throw failedCe;
    }
    return bytesWritten;
  }

  private ConnectExceptions readAcks(List sentCons, long startTime, long ackTimeout,
      long ackSDTimeout, ConnectExceptions cumulativeExceptions, DirectReplyProcessor processor) {

    ConnectExceptions ce = cumulativeExceptions;

    for (Iterator it = sentCons.iterator(); it.hasNext();) {
      Connection con = (Connection) it.next();
      // We don't expect replies on shared connections.
      if (con.isSharedResource()) {
        continue;
      }
      try {
        try {
          con.readAck(processor);
        } catch (SocketTimeoutException ex) {
          handleAckTimeout(ackTimeout, ackSDTimeout, con, processor);
        }
      } catch (ConnectionException conEx) {
        if (ce == null) {
          ce = new ConnectExceptions();
        }
        ce.addFailure(con.getRemoteAddress(), conEx);
      }
    }
    return ce;
  }

  /**
   * Obtain the connections needed to transmit a message. The connections are put into the cons
   * object (the last parameter)
   *
   * @param mgr the membership manager
   * @param msg the message to send
   * @param destinations who to send the message to
   * @param preserveOrder true if the msg should ordered
   * @param retry whether this is a retransmission
   * @param ackTimeout the ack warning timeout
   * @param ackSDTimeout the ack severe alert timeout
   * @param cons a list to hold the connections
   * @return null if everything went okay, or a ConnectExceptions object if some connections
   *         couldn't be obtained
   */
  private ConnectExceptions getConnections(MembershipManager mgr, DistributionMessage msg,
      InternalDistributedMember[] destinations, boolean preserveOrder, boolean retry,
      long ackTimeout, long ackSDTimeout, List cons) {
    ConnectExceptions ce = null;
    for (int i = 0; i < destinations.length; i++) {
      InternalDistributedMember destination = destinations[i];
      if (destination == null) {
        continue;
      }
      if (localAddr.equals(destination)) {
        // jgroups does not deliver messages to a sender, so we don't support
        // it here either.
        continue;
      }

      if (!mgr.memberExists(destination) || mgr.shutdownInProgress()
          || mgr.isShunned(destination)) {
        // This should only happen if the member is no longer in the view.
        if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
          logger.trace(LogMarker.DM_VERBOSE, "Not a member: {}", destination);
        }
        if (ce == null)
          ce = new ConnectExceptions();
        ce.addFailure(destination, new ShunnedMemberException(
            String.format("Member is being shunned: %s", destination)));
      } else {
        try {
          long startTime = 0;
          if (ackTimeout > 0) {
            startTime = System.currentTimeMillis();
          }
          Connection con = conduit.getConnection(destination, preserveOrder, retry, startTime,
              ackTimeout, ackSDTimeout);

          con.setInUse(true, startTime, 0, 0, null); // fix for bug#37657
          cons.add(con);
          if (con.isSharedResource() && msg instanceof DirectReplyMessage) {
            DirectReplyMessage directMessage = (DirectReplyMessage) msg;
            directMessage.registerProcessor();
          }
        } catch (IOException ex) {
          if (ce == null)
            ce = new ConnectExceptions();
          ce.addFailure(destination, ex);
        }
      }
    } // for
    return ce;
  }


  /**
   * Method send.
   *
   * @param mgr - the membership manager
   * @param destinations - the address(es) to send the message to.
   * @param msg - the message to send
   * @param ackSAThreshold severe alert threshold
   * @return number of bytes sent
   * @throws ConnectExceptions if message could not be send to one or more of the
   *         <code>destinations</code>
   * @throws NotSerializableException If the content cannot be serialized
   * @throws ConnectionException if the conduit has stopped
   */
  public int send(MembershipManager mgr, InternalDistributedMember[] destinations,
      DistributionMessage msg, long ackWaitThreshold, long ackSAThreshold)
      throws ConnectExceptions, NotSerializableException {

    if (disconnected) {
      if (logger.isDebugEnabled()) {
        logger.debug("Returning from DirectChannel send because channel is disconnected: {}", msg);
      }
      return 0;
    }
    if (destinations == null) {
      if (logger.isDebugEnabled()) {
        logger.debug("Returning from DirectChannel send because null set passed in: {}", msg);
      }
      return 0;
    }
    if (destinations.length == 0) {
      if (logger.isDebugEnabled()) {
        logger.debug("Returning from DirectChannel send because empty destinations passed in {}",
            msg);
      }
      return 0;
    }

    msg.setSender(localAddr);
    if (destinations.length == 1) {
      return sendToOne(mgr, destinations, msg, ackWaitThreshold, ackSAThreshold);
    } else {
      return sendToMany(mgr, destinations, msg, ackWaitThreshold, ackSAThreshold);
    }
  }


  /**
   * Returns null if no stats available.
   */
  public DMStats getDMStats() {
    DistributionManager dm = getDM();
    if (dm != null) {
      return dm.getStats(); // fix for bug#34004
    } else {
      return null;
    }
  }

  /**
   * Returns null if no config is available.
   *
   * @since GemFire 4.2.2
   */
  public DistributionConfig getDMConfig() {
    DistributionManager dm = getDM();
    if (dm != null) {
      return dm.getConfig();
    } else {
      return null;
    }
  }

  /**
   * Returns null if no dm available.
   */
  public DistributionManager getDM() {
    return this.receiver.getDM();
  }

  /**
   *
   * @param ackTimeout ack wait threshold
   * @param ackSATimeout severe alert threshold
   */
  private void handleAckTimeout(long ackTimeout, long ackSATimeout, Connection c,
      DirectReplyProcessor processor) throws ConnectionException {
    DistributionManager dm = getDM();
    Set activeMembers = dm.getDistributionManagerIds();

    // Increment the stat
    dm.getStats().incReplyTimeouts();

    // an alert that will show up in the console
    {
      String msg =
          "%s seconds have elapsed while waiting for reply from %s on %s whose current membership list is: [%s]";
      final Object[] msgArgs = new Object[] {Long.valueOf(ackTimeout / 1000), c.getRemoteAddress(),
          dm.getId(), activeMembers};
      logger.warn(String.format(msg, msgArgs));
      msgArgs[3] = "(omitted)";
      Breadcrumbs.setProblem(msg, msgArgs);

      if (ReplyProcessor21.THROW_EXCEPTION_ON_TIMEOUT) {
        // init the cause to be a TimeoutException so catchers can determine cause
        TimeoutException cause =
            new TimeoutException("Timed out waiting for ACKS.");
        throw new InternalGemFireException(String.format(msg, msgArgs), cause);
      }
    }

    if (activeMembers.contains(c.getRemoteAddress())) {
      // wait for ack-severe-alert-threshold period first, then wait forever
      if (ackSATimeout > 0) {
        try {
          c.readAck(processor);
          return;
        } catch (SocketTimeoutException e) {
          Object[] args = new Object[] {Long.valueOf((ackSATimeout + ackTimeout) / 1000),
              c.getRemoteAddress(), dm.getId(), activeMembers};
          logger.fatal(
              "{} seconds have elapsed while waiting for reply from {} on {} whose currentFull membership list is: [{}]",
              args);
        }
      }
      try {
        c.readAck(processor);
      } catch (SocketTimeoutException ex) {
        // this can never happen when called with timeout of 0
        logger.error(String.format("Unexpected timeout while waiting for ack from %s",
            c.getRemoteAddress()),
            ex);
      }
    } else {
      logger.warn("View no longer has {} as an active member, so we will no longer wait for it.",
          c.getRemoteAddress());
      processor.memberDeparted(getDM(), c.getRemoteAddress(), true);
    }
  }

  public void receive(DistributionMessage msg, int bytesRead) {
    if (disconnected) {
      return;
    }
    try {
      receiver.messageReceived(msg);
    } catch (MemberShunnedException e) {
      throw e;
    } catch (CancelException e) {
      // ignore
    } catch (Exception ex) {
      // Don't freak out if the DM is shutting down
      if (!conduit.getCancelCriterion().isCancelInProgress()) {
        logger.fatal("While pulling a message", ex);
      }
    }
  }

  /**
   * Ensure that the TCPConduit class gets loaded.
   *
   * @see SystemFailure#loadEmergencyClasses()
   */
  public static void loadEmergencyClasses() {
    TCPConduit.loadEmergencyClasses();
  }

  /**
   * Close the Conduit
   *
   * @see SystemFailure#emergencyClose()
   */
  public void emergencyClose() {
    this.conduit.emergencyClose();
  }

  /**
   * This closes down the Direct connection. Theoretically you can disconnect and, if you need to
   * use the channel again you can and it will automatically reconnect. Reconnection will cause a
   * new local address to be generated.
   */
  public synchronized void disconnect(Exception cause) {
    this.disconnected = true;
    this.disconnectCompleted = false;
    try {
      groupOrderedSenderSem.release();
    } catch (Error e) {
      // GEODE-1076 - already released
    }
    try {
      groupUnorderedSenderSem.release();
    } catch (Error e) {
      // GEODE-1076 - already released
    }
    this.conduit.stop(cause);
    this.disconnectCompleted = true;
  }

  public boolean isOpen() {
    return !disconnectCompleted;
  }

  /** returns the receiver to which this DirectChannel is delivering messages */
  protected DirectChannelListener getReceiver() {
    return receiver;
  }

  /**
   * Returns the port on which this direct channel sends messages
   */
  public int getPort() {
    return this.conduit.getPort();
  }

  /**
   * Returns the conduit over which this channel sends messages
   *
   * @since GemFire 2.1
   */
  public TCPConduit getConduit() {
    return this.conduit;
  }

  private InetAddress initAddress(DistributionConfig dc) {

    String bindAddress = dc.getBindAddress();

    try {
      /*
       * note: had to change the following to make sure the prop wasn't empty in addition to not
       * null for admin.DistributedSystemFactory
       */
      if (bindAddress != null && bindAddress.length() > 0) {
        return InetAddress.getByName(bindAddress);

      } else {
        return SocketCreator.getLocalHost();
      }
    } catch (java.net.UnknownHostException unhe) {
      throw new RuntimeException(unhe);

    }
  }

  public void closeEndpoint(InternalDistributedMember member, String reason) {
    closeEndpoint(member, reason, true);
  }


  /**
   * Closes any connections used to communicate with the given jgroupsAddress.
   */
  public void closeEndpoint(InternalDistributedMember member, String reason,
      boolean notifyDisconnect) {
    TCPConduit tc = this.conduit;
    if (tc != null) {
      tc.removeEndpoint(member, reason, notifyDisconnect);
    }
  }

  /**
   * adds state for thread-owned serial connections to the given member to the parameter
   * <i>result</i>. This can be used to wait for the state to reach the given level in the member's
   * vm.
   *
   * @param member the member whose state is to be captured
   * @param result the map to add the state to
   * @since GemFire 5.1
   */
  public void getChannelStates(DistributedMember member, Map result) {
    TCPConduit tc = this.conduit;
    if (tc != null) {
      tc.getThreadOwnedOrderedConnectionState(member, result);
    }
  }

  /**
   * wait for the given connections to process the number of messages associated with the connection
   * in the given map
   */
  public void waitForChannelState(DistributedMember member, Map channelState)
      throws InterruptedException {
    if (Thread.interrupted())
      throw new InterruptedException();
    TCPConduit tc = this.conduit;
    if (tc != null) {
      tc.waitForThreadOwnedOrderedConnectionState(member, channelState);
    }
  }

  /**
   * returns true if there are still receiver threads for the given member
   */
  public boolean hasReceiversFor(DistributedMember mbr) {
    return this.conduit.hasReceiversFor(mbr);
  }
}