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

import static java.lang.Boolean.FALSE;
import static java.lang.Boolean.TRUE;
import static java.lang.String.format;
import static java.lang.ThreadLocal.withInitial;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_PEER_AUTH_INIT;
import static org.apache.geode.distributed.internal.DistributionConfigImpl.SECURITY_SYSTEM_PREFIX;
import static org.apache.geode.internal.monitoring.ThreadsMonitoring.Mode.P2PReaderExecutor;
import static org.apache.geode.internal.net.BufferPool.BufferType.TRACKED_RECEIVER;
import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX;

import java.io.DataInput;
import java.io.DataInputStream;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SocketChannel;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLHandshakeException;

import org.apache.commons.validator.routines.InetAddressValidator;
import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.TestOnly;

import org.apache.geode.CancelException;
import org.apache.geode.SerializationException;
import org.apache.geode.SystemFailure;
import org.apache.geode.alerting.internal.spi.AlertingAction;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.ConflationKey;
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.DistributionStats;
import org.apache.geode.distributed.internal.OperationExecutors;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.ReplySender;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.api.MemberShunnedException;
import org.apache.geode.distributed.internal.membership.api.Membership;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.DSFIDFactory;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.SystemTimer.SystemTimerTask;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
import org.apache.geode.internal.monitoring.executor.AbstractExecutor;
import org.apache.geode.internal.net.BufferPool;
import org.apache.geode.internal.net.ByteBufferSharing;
import org.apache.geode.internal.net.ByteBufferVendor;
import org.apache.geode.internal.net.NioFilter;
import org.apache.geode.internal.net.NioPlainEngine;
import org.apache.geode.internal.net.SocketCreator;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.Version;
import org.apache.geode.internal.serialization.Versioning;
import org.apache.geode.internal.serialization.VersioningIO;
import org.apache.geode.internal.tcp.MsgReader.Header;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * Connection is a socket holder that sends and receives serialized message objects. A Connection
 * may be closed to preserve system resources and will automatically be reopened when it's needed.
 *
 * @since GemFire 2.0
 */
public class Connection implements Runnable {
  private static final Logger logger = LogService.getLogger();

  public static final String THREAD_KIND_IDENTIFIER = "P2P message reader";

  @MakeNotStatic
  private static int P2P_CONNECT_TIMEOUT;
  @MakeNotStatic
  private static boolean IS_P2P_CONNECT_TIMEOUT_INITIALIZED;

  static final int NORMAL_MSG_TYPE = 0x4c;
  static final int CHUNKED_MSG_TYPE = 0x4d; // a chunk of one logical msg
  static final int END_CHUNKED_MSG_TYPE = 0x4e; // last in a series of chunks
  static final int DIRECT_ACK_BIT = 0x20;

  static final int MSG_HEADER_SIZE_OFFSET = 0;
  static final int MSG_HEADER_TYPE_OFFSET = 4;
  static final int MSG_HEADER_ID_OFFSET = 5;
  static final int MSG_HEADER_BYTES = 7;

  /**
   * Small buffer used for send socket buffer on receiver connections and receive buffer on sender
   * connections.
   */
  public static final int SMALL_BUFFER_SIZE =
      Integer.getInteger(GEMFIRE_PREFIX + "SMALL_BUFFER_SIZE", 4096);

  /**
   * counter to give connections a unique id
   */
  @MakeNotStatic
  private static final AtomicLong ID_COUNTER = new AtomicLong(1);

  /**
   * string used as the reason for initiating suspect processing
   */
  @VisibleForTesting
  public static final String INITIATING_SUSPECT_PROCESSING =
      "member unexpectedly shut down shared, unordered connection";

  /**
   * the table holding this connection
   */
  private final ConnectionTable owner;

  private final TCPConduit conduit;
  private NioFilter ioFilter;

  /**
   * Set to false once run() is terminating. Using this instead of Thread.isAlive as the reader
   * thread may be a pooled thread.
   */
  private volatile boolean isRunning;

  /**
   * true if connection is a shared resource that can be used by more than one thread
   */
  private boolean sharedResource;

  /**
   * The idle timeout timer task for this connection
   */
  private volatile SystemTimerTask idleTask;

