cppcache/src/TcrConnection.hpp - geode-native - Git at Google

 /*
  * 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.
  */

 #pragma once

 #ifndef GEODE_TCRCONNECTION_H_
 #define GEODE_TCRCONNECTION_H_

 #include <atomic>
 #include <chrono>

 #include <ace/Semaphore.h>

 #include <geode/CacheableBuiltins.hpp>
 #include <geode/ExceptionTypes.hpp>
 #include <geode/internal/geode_globals.hpp>

 #include "Connector.hpp"
 #include "DiffieHellman.hpp"
 #include "TcrMessage.hpp"
 #include "util/synchronized_set.hpp"

 #define DEFAULT_TIMEOUT_RETRIES 12
 #define PRIMARY_SERVER_TO_CLIENT 101
 #define SECONDARY_SERVER_TO_CLIENT 102
 #define SUCCESSFUL_SERVER_TO_CLIENT 105
 #define UNSUCCESSFUL_SERVER_TO_CLIENT 106
 #define CLIENT_TO_SERVER 100
 #define REPLY_OK 59
 #define REPLY_OK_CS43 58
 #define REPLY_REFUSED 60
 #define REPLY_INVALID 61
 #define REPLY_SSL_ENABLED 21
 #define REPLY_AUTHENTICATION_REQUIRED 62
 #define REPLY_AUTHENTICATION_FAILED 63
 #define REPLY_DUPLICATE_DURABLE_CLIENT 64

 #define SECURITY_CREDENTIALS_NONE 0
 #define SECURITY_CREDENTIALS_NORMAL 1
 #define SECURITY_CREDENTIALS_DHENCRYPT 2
 #define SECURITY_MULTIUSER_NOTIFICATIONCHANNEL 3

 /** Closes and Deletes connection only if it exists */
 #define GF_SAFE_DELETE_CON(x) \
   do {                        \
     x->close();               \
     delete x;                 \
     x = nullptr;              \
   } while (0)

 namespace apache {
 namespace geode {
 namespace client {

 struct chunkHeader {
   int32_t chunkLength;
   int8_t flags;
 };

 struct chunkedResponseHeader {
   int32_t messageType;
   int32_t numberOfParts;
   int32_t transactionId;
   chunkHeader header;
 };

 enum ConnErrType {
   CONN_NOERR = 0x0,
   CONN_NODATA = 0x1,
   CONN_TIMEOUT = 0x3,
   CONN_IOERR = 0x4,
   CONN_OTHERERR = 0x8
 };

 enum ServerQueueStatus {
   NON_REDUNDANT_SERVER = 0,
   REDUNDANT_SERVER = 1,
   PRIMARY_SERVER = 2
 };

 class TcrEndpoint;
 class SystemProperties;
 class ThinClientPoolDM;
 class TcrConnectionManager;
 class APACHE_GEODE_EXPORT TcrConnection {
  public:
   using clock = std::chrono::steady_clock;
   using time_point = clock::time_point;

   /** Create one connection, endpoint is in format of hostname:portno
    * It will do handshake with j-server. There're 2 types of handshakes:
    * 1) handshake for request
    *    send following bytes:
    *    CLIENT_TO_SERVER
    *    REPLY_OK
    *    2 bytes for the length of idenfifier
    *    a string with "hostname:processId" as identifier
    *
    *    if send succeeds, handshake succeeds. Otherwise, construction
    *    fails.
    *
    * 2) handshake for client notification
    *    send following bytes:
    *    SERVER_TO_CLIENT
    *    1 (4 bytes, we can hard-code)
    *    12345 (4 bytes, we can hard-code)
    *
    *    So the total bytes to send are 9
    *    read one byte from server, it should be CLIENT_TO_SERVER
    *    Otherwise, construction fails.
    * @param     ports     List of local ports for connections to endpoint
    * @param     numPorts  Size of ports list
    */
   bool initTcrConnection(
       TcrEndpoint* endpointObj, const char* endpoint,
       synchronized_set<std::unordered_set<uint16_t>>& ports,
       bool isClientNotification = false, bool isSecondary = false,
       std::chrono::microseconds connectTimeout = DEFAULT_CONNECT_TIMEOUT);

   TcrConnection(const TcrConnectionManager& connectionManager,
                 volatile const bool& isConnected)
       : connectionId(0),
         m_connectionManager(&connectionManager),
         m_dh(nullptr),
         m_endpoint(nullptr),
         m_endpointObj(nullptr),
         m_connected(isConnected),
         m_conn(nullptr),
         m_hasServerQueue(NON_REDUNDANT_SERVER),
         m_queueSize(0),
         m_port(0),
         m_chunksProcessSema(0),
         m_isBeingUsed(false),
         m_isUsed(0),
         m_poolDM(nullptr) {}

   /* destroy the connection */
   ~TcrConnection();

   /**
    * send a synchronized request to server.
    *
    * It will send the buffer, then wait to receive 17 bytes and save in
    * msg_header.
    * msg_header[0] is message type.
    * msg_header[1],msg_header[2],msg_header[3],msg_header[4] will be a 4 bytes
    * integer,
    * let's say, msgLen, which specifies the length of next read. byteReads some
    * number of
    * call read again for msgLen bytes, and save the bytes into msg_body.
    * concatenate the msg_header and msg_body into buffer, msg. The msg should be
    * a '0' ended
    * string. i.e. If the msg_header plus msg_body has 100 chars, msg should be a
    * 101 char array
    * to contain the '0' in the end. We need it to get length of the msg.
    * Return the msg.
    *
    * @param      buffer the buffer to send
    * @param      len length of the data to send
    * @param      sendTimeoutSec write timeout in sec
    * @param      recvLen output parameter for length of the received message
    * @param      receiveTimeoutSec read timeout in sec
    * @return     byte arrary of response. '0' ended.
    * @exception  GeodeIOException  if an I/O error occurs (socket failure).
    * @exception  TimeoutException  if timeout happens at any of the 3 socket
    * operation: 1 write, 2 read
    */
   char* sendRequest(
       const char* buffer, size_t len, size_t* recvLen,
       std::chrono::microseconds sendTimeoutSec = DEFAULT_WRITE_TIMEOUT,
       std::chrono::microseconds receiveTimeoutSec = DEFAULT_READ_TIMEOUT_SECS,
       int32_t request = -1);

   /**
    * send a synchronized request to server for REGISTER_INTEREST_LIST.
    *
    * @param      buffer the buffer to send
    *             len length of the data to send
    *             message vector, which will return chunked TcrMessage.
    *             sendTimeoutSec write timeout in sec
    *             receiveTimeoutSec read timeout in sec
    * @exception  GeodeIOException  if an I/O error occurs (socket failure).
    * @exception  TimeoutException  if timeout happens at any of the 3 socket
    * operation: 1 write, 2 read
    */
   void sendRequestForChunkedResponse(
       const TcrMessage& request, size_t len, TcrMessageReply& message,
       std::chrono::microseconds sendTimeoutSec = DEFAULT_WRITE_TIMEOUT,
       std::chrono::microseconds receiveTimeoutSec = DEFAULT_READ_TIMEOUT_SECS);

   /**
    * send an asynchronized request to server. No response is expected.
    * we need to use it to send CLOSE_CONNECTION msg
    *
    * @param      buffer the buffer to send
    *             len length of the data to send
    *             sendTimeoutSec write timeout in sec
    * @return     no return. Because it either succeeds, or throw exception.
    * @exception  GeodeIOException  if an I/O error occurs (socket failure).
    * @exception  TimeoutException  if timeout happens at any of the 3 socket
    * operation: 1 write, 2 read
    */
   void send(const char* buffer, size_t len,
             std::chrono::microseconds sendTimeoutSec = DEFAULT_WRITE_TIMEOUT,
             bool checkConnected = true);

   void send(std::chrono::microseconds& timeSpent, const char* buffer,
             size_t len,
             std::chrono::microseconds sendTimeoutSec = DEFAULT_WRITE_TIMEOUT,
             bool checkConnected = true);

   /**
    * This method is for receiving client notification. It will read 2 times as
    * reading reply in sendRequest()
    *
    * @param      recvLen output parameter for length of the received message
    * @param      receiveTimeoutSec read timeout in sec
    * @return     byte arrary of response. '0' ended.
    * @exception  GeodeIOException  if an I/O error occurs (socket failure).
    * @exception  TimeoutException  if timeout happens at any of the 3 socket
    * operation: 1 write, 2 read
    */
   char* receive(
       size_t* recvLen, ConnErrType* opErr,
       std::chrono::microseconds receiveTimeoutSec = DEFAULT_READ_TIMEOUT_SECS);

   //  readMessage is now public
   /**
    * This method reads a message from the socket connection and returns the byte
    * array of response.
    * @param      recvLen output parameter for length of the received message
    * @param      receiveTimeoutSec read timeout in seconds
    * @param      doHeaderTimeoutRetries retry when header receive times out
    * @return     byte array of response. '0' ended.
    * @exception  GeodeIOException  if an I/O error occurs (socket failure).
    * @exception  TimeoutException  if timeout happens during read
    */
   char* readMessage(size_t* recvLen,
                     std::chrono::microseconds receiveTimeoutSec,
                     bool doHeaderTimeoutRetries, ConnErrType* opErr,
                     bool isNotificationMessage = false, int32_t request = -1);

   /**
    * This method reads an interest list response  message from the socket
    * connection and sets the reply message
    * parameter.
    * @param      reply response message
    * @param      receiveTimeout read timeout
    * @param      doHeaderTimeoutRetries retry when header receive times out
    * @exception  GeodeIOException  if an I/O error occurs (socket failure).
    * @exception  TimeoutException  if timeout happens during read
    */
   void readMessageChunked(TcrMessageReply& reply,
                           std::chrono::microseconds receiveTimeout,
                           bool doHeaderTimeoutRetries);

   /**
    * Send close connection message to the server.
    */
   void close();

   //  Durable clients: return true if server has HA queue.
   ServerQueueStatus inline getServerQueueStatus(int32_t& queueSize) {
     queueSize = m_queueSize;
     return m_hasServerQueue;
   }

   uint16_t inline getPort() { return m_port; }

   TcrEndpoint* getEndpointObject() const { return m_endpointObj; }
   bool isBeingUsed() { return m_isBeingUsed; }
   bool setAndGetBeingUsed(
       volatile bool isBeingUsed,
       bool forTransaction);  // { m_isBeingUsed = isBeingUsed ;}

   // helpers for pool connection manager
   void touch();
   bool hasExpired(const std::chrono::milliseconds& expiryTime);
   bool isIdle(const std::chrono::milliseconds& idleTime);
   time_point getLastAccessed();
   void updateCreationTime();

   int64_t getConnectionId() {
     LOGDEBUG("TcrConnection::getConnectionId() = %" PRId64, connectionId);
     return connectionId;
   }

   void setConnectionId(int64_t id) {
     LOGDEBUG("Tcrconnection:setConnectionId() = %" PRId64, id);
     connectionId = id;
   }

   const TcrConnectionManager& getConnectionManager() {
     return *m_connectionManager;
   }

   std::shared_ptr<CacheableBytes> encryptBytes(
       std::shared_ptr<CacheableBytes> data) {
     if (m_dh != nullptr) {
       return m_dh->encrypt(data);
     } else {
       return data;
     }
   }

   std::shared_ptr<CacheableBytes> decryptBytes(
       std::shared_ptr<CacheableBytes> data) {
     if (m_dh != nullptr) {
       return m_dh->decrypt(data);
     } else {
       return data;
     }
   }

  private:
   int64_t connectionId;
   const TcrConnectionManager* m_connectionManager;
   DiffieHellman* m_dh;

   std::chrono::microseconds calculateHeaderTimeout(
       std::chrono::microseconds receiveTimeout, bool retry);

   chunkedResponseHeader readResponseHeader(std::chrono::microseconds timeout);

   chunkHeader readChunkHeader(std::chrono::microseconds timeout);

   std::vector<uint8_t> readChunkBody(std::chrono::microseconds timeout,
                                      int32_t chunkLength);

   bool processChunk(TcrMessageReply& reply, std::chrono::microseconds timeout,
                     int32_t chunkLength, int8_t lastChunkAndSecurityFlags);

   /**
    * To read Intantiator message(which meant for java client), here we are
    * ignoring it
    */
   void readHandshakeInstantiatorMsg(std::chrono::microseconds connectTimeout);

   /**
    * Packs the override settings bits into bytes - currently a single byte for
    * conflation, remove-unresponsive-client and notify-by-subscription.
    */
   uint8_t getOverrides(const SystemProperties* props);

   /**
    * To read the from stream
    */
   int32_t readHandShakeInt(std::chrono::microseconds connectTimeout);

   /*
    * To read the arraysize
    */
   int32_t readHandshakeArraySize(std::chrono::microseconds connectTimeout);

   /*
    * This function reads "numberOfBytes" and ignores it.
    */
   void readHandShakeBytes(int numberOfBytes,
                           std::chrono::microseconds connectTimeout);

   /** Create a normal or SSL connection */
   Connector* createConnection(
       const char* ipaddr,
       std::chrono::microseconds waitSeconds = DEFAULT_CONNECT_TIMEOUT,
       int32_t maxBuffSizePool = 0);

   /**
    * Reads bytes from socket and handles error conditions in case of Handshake.
    */
   std::vector<int8_t> readHandshakeData(
       int32_t msgLength, std::chrono::microseconds connectTimeout);

   /**
    * Reads raw bytes (without appending nullptr terminator) from socket and
    * handles error conditions in case of Handshake.
    */
   std::shared_ptr<CacheableBytes> readHandshakeRawData(
       int32_t msgLength, std::chrono::microseconds connectTimeout);
   /**
    * Reads a string from socket and handles error conditions in case of
    * Handshake.
    */
   std::shared_ptr<CacheableString> readHandshakeString(
       std::chrono::microseconds connectTimeout);

   /**
    * Reads a byte array (using initial length) from socket and handles error
    * conditions in case of Handshake.
    */
   std::shared_ptr<CacheableBytes> readHandshakeByteArray(
       std::chrono::microseconds connectTimeout);

   /**
    * Send data to the connection till sendTimeout
    */
   ConnErrType sendData(const char* buffer, size_t length,
                        std::chrono::microseconds sendTimeout,
                        bool checkConnected = true);

   ConnErrType sendData(std::chrono::microseconds& timeSpent, const char* buffer,
                        size_t length, std::chrono::microseconds sendTimeout,
                        bool checkConnected = true);

   /**
    * Read data from the connection till receiveTimeoutSec
    */
   ConnErrType receiveData(char* buffer, size_t length,
                           std::chrono::microseconds receiveTimeoutSec,
                           bool checkConnected = true,
                           bool isNotificationMessage = false);

   const char* m_endpoint;
   TcrEndpoint* m_endpointObj;
   volatile const bool& m_connected;
   Connector* m_conn;
   ServerQueueStatus m_hasServerQueue;
   int32_t m_queueSize;
   uint16_t m_port;

   // semaphore to synchronize with the chunked response processing thread
   ACE_Semaphore m_chunksProcessSema;

   time_point m_creationTime;
   time_point m_lastAccessed;

   // Disallow copy constructor and assignment operator.
   TcrConnection(const TcrConnection&);
   TcrConnection& operator=(const TcrConnection&);
   volatile bool m_isBeingUsed;
   std::atomic<uint32_t> m_isUsed;
   ThinClientPoolDM* m_poolDM;
   bool useReplyTimeout(const TcrMessage& request) const;
   std::chrono::microseconds sendWithTimeouts(
       const char* data, size_t len, std::chrono::microseconds sendTimeout,
       std::chrono::microseconds receiveTimeout);
   bool replyHasResult(const TcrMessage& request, TcrMessageReply& reply);
 };
 }  // namespace client
 }  // namespace geode
 }  // namespace apache

 #endif  // GEODE_TCRCONNECTION_H_
	/*
	* 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.
	*/

	#pragma once

	#ifndef GEODE_TCRCONNECTION_H_
	#define GEODE_TCRCONNECTION_H_

	#include <atomic>
	#include <chrono>

	#include <ace/Semaphore.h>

	#include <geode/CacheableBuiltins.hpp>
	#include <geode/ExceptionTypes.hpp>
	#include <geode/internal/geode_globals.hpp>

	#include "Connector.hpp"
	#include "DiffieHellman.hpp"
	#include "TcrMessage.hpp"
	#include "util/synchronized_set.hpp"

	#define DEFAULT_TIMEOUT_RETRIES 12
	#define PRIMARY_SERVER_TO_CLIENT 101
	#define SECONDARY_SERVER_TO_CLIENT 102
	#define SUCCESSFUL_SERVER_TO_CLIENT 105
	#define UNSUCCESSFUL_SERVER_TO_CLIENT 106
	#define CLIENT_TO_SERVER 100
	#define REPLY_OK 59
	#define REPLY_OK_CS43 58
	#define REPLY_REFUSED 60
	#define REPLY_INVALID 61
	#define REPLY_SSL_ENABLED 21
	#define REPLY_AUTHENTICATION_REQUIRED 62
	#define REPLY_AUTHENTICATION_FAILED 63
	#define REPLY_DUPLICATE_DURABLE_CLIENT 64

	#define SECURITY_CREDENTIALS_NONE 0
	#define SECURITY_CREDENTIALS_NORMAL 1
	#define SECURITY_CREDENTIALS_DHENCRYPT 2
	#define SECURITY_MULTIUSER_NOTIFICATIONCHANNEL 3

	/** Closes and Deletes connection only if it exists */
	#define GF_SAFE_DELETE_CON(x) \
	do { \
	x->close(); \
	delete x; \
	x = nullptr; \
	} while (0)

	namespace apache {
	namespace geode {
	namespace client {

	struct chunkHeader {
	int32_t chunkLength;
	int8_t flags;
	};

	struct chunkedResponseHeader {
	int32_t messageType;
	int32_t numberOfParts;
	int32_t transactionId;
	chunkHeader header;
	};

	enum ConnErrType {
	CONN_NOERR = 0x0,
	CONN_NODATA = 0x1,
	CONN_TIMEOUT = 0x3,
	CONN_IOERR = 0x4,
	CONN_OTHERERR = 0x8
	};

	enum ServerQueueStatus {
	NON_REDUNDANT_SERVER = 0,
	REDUNDANT_SERVER = 1,
	PRIMARY_SERVER = 2
	};

	class TcrEndpoint;
	class SystemProperties;
	class ThinClientPoolDM;
	class TcrConnectionManager;
	class APACHE_GEODE_EXPORT TcrConnection {
	public:
	using clock = std::chrono::steady_clock;
	using time_point = clock::time_point;

	/** Create one connection, endpoint is in format of hostname:portno
	* It will do handshake with j-server. There're 2 types of handshakes:
	* 1) handshake for request
	* send following bytes:
	* CLIENT_TO_SERVER
	* REPLY_OK
	* 2 bytes for the length of idenfifier
	* a string with "hostname:processId" as identifier
	*
	* if send succeeds, handshake succeeds. Otherwise, construction
	* fails.
	*
	* 2) handshake for client notification
	* send following bytes:
	* SERVER_TO_CLIENT
	* 1 (4 bytes, we can hard-code)
	* 12345 (4 bytes, we can hard-code)
	*
	* So the total bytes to send are 9
	* read one byte from server, it should be CLIENT_TO_SERVER
	* Otherwise, construction fails.
	* @param ports List of local ports for connections to endpoint
	* @param numPorts Size of ports list
	*/
	bool initTcrConnection(
	TcrEndpoint* endpointObj, const char* endpoint,
	synchronized_set<std::unordered_set<uint16_t>>& ports,
	bool isClientNotification = false, bool isSecondary = false,
	std::chrono::microseconds connectTimeout = DEFAULT_CONNECT_TIMEOUT);

	TcrConnection(const TcrConnectionManager& connectionManager,
	volatile const bool& isConnected)
	: connectionId(0),
	m_connectionManager(&connectionManager),
	m_dh(nullptr),
	m_endpoint(nullptr),
	m_endpointObj(nullptr),
	m_connected(isConnected),
	m_conn(nullptr),
	m_hasServerQueue(NON_REDUNDANT_SERVER),
	m_queueSize(0),
	m_port(0),
	m_chunksProcessSema(0),
	m_isBeingUsed(false),
	m_isUsed(0),
	m_poolDM(nullptr) {}

	/* destroy the connection */
	~TcrConnection();

	/**
	* send a synchronized request to server.
	*
	* It will send the buffer, then wait to receive 17 bytes and save in
	* msg_header.
	* msg_header[0] is message type.
	* msg_header[1],msg_header[2],msg_header[3],msg_header[4] will be a 4 bytes
	* integer,
	* let's say, msgLen, which specifies the length of next read. byteReads some
	* number of
	* call read again for msgLen bytes, and save the bytes into msg_body.
	* concatenate the msg_header and msg_body into buffer, msg. The msg should be
	* a '0' ended
	* string. i.e. If the msg_header plus msg_body has 100 chars, msg should be a
	* 101 char array
	* to contain the '0' in the end. We need it to get length of the msg.
	* Return the msg.
	*
	* @param buffer the buffer to send
	* @param len length of the data to send
	* @param sendTimeoutSec write timeout in sec
	* @param recvLen output parameter for length of the received message
	* @param receiveTimeoutSec read timeout in sec
	* @return byte arrary of response. '0' ended.
	* @exception GeodeIOException if an I/O error occurs (socket failure).
	* @exception TimeoutException if timeout happens at any of the 3 socket
	* operation: 1 write, 2 read
	*/
	char* sendRequest(
	const char* buffer, size_t len, size_t* recvLen,
	std::chrono::microseconds sendTimeoutSec = DEFAULT_WRITE_TIMEOUT,
	std::chrono::microseconds receiveTimeoutSec = DEFAULT_READ_TIMEOUT_SECS,
	int32_t request = -1);

	/**
	* send a synchronized request to server for REGISTER_INTEREST_LIST.
	*
	* @param buffer the buffer to send
	* len length of the data to send
	* message vector, which will return chunked TcrMessage.
	* sendTimeoutSec write timeout in sec
	* receiveTimeoutSec read timeout in sec
	* @exception GeodeIOException if an I/O error occurs (socket failure).
	* @exception TimeoutException if timeout happens at any of the 3 socket
	* operation: 1 write, 2 read
	*/
	void sendRequestForChunkedResponse(
	const TcrMessage& request, size_t len, TcrMessageReply& message,
	std::chrono::microseconds sendTimeoutSec = DEFAULT_WRITE_TIMEOUT,
	std::chrono::microseconds receiveTimeoutSec = DEFAULT_READ_TIMEOUT_SECS);

	/**
	* send an asynchronized request to server. No response is expected.
	* we need to use it to send CLOSE_CONNECTION msg
	*
	* @param buffer the buffer to send
	* len length of the data to send
	* sendTimeoutSec write timeout in sec
	* @return no return. Because it either succeeds, or throw exception.
	* @exception GeodeIOException if an I/O error occurs (socket failure).
	* @exception TimeoutException if timeout happens at any of the 3 socket
	* operation: 1 write, 2 read
	*/
	void send(const char* buffer, size_t len,
	std::chrono::microseconds sendTimeoutSec = DEFAULT_WRITE_TIMEOUT,
	bool checkConnected = true);

	void send(std::chrono::microseconds& timeSpent, const char* buffer,
	size_t len,
	std::chrono::microseconds sendTimeoutSec = DEFAULT_WRITE_TIMEOUT,
	bool checkConnected = true);

	/**
	* This method is for receiving client notification. It will read 2 times as
	* reading reply in sendRequest()
	*
	* @param recvLen output parameter for length of the received message
	* @param receiveTimeoutSec read timeout in sec
	* @return byte arrary of response. '0' ended.
	* @exception GeodeIOException if an I/O error occurs (socket failure).
	* @exception TimeoutException if timeout happens at any of the 3 socket
	* operation: 1 write, 2 read
	*/
	char* receive(
	size_t* recvLen, ConnErrType* opErr,
	std::chrono::microseconds receiveTimeoutSec = DEFAULT_READ_TIMEOUT_SECS);

	// readMessage is now public
	/**
	* This method reads a message from the socket connection and returns the byte
	* array of response.
	* @param recvLen output parameter for length of the received message
	* @param receiveTimeoutSec read timeout in seconds
	* @param doHeaderTimeoutRetries retry when header receive times out
	* @return byte array of response. '0' ended.
	* @exception GeodeIOException if an I/O error occurs (socket failure).
	* @exception TimeoutException if timeout happens during read
	*/
	char* readMessage(size_t* recvLen,
	std::chrono::microseconds receiveTimeoutSec,
	bool doHeaderTimeoutRetries, ConnErrType* opErr,
	bool isNotificationMessage = false, int32_t request = -1);

	/**
	* This method reads an interest list response message from the socket
	* connection and sets the reply message
	* parameter.
	* @param reply response message
	* @param receiveTimeout read timeout
	* @param doHeaderTimeoutRetries retry when header receive times out
	* @exception GeodeIOException if an I/O error occurs (socket failure).
	* @exception TimeoutException if timeout happens during read
	*/
	void readMessageChunked(TcrMessageReply& reply,
	std::chrono::microseconds receiveTimeout,
	bool doHeaderTimeoutRetries);

	/**
	* Send close connection message to the server.
	*/
	void close();

	// Durable clients: return true if server has HA queue.
	ServerQueueStatus inline getServerQueueStatus(int32_t& queueSize) {
	queueSize = m_queueSize;
	return m_hasServerQueue;
	}

	uint16_t inline getPort() { return m_port; }

	TcrEndpoint* getEndpointObject() const { return m_endpointObj; }
	bool isBeingUsed() { return m_isBeingUsed; }
	bool setAndGetBeingUsed(
	volatile bool isBeingUsed,
	bool forTransaction); // { m_isBeingUsed = isBeingUsed ;}

	// helpers for pool connection manager
	void touch();
	bool hasExpired(const std::chrono::milliseconds& expiryTime);
	bool isIdle(const std::chrono::milliseconds& idleTime);
	time_point getLastAccessed();
	void updateCreationTime();

	int64_t getConnectionId() {
	LOGDEBUG("TcrConnection::getConnectionId() = %" PRId64, connectionId);
	return connectionId;
	}

	void setConnectionId(int64_t id) {
	LOGDEBUG("Tcrconnection:setConnectionId() = %" PRId64, id);
	connectionId = id;
	}

	const TcrConnectionManager& getConnectionManager() {
	return *m_connectionManager;
	}

	std::shared_ptr<CacheableBytes> encryptBytes(
	std::shared_ptr<CacheableBytes> data) {
	if (m_dh != nullptr) {
	return m_dh->encrypt(data);
	} else {
	return data;
	}
	}

	std::shared_ptr<CacheableBytes> decryptBytes(
	std::shared_ptr<CacheableBytes> data) {
	if (m_dh != nullptr) {
	return m_dh->decrypt(data);
	} else {
	return data;
	}
	}

	private:
	int64_t connectionId;
	const TcrConnectionManager* m_connectionManager;
	DiffieHellman* m_dh;

	std::chrono::microseconds calculateHeaderTimeout(
	std::chrono::microseconds receiveTimeout, bool retry);

	chunkedResponseHeader readResponseHeader(std::chrono::microseconds timeout);

	chunkHeader readChunkHeader(std::chrono::microseconds timeout);

	std::vector<uint8_t> readChunkBody(std::chrono::microseconds timeout,
	int32_t chunkLength);

	bool processChunk(TcrMessageReply& reply, std::chrono::microseconds timeout,
	int32_t chunkLength, int8_t lastChunkAndSecurityFlags);

	/**
	* To read Intantiator message(which meant for java client), here we are
	* ignoring it
	*/
	void readHandshakeInstantiatorMsg(std::chrono::microseconds connectTimeout);

	/**
	* Packs the override settings bits into bytes - currently a single byte for
	* conflation, remove-unresponsive-client and notify-by-subscription.
	*/
	uint8_t getOverrides(const SystemProperties* props);

	/**
	* To read the from stream
	*/
	int32_t readHandShakeInt(std::chrono::microseconds connectTimeout);

	/*
	* To read the arraysize
	*/
	int32_t readHandshakeArraySize(std::chrono::microseconds connectTimeout);

	/*
	* This function reads "numberOfBytes" and ignores it.
	*/
	void readHandShakeBytes(int numberOfBytes,
	std::chrono::microseconds connectTimeout);

	/** Create a normal or SSL connection */
	Connector* createConnection(
	const char* ipaddr,
	std::chrono::microseconds waitSeconds = DEFAULT_CONNECT_TIMEOUT,
	int32_t maxBuffSizePool = 0);

	/**
	* Reads bytes from socket and handles error conditions in case of Handshake.
	*/
	std::vector<int8_t> readHandshakeData(
	int32_t msgLength, std::chrono::microseconds connectTimeout);

	/**
	* Reads raw bytes (without appending nullptr terminator) from socket and
	* handles error conditions in case of Handshake.
	*/
	std::shared_ptr<CacheableBytes> readHandshakeRawData(
	int32_t msgLength, std::chrono::microseconds connectTimeout);
	/**
	* Reads a string from socket and handles error conditions in case of
	* Handshake.
	*/
	std::shared_ptr<CacheableString> readHandshakeString(
	std::chrono::microseconds connectTimeout);

	/**
	* Reads a byte array (using initial length) from socket and handles error
	* conditions in case of Handshake.
	*/
	std::shared_ptr<CacheableBytes> readHandshakeByteArray(
	std::chrono::microseconds connectTimeout);

	/**
	* Send data to the connection till sendTimeout
	*/
	ConnErrType sendData(const char* buffer, size_t length,
	std::chrono::microseconds sendTimeout,
	bool checkConnected = true);

	ConnErrType sendData(std::chrono::microseconds& timeSpent, const char* buffer,
	size_t length, std::chrono::microseconds sendTimeout,
	bool checkConnected = true);

	/**
	* Read data from the connection till receiveTimeoutSec
	*/
	ConnErrType receiveData(char* buffer, size_t length,
	std::chrono::microseconds receiveTimeoutSec,
	bool checkConnected = true,
	bool isNotificationMessage = false);

	const char* m_endpoint;
	TcrEndpoint* m_endpointObj;
	volatile const bool& m_connected;
	Connector* m_conn;
	ServerQueueStatus m_hasServerQueue;
	int32_t m_queueSize;
	uint16_t m_port;

	// semaphore to synchronize with the chunked response processing thread
	ACE_Semaphore m_chunksProcessSema;

	time_point m_creationTime;
	time_point m_lastAccessed;

	// Disallow copy constructor and assignment operator.
	TcrConnection(const TcrConnection&);
	TcrConnection& operator=(const TcrConnection&);
	volatile bool m_isBeingUsed;
	std::atomic<uint32_t> m_isUsed;
	ThinClientPoolDM* m_poolDM;
	bool useReplyTimeout(const TcrMessage& request) const;
	std::chrono::microseconds sendWithTimeouts(
	const char* data, size_t len, std::chrono::microseconds sendTimeout,
	std::chrono::microseconds receiveTimeout);
	bool replyHasResult(const TcrMessage& request, TcrMessageReply& reply);
	};
	} // namespace client
	} // namespace geode
	} // namespace apache

	#endif // GEODE_TCRCONNECTION_H_