  /**
   * If true then readers for thread owned sockets will send all messages on thread owned senders.
   * Even normally unordered msgs get send on TO socks.
   */
  private static final boolean DOMINO_THREAD_OWNED_SOCKETS =
      Boolean.getBoolean("p2p.ENABLE_DOMINO_THREAD_OWNED_SOCKETS");

  private static final ThreadLocal<Boolean> isDominoThread = withInitial(() -> FALSE);

  /**
   * the socket entrusted to this connection
   */
  private final Socket socket;

  /**
   * output stream/channel lock
   */
  private final Object outLock = new Object();

  /**
   * the ID string of the conduit (for logging)
   */
  private final String conduitIdStr;

  private InternalDistributedMember remoteMember;

  /**
   * Identifies the version of the member on the other side of the connection.
   */
  private KnownVersion remoteVersion;

  /**
   * True if this connection was accepted by a listening socket. This makes it a receiver. False if
   * this connection was explicitly created by a connect call. This makes it a sender.
   */
  private final boolean isReceiver;

  /**
   * count of how many unshared p2p-readers removed from the original action this thread is. For
   * instance, server-connection -> owned p2p reader (count 0) -> owned p2p reader (count 1) ->
   * owned p2p reader (count 2). This shows up in thread names as "DOM #x" (domino #x)
   */
  private static final ThreadLocal<Integer> dominoCount = withInitial(() -> 0);

  /**
   * How long to wait if receiver will not accept a message before we go into queue mode.
   *
   * @since GemFire 4.2.2
   */
  private int asyncDistributionTimeout;

  /**
   * How long to wait, with the receiver not accepting any messages, before kicking the receiver out
   * of the distributed system. Ignored if asyncDistributionTimeout is zero.
   *
   * @since GemFire 4.2.2
   */
  private int asyncQueueTimeout;

  /**
   * How much queued data we can have, with the receiver not accepting any messages, before kicking
   * the receiver out of the distributed system. Ignored if asyncDistributionTimeout is zero.
   * Canonicalized to bytes (property file has it as megabytes
   *
   * @since GemFire 4.2.2
   */
  private long asyncMaxQueueSize;

  /**
   * True if an async queue is already being filled.
   */
  private volatile boolean asyncQueuingInProgress;

  /**
   * Maps ConflatedKey instances to ConflatedKey instance. Note that even though the key and value
   * for an entry is the map will always be "equal" they will not always be "==".
   */
  private final Map<ConflationKey, ConflationKey> conflatedKeys = new HashMap<>();

  /**
   * NOTE: LinkedBlockingQueue has a bug in which removes from the queue
   * cause future offer to increase the size without adding anything to the queue.
   * So I've changed from this backport class to a java.util.LinkedList
   */
  private final LinkedList<Object> outgoingQueue = new LinkedList<>();

  /**
   * Number of bytes in the outgoingQueue. Used to control capacity.
   */
  private long queuedBytes;

  /** used for async writes */
  private Thread pusherThread;

  /** Set to true once the handshake has been read */
  private volatile boolean handshakeRead;
  private volatile boolean handshakeCancelled;

  private static final byte REPLY_CODE_OK = (byte) 69;
  private static final byte REPLY_CODE_OK_WITH_ASYNC_INFO = (byte) 70;

  private final Object handshakeSync = new Object();

  /** message reader thread */
  private volatile Thread readerThread;

  /** whether the reader thread is, or should be, running */
  volatile boolean stopped = true;

  /** set to true once a close begins */
  private final AtomicBoolean closing = new AtomicBoolean(false);

  private volatile boolean readerShuttingDown;

  /** whether the socket is connected */
  volatile boolean connected;

  /**
   * Set to true once a connection finishes its constructor
   */
  private volatile boolean finishedConnecting;

  private volatile boolean accessed = true;
  private volatile boolean socketInUse;
  volatile boolean timedOut;

  /**
   * task for detecting ack timeouts and issuing alerts
   */
  private volatile SystemTimer.SystemTimerTask ackTimeoutTask;

  /**
   * millisecond clock at the time message transmission started, if doing forced-disconnect
   * processing
   */
  private long transmissionStartTime;

  /** ack wait timeout - if socketInUse, use this to trigger SUSPECT processing */
  private long ackWaitTimeout;

  /** ack severe alert timeout - if socketInUse, use this to send alert */
  private long ackSATimeout;

  /**
   * other connections participating in the current transmission. we notify them if ackSATimeout
   * expires to keep all members from generating alerts when only one is slow
   */
  private List<Connection> ackConnectionGroup;

  /** name of thread that we're currently performing an operation in (may be null) */
  private String ackThreadName;

  /*
   * This object mediates access to the input ByteBuffer and ensures its return to
   * pool after last use. This reference couldn't be final since it is initialized
   * in createIoFilter() not in the constructors. It had to be initialized there
   * because in general we have to construct an SSLEngine before we know the buffer
   * size and createIoFilter() is where we create that object.
   */
  private ByteBufferVendor inputBufferVendor;

  /** the length of the next message to be dispatched */
  private int messageLength;

  /** the type of message being received */
  private byte messageType;

  /**
   * when messages are chunked by a MsgStreamer we track the destreamers on
   * the receiving side using a message identifier
   */
  private short messageId;

  /** whether the length of the next message has been established */
  private boolean lengthSet;

  /** used to lock access to destreamer data */
  private final Object destreamerLock = new Object();

  /** caches a msg destreamer that is currently not being used */
  private MsgDestreamer idleMsgDestreamer;

  /**
   * used to map a msgId to a MsgDestreamer which are used for destreaming chunked messages
   */
  private HashMap<Short, MsgDestreamer> destreamerMap;

  private boolean directAck;

  private boolean asyncMode;

  /**
   * Is this connection used for serial message delivery?
   * May be mutated during {@link #readHandshakeForReceiver(DataInput)}
   */
  private boolean preserveOrder;

  /** number of messages sent on this connection */
  private long messagesSent;

  /** number of messages received on this connection */
  private long messagesReceived;

  /** unique ID of this connection (remote if isReceiver==true) */
  private volatile long uniqueId;

  private int sendBufferSize = -1;
  private int recvBufferSize = -1;

  @MakeNotStatic
  private static final ByteBuffer okHandshakeBuf;
  static {
    int msglen = 1; // one byte for reply code
    byte[] bytes = new byte[MSG_HEADER_BYTES + msglen];
    msglen = calcHdrSize(msglen);
    bytes[MSG_HEADER_SIZE_OFFSET] = (byte) (msglen / 0x1000000 & 0xff);
    bytes[MSG_HEADER_SIZE_OFFSET + 1] = (byte) (msglen / 0x10000 & 0xff);
    bytes[MSG_HEADER_SIZE_OFFSET + 2] = (byte) (msglen / 0x100 & 0xff);
    bytes[MSG_HEADER_SIZE_OFFSET + 3] = (byte) (msglen & 0xff);
    bytes[MSG_HEADER_TYPE_OFFSET] = (byte) NORMAL_MSG_TYPE; // message type
    bytes[MSG_HEADER_ID_OFFSET] = (byte) (MsgIdGenerator.NO_MSG_ID >> 8 & 0xff);
    bytes[MSG_HEADER_ID_OFFSET + 1] = (byte) (MsgIdGenerator.NO_MSG_ID & 0xff);
    bytes[MSG_HEADER_BYTES] = REPLY_CODE_OK;
    int allocSize = bytes.length;
    ByteBuffer bb;
    if (BufferPool.useDirectBuffers) {
      bb = ByteBuffer.allocateDirect(allocSize);
    } else {
      bb = ByteBuffer.allocate(allocSize);
    }
    bb.put(bytes);
    okHandshakeBuf = bb;
  }

  /**
   * maximum message buffer size
   */
  public static final int MAX_MSG_SIZE = 0x00ffffff;

  private static final int HANDSHAKE_TIMEOUT_MS =
      Integer.getInteger("p2p.handshakeTimeoutMs", 59000);

  /**
   * product version exchange during handshake.
   *
   * NOTICE: handshake_version should not be changed anymore. Use the gemfire version transmitted
   * with the handshake bits and handle old handshakes based on that
   */
  static final byte HANDSHAKE_VERSION = 7;

  private final AtomicBoolean asyncCloseCalled = new AtomicBoolean();

  private static final int CONNECT_HANDSHAKE_SIZE = 4096;

  /** time between connection attempts */
  private static final int RECONNECT_WAIT_TIME =
      Integer.getInteger(GEMFIRE_PREFIX + "RECONNECT_WAIT_TIME", 2000);

  /**
   * Batch sends currently should not be turned on because: 1. They will be used for all sends
   * (instead of just no-ack) and thus will break messages that wait for a response (or kill perf).
   * 2. The buffer is not properly flushed and closed on shutdown. The code attempts to do this but
   * must not be doing it correctly.
   */
  private static final boolean BATCH_SENDS = Boolean.getBoolean("p2p.batchSends");
  private static final int BATCH_BUFFER_SIZE =
      Integer.getInteger("p2p.batchBufferSize", 1024 * 1024);
  private static final int BATCH_FLUSH_MS = Integer.getInteger("p2p.batchFlushTime", 50);
  private final Object batchLock = new Object();
  private ByteBuffer fillBatchBuffer;
  private ByteBuffer sendBatchBuffer;
  private BatchBufferFlusher batchFlusher;

  /**
   * use to test message prep overhead (no socket write). WARNING: turning this on completely
   * disables distribution of batched sends
   */
  private static final boolean SOCKET_WRITE_DISABLED = Boolean.getBoolean("p2p.disableSocketWrite");

  private final Object pusherSync = new Object();

  private boolean disconnectRequested;

  /**
   * If true then act as if the socket buffer is full and start async queuing
   */
  @MutableForTesting
  public static volatile boolean FORCE_ASYNC_QUEUE;

  private static final int MAX_WAIT_TIME = 32; // ms (must be a power of 2)

  /**
   * stateLock is used to synchronize state changes.
   */
  private final Object stateLock = new Object();

  /**
   * for timeout processing, this is the current state of the connection
   */
  private byte connectionState = STATE_IDLE;

  /* ~~~~~~~~~~~~~ connection states ~~~~~~~~~~~~~~~ */
  /** the connection is idle, but may be in use */
  private static final byte STATE_IDLE = 0;
  /** the connection is in use and is transmitting data */
  private static final byte STATE_SENDING = 1;
  /** the connection is in use and is done transmitting */
  private static final byte STATE_POST_SENDING = 2;
  /** the connection is in use and is reading a direct-ack */
  private static final byte STATE_READING_ACK = 3;
  /** the connection is in use and has finished reading a direct-ack */
  private static final byte STATE_RECEIVED_ACK = 4;
  /** the connection is in use and is reading a message */
  private static final byte STATE_READING = 5;
  /* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ */

  /** set to true if we exceeded the ack-wait-threshold waiting for a response */
  private volatile boolean ackTimedOut;

  /**
   * a Reader thread for an shared Connection will remain around in order to
   * ensure that the socket is properly closed.
   */
  private volatile boolean hasResidualReaderThread;

  /**
   * creates a "reader" connection that we accepted (it was initiated by an explicit connect being
   * done on the other side).
   */
  protected Connection(ConnectionTable connectionTable, Socket socket) throws ConnectionException {
    if (connectionTable == null) {
      throw new IllegalArgumentException("Null ConnectionTable");
    }
    conduit = connectionTable.getConduit();
    isReceiver = true;
    owner = connectionTable;
    this.socket = socket;
    InetSocketAddress conduitSocketId = conduit.getSocketId();
    conduitIdStr = conduitSocketId.toString();
    handshakeRead = false;
    handshakeCancelled = false;
    connected = true;
    asyncMode = false;

    try {
      socket.setTcpNoDelay(true);
      socket.setKeepAlive(true);
      setSendBufferSize(socket, SMALL_BUFFER_SIZE);
      setReceiveBufferSize(socket);
    } catch (SocketException e) {
      // unable to get the settings we want. Don't log an error because it will likely happen a lot
    }
  }

  private ThreadsMonitoring getThreadMonitoring() {
    return conduit.getDM().getThreadMonitoring();
  }

  public boolean isSharedResource() {
    return sharedResource;
  }

  @VisibleForTesting
  int getP2PConnectTimeout(DistributionConfig config) {
    if (AlertingAction.isThreadAlerting()) {
      return config.getMemberTimeout();
    }
    if (IS_P2P_CONNECT_TIMEOUT_INITIALIZED) {
      return P2P_CONNECT_TIMEOUT;
    }
    String connectTimeoutStr = System.getProperty("p2p.connectTimeout");
    if (connectTimeoutStr != null) {
      P2P_CONNECT_TIMEOUT = Integer.parseInt(connectTimeoutStr);
    } else {
      P2P_CONNECT_TIMEOUT = 6 * config.getMemberTimeout();
    }
    IS_P2P_CONNECT_TIMEOUT_INITIALIZED = true;
    return P2P_CONNECT_TIMEOUT;
  }

  @TestOnly
  static void clearP2PConnectTimeout() {
    IS_P2P_CONNECT_TIMEOUT_INITIALIZED = false;
    P2P_CONNECT_TIMEOUT = 0;
  }

  /**
   * @return true if this thread is a reader thread, otherwise false
   */
  private static boolean tipDomino() {
    if (DOMINO_THREAD_OWNED_SOCKETS) {
      // mark this thread as one who wants to send ALL on TO sockets
      ConnectionTable.threadWantsOwnResources();
      isDominoThread.set(TRUE);
      return true;
    }
    return false;
  }

  public static boolean isDominoThread() {
    return isDominoThread.get();
  }

  private void setSendBufferSize(Socket sock) {
    setSendBufferSize(sock, owner.getConduit().tcpBufferSize);
  }

  private void setReceiveBufferSize(Socket sock) {
    setReceiveBufferSize(sock, owner.getConduit().tcpBufferSize);
  }

  private void setSendBufferSize(Socket sock, int requestedSize) {
    setSocketBufferSize(sock, true, requestedSize);
  }

  private void setReceiveBufferSize(Socket sock, int requestedSize) {
    setSocketBufferSize(sock, false, requestedSize);
  }

  private void setSocketBufferSize(Socket sock, boolean send, int requestedSize) {
    if (requestedSize > 0) {
      try {
        int currentSize = send ? sock.getSendBufferSize() : sock.getReceiveBufferSize();
        if (currentSize == requestedSize) {
          if (send) {
            sendBufferSize = currentSize;
          }
          return;
        }
        if (send) {
          sock.setSendBufferSize(requestedSize);
        } else {
          sock.setReceiveBufferSize(requestedSize);
        }
      } catch (SocketException ignore) {
      }
      try {
        int actualSize = send ? sock.getSendBufferSize() : sock.getReceiveBufferSize();
        if (send) {
          sendBufferSize = actualSize;
        } else {
          recvBufferSize = actualSize;
        }
        if (actualSize < requestedSize) {
          logger.info("Socket {} is {} instead of the requested {}.",
              send ? "send buffer size" : "receive buffer size",
              actualSize, requestedSize);
        } else if (actualSize > requestedSize) {
          if (logger.isTraceEnabled()) {
            logger.trace("Socket {} buffer size is {} instead of the requested {}",
                send ? "send" : "receive", actualSize, requestedSize);
          }
          // Remember the request size which is smaller.
          // This remembered value is used for allocating direct mem buffers.
          if (send) {
            sendBufferSize = requestedSize;
          } else {
            recvBufferSize = requestedSize;
          }
        }
      } catch (SocketException ignore) {
        if (send) {
          sendBufferSize = requestedSize;
        } else {
          recvBufferSize = requestedSize;
        }
      }
    }
  }

  /**
   * Returns the size of the send buffer on this connection's socket.
   */
  int getSendBufferSize() {
    int result = sendBufferSize;
    if (result != -1) {
      return result;
    }
    try {
      result = getSocket().getSendBufferSize();
    } catch (SocketException ignore) {
      // just return a default
      result = owner.getConduit().tcpBufferSize;
    }
    sendBufferSize = result;
    return result;
  }

  void initReceiver() {
    startReader(owner);
  }

  void setIdleTimeoutTask(SystemTimerTask task) {
    idleTask = task;
  }

  /**
   * Returns true if an idle connection was detected.
   */
  boolean checkForIdleTimeout() {
    if (isSocketClosed()) {
      return true;
    }
    if (isSocketInUse() || sharedResource && !preserveOrder) {
      // shared/unordered connections are used for failure-detection
      // and are not subject to idle-timeout
      return false;
    }
    boolean isIdle = !accessed;
    accessed = false;
    if (isIdle) {
      timedOut = true;
      owner.getConduit().getStats().incLostLease();
      if (logger.isDebugEnabled()) {
        logger.debug("Closing idle connection {} shared={} ordered={}", this, sharedResource,
            preserveOrder);
      }
      try {
        // Instead of calling requestClose we call closeForReconnect.
        // We don't want this timeout close to close any other connections.
        // The problem with requestClose has removeEndpoint set to true
        // which will close an receivers we have if this connection is a shared one.
        closeForReconnect("idle connection timed out");
      } catch (Exception ignore) {
      }
    }
    return isIdle;
  }

  static int calcHdrSize(int byteSize) {
    if (byteSize > MAX_MSG_SIZE) {
      throw new IllegalStateException(format("tcp message exceeded max size of %s",
          MAX_MSG_SIZE));
    }
    int hdrSize = byteSize;
    hdrSize |= HANDSHAKE_VERSION << 24;
    return hdrSize;
  }

  static int calcMsgByteSize(int hdrSize) {
    return hdrSize & MAX_MSG_SIZE;
  }

  static void calcHdrVersion(int hdrSize) throws IOException {
    byte ver = (byte) (hdrSize >> 24);
    if (ver != HANDSHAKE_VERSION) {
      throw new IOException(
          format(
              "Detected wrong version of GemFire product during handshake. Expected %s but found %s",
              HANDSHAKE_VERSION, ver));
    }
  }

  private void sendOKHandshakeReply() throws IOException, ConnectionException {
    ByteBuffer my_okHandshakeBuf;
    if (isReceiver) {
      DistributionConfig cfg = owner.getConduit().getConfig();
      ByteBuffer bb;
      if (BufferPool.useDirectBuffers) {
        bb = ByteBuffer.allocateDirect(128);
      } else {
        bb = ByteBuffer.allocate(128);
      }
      bb.putInt(0); // reserve first 4 bytes for packet length
      bb.put((byte) NORMAL_MSG_TYPE);
      bb.putShort(MsgIdGenerator.NO_MSG_ID);
      bb.put(REPLY_CODE_OK_WITH_ASYNC_INFO);
      bb.putInt(cfg.getAsyncDistributionTimeout());
      bb.putInt(cfg.getAsyncQueueTimeout());
      bb.putInt(cfg.getAsyncMaxQueueSize());
      // write own product version
      VersioningIO.writeOrdinal(bb, KnownVersion.CURRENT.ordinal(), true);
      // now set the msg length into position 0
      bb.putInt(0, calcHdrSize(bb.position() - MSG_HEADER_BYTES));
      my_okHandshakeBuf = bb;
      bb.flip();
    } else {
      my_okHandshakeBuf = okHandshakeBuf;
    }
    my_okHandshakeBuf.position(0);
    writeFully(getSocket().getChannel(), my_okHandshakeBuf, false, null);
  }

  /**
   * @throws ConnectionException if the conduit has stopped
   */
  private void waitForHandshake() throws ConnectionException {
    boolean needToClose = false;
    String reason = null;
    try {
      synchronized (handshakeSync) {
        if (!handshakeRead && !handshakeCancelled) {
          reason = "unknown";
          boolean interrupted = Thread.interrupted();
          boolean success = false;
          try {
            final long endTime = System.currentTimeMillis() + HANDSHAKE_TIMEOUT_MS;
            long msToWait = HANDSHAKE_TIMEOUT_MS;
            while (!handshakeRead && !handshakeCancelled && msToWait > 0) {
              handshakeSync.wait(msToWait); // spurious wakeup ok
              if (!handshakeRead && !handshakeCancelled) {
                msToWait = endTime - System.currentTimeMillis();
              }
            }
            if (!handshakeRead && !handshakeCancelled) {
              reason = "handshake timed out";
              String peerName;
              if (remoteMember != null) {
                peerName = remoteMember.toString();
                // late in the life of jdk 1.7 we started seeing connections accepted
                // when accept() was not even being called. This started causing timeouts
                // to occur in the handshake threads instead of causing failures in
                // connection-formation. So, we need to initiate suspect processing here
                owner.getDM().getDistribution().suspectMember(remoteMember,
                    format(
                        "Connection handshake with %s timed out after waiting %s milliseconds.",
                        peerName, HANDSHAKE_TIMEOUT_MS));
              } else {
                peerName = "socket " + socket.getRemoteSocketAddress() + ":" + socket.getPort();
              }
              throw new ConnectionException(
                  format(
                      "Connection handshake with %s timed out after waiting %s milliseconds.",
                      peerName, HANDSHAKE_TIMEOUT_MS));
            }
            success = handshakeRead;
          } catch (InterruptedException ex) {
            interrupted = true;
            owner.getConduit().getCancelCriterion().checkCancelInProgress(ex);
            reason = "interrupted";
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
            if (success) {
              if (isReceiver) {
                needToClose =
                    !owner.getConduit().getMembership().addSurpriseMember(remoteMember);
                if (needToClose) {
                  reason = "this member is shunned";
                }
              }
            } else {
              needToClose = true;
            }
          }
        }
      }

    } finally {
      if (needToClose) {
        try {
          requestClose(reason);
        } catch (Exception ignore) {
        }
      }
    }
  }

  @VisibleForTesting
  void clearSSLInputBuffer() {
    if (getConduit().useSSL() && ioFilter != null) {
      try (final ByteBufferSharing sharedBuffer = ioFilter.getUnwrappedBuffer()) {
        // clear out any remaining handshake bytes
        sharedBuffer.getBuffer().position(0).limit(0);
      } catch (IOException e) {
        // means the NioFilter was already closed
      }
    }
  }

  @VisibleForTesting
  void notifyHandshakeWaiter(boolean success) {
    synchronized (handshakeSync) {
      /*
       * Return early to avoid modifying ioFilter's buffer more than once.
       */
      if (handshakeRead || handshakeCancelled) {
        return;
      }
      clearSSLInputBuffer();
      if (success) {
        handshakeRead = true;
      } else {
        handshakeCancelled = true;
      }
      handshakeSync.notifyAll();
    }
  }

  /**
   * asynchronously close this connection
   *
   * @param beingSickForTests test hook to simulate sickness in communications & membership
   */
  private void asyncClose(boolean beingSickForTests) {
    // note: remoteAddr may be null if this is a receiver that hasn't finished its handshake

    // we do the close in a background thread because the operation may hang if
    // there is a problem with the network

    // if simulating sickness, sockets must be closed in-line so that tests know
    // that the vm is sick when the beSick operation completes
    if (beingSickForTests) {
      prepareForAsyncClose();
    } else {
      if (asyncCloseCalled.compareAndSet(false, true)) {
        Socket s = socket;
        if (s != null && !s.isClosed()) {
          prepareForAsyncClose();
          owner.getSocketCloser().asyncClose(s, String.valueOf(remoteMember),
              () -> ioFilter.close(s.getChannel()));
        }
      }
    }
  }

  private void prepareForAsyncClose() {
    synchronized (stateLock) {
      if (readerThread != null && isRunning && !readerShuttingDown
          && (connectionState == STATE_READING || connectionState == STATE_READING_ACK)) {
        readerThread.interrupt();
      }
    }
  }

  /**
   * waits until we've joined the distributed system before returning
   */
  private void waitForAddressCompletion() {
    InternalDistributedMember myAddr = owner.getConduit().getMemberId();
    synchronized (myAddr) {
      while (!owner.getConduit().getCancelCriterion().isCancelInProgress()
          && myAddr.getInetAddress() == null && myAddr.getVmViewId() < 0) {
        try {
          myAddr.wait(100); // spurious wakeup ok
        } catch (InterruptedException ie) {
          Thread.currentThread().interrupt();
          owner.getConduit().getCancelCriterion().checkCancelInProgress(ie);
        }
      }
      Assert.assertTrue(myAddr.getDirectChannelPort() == owner.getConduit().getPort());
    }
  }

  private void handshakeFromNewSender() throws IOException {
    waitForAddressCompletion();

    InternalDistributedMember myAddr = owner.getConduit().getMemberId();
