/***********************************************************************
 * 
 *  Copyright (C) 2006 Novell, Inc. All Rights Reserved.
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; version 2.1
 *  of the License.
 *
 *  This library is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 *  Library Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, Novell, Inc.
 * 
 *  To contact Novell about this file by physical or electronic mail, 
 *  you may find current contact information at www.novell.com.
 * 
 *  Author: Juan Carlos Luciani <jluciani@novell.com>
 *
 ***********************************************************************/


//===[ Include files ]=====================================================

#include "ipcint.h"
#include "cchannel.h"
#include "clientreq.h"

//===[ External data ]=====================================================

//===[ External prototypes ]===============================================

//===[ Manifest constants ]================================================

//
// Socket Mapping definitions
//
#define INVALID_SOCKET -1
#define SOCKET_ERROR -1
#define LINGER struct linger
#define SOCKADDR_IN struct sockaddr_in
#define closesocket close

//===[ Type definitions ]==================================================

//===[ Function prototypes ]===============================================

//===[ Global variables ]==================================================

//
// Object Counters
//
unsigned long numCChannelObjects = 0;


//===[ Type definitions ]==================================================

//===[ Function prototypes ]===============================================

//===[ Global variables ]==================================================


//++=======================================================================
CChannel::CChannel(
   struct sockaddr_in *pRemoteAddress) :

   m_state (State_Uninitialized),
   m_socket (INVALID_SOCKET),
   m_reqIdAllocator (1)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   DbgTrace(1, "CChannel::CChannel(TcpSocket)- Start, Obj = %08X\n", this);

   // Use Tcp socket
   m_useTcpSocket = true;

   // Save a copy of the remote address
   memcpy(&m_remoteAddrIn, pRemoteAddress, sizeof(struct sockaddr_in));

   // Initialize the mutex
   if (pthread_mutex_init(&m_mutex, NULL) != 0)
   {
      DbgTrace(0, "CChannel::CChannel- Mutex initialization failed\n", 0);

      // Throw exception
      throw bad_alloc();
   }

   // Increment the object count
   InterlockedIncrement(&numCChannelObjects);

   DbgTrace(1, "CChannel::CChannel(TcpSocket)- End\n", 0);

}  /*-- CChannel::CChannel(TcpSocket) --*/


//++=======================================================================
CChannel::CChannel(
   struct sockaddr_un *pRemoteAddress) :

   m_state (State_Uninitialized),
   m_socket (INVALID_SOCKET),
   m_reqIdAllocator (1)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   DbgTrace(1, "CChannel::CChannel(DomainSocket)- Start, Obj = %08X\n", this);

   // Do not use Tcp socket
   m_useTcpSocket = false;

   // Save a copy of the remote address
   memcpy(&m_remoteAddrUn, pRemoteAddress, sizeof(struct sockaddr_un));

   // Initialize the mutex
   if (pthread_mutex_init(&m_mutex, NULL) != 0)
   {
      DbgTrace(0, "CChannel::CChannel- Mutex initialization failed\n", 0);

      // Throw exception
      throw bad_alloc();
   }

   // Increment the object count
   InterlockedIncrement(&numCChannelObjects);

   DbgTrace(1, "CChannel::CChannel(DomainSocket)- End\n", 0);

}  /*-- CChannel::CChannel(DomainSocket) --*/


//++=======================================================================
CChannel::~CChannel(void)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   DbgTrace(1, "CChannel::~CChannel- Start, Obj = %08X\n", this);

   // Cleanup resources allocated for the object
   pthread_mutex_destroy(&m_mutex);

   // Free connection socket if necessary
   if (m_socket != INVALID_SOCKET)
   {
      shutdown(m_socket, SHUT_RDWR);
      struct linger linger_opt = {1, 15};
      setsockopt(m_socket, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(linger_opt));
      closesocket(m_socket);
   }

   // Decrement the object count
   InterlockedDecrement(&numCChannelObjects);

   DbgTrace(1, "CChannel::~CChannel- End\n", 0);

}  /*-- CChannel::~CChannel() --*/


//++=======================================================================
void
CChannel::openSocket(void)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   DbgTrace(1, "CChannel::openSocket- Start\n", 0);

   // Open a domain socket if not using Tcp
   if (m_useTcpSocket == false)
      m_socket = socket(PF_UNIX, SOCK_STREAM, 0);
   else
      m_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

   DbgTrace(1, "CChannel::openSocket- End\n", 0);

}   /*-- CChannel::openSocket() --*/


//++=======================================================================
int
CChannel::connectSocket(void)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   int   retStatus;

   DbgTrace(1, "CChannel::connectSocket- Start\n", 0);

   // Proceed based on whether or not we need
   // to use Tcp sockets.
   if (m_useTcpSocket)
   {
      struct sockaddr_in localAddr = {0};

      // Setup the address structure
      localAddr.sin_family = AF_INET;
      localAddr.sin_addr.s_addr = htonl(INADDR_ANY);

      // Perform bind operation
      retStatus = bind(m_socket,
                       (const sockaddr*) &localAddr,
                       sizeof(struct sockaddr_in));
      if (!retStatus)
      {
         // Perform connect operation
         retStatus = connect(m_socket,
                             (struct sockaddr*) &m_remoteAddrIn,
                             sizeof(struct sockaddr_in));
         if (retStatus == SOCKET_ERROR)
         {
            DbgTrace(0, "CChannel::connectSocket- Connection creation failed, error = %d\n", errno);
         }
      }
      else
      {
         DbgTrace(0, "CChannel::connectSocket- Unable to bind socket, error = %d", errno);
      }
   }
   else
   {
      // Perform connect operation
      retStatus = connect(m_socket,
                          (struct sockaddr*) &m_remoteAddrUn,
                          sizeof(m_remoteAddrUn.sun_family) + strlen(m_remoteAddrUn.sun_path));
      if (retStatus == SOCKET_ERROR)
      {
         DbgTrace(0, "CChannel::connectSocket- Connection creation failed, error = %d\n", errno);
      }
   }

   DbgTrace(1, "CChannel::connectSocket- End, status = %08X\n", retStatus);

   return retStatus;

}   /*-- CChannel::connectSocket() --*/


//++=======================================================================
int
CChannel::init(void)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   int            retStatus = -1;
   SmartCChannel  *pSmartCChannel = NULL;

   DbgTrace(1, "CChannel::init- Start, Obj = %08X\n", this);

   // Verify the state of the object
   if (m_state == State_Uninitialized)
   {
      openSocket();
      if (m_socket != INVALID_SOCKET)
      {
         if (connectSocket() != SOCKET_ERROR)
         {
            // Advance the object state
            m_state = State_Connected;

            // Launch a thread to service the channel connection
            try {
               // Create a SmartCChannel object to make sure that the object
               // does not get deleted prematurely.
               pSmartCChannel = new SmartCChannel(this);

               // Create the channel connection thread
               pthread_t thread;
               int threadCreateStatus = pthread_create(&thread,
                                                       NULL,
                                                       (void*(*)(void*))CChannel::connectionThread,
                                                       pSmartCChannel);
               if (threadCreateStatus == 0)
               {
                  // We succeeded
                  retStatus = 0;
               }
               else
               {
                  DbgTrace(0, "CChannel::init- Unable to create channel connection thread, error = %08X\n", threadCreateStatus);
               }
            }
            catch (...) {
               DbgTrace(0, "CChannel::init- Exception caught creating smart pointer\n", 0);
            }
         }
         else
         {
            DbgTrace(0, "CChannel::init- Connection creation failed, error = %d\n", errno);
         }
      }
      else
      {
         DbgTrace(0, "CChannel::init- Unable to open socket, error = %d\n", errno);
      }
   }
   else
   {
      DbgTrace(0, "CChannel::init- invalid state, state = %d\n", m_state);
   }

   // Deal with initialization failures
   if (retStatus)
   {
      // Adjust the object state
      m_state = State_FailedInitialization;

      // Free SmartCChannel just in case
      delete pSmartCChannel;
   }

   DbgTrace(1, "CChannel::init- End, status = %08X\n", retStatus);

   return retStatus;

}  /*-- CChannel::init() --*/


//++=======================================================================
void*
CChannel::connectionThread(
   SmartPtr<CChannel> *pSmartCChannel)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L1
//=======================================================================--
{
   CChannel       *pCChannel = *pSmartCChannel;
   bool           doneReceivingData = false;
   unsigned long  bytesReceived;
   uint32_t       reqId;
   int            payloadLength;
   unsigned long  totalPayloadBytesReceived = 0;
   char           reqDataPktHdr[sizeof(ReqDataPktHdrTemplate) - 1];
   char           *pRecvBuff;
   RCMapIter      iter;
   ClientReq      *pClientReq;

   DbgTrace(1, "CChannel::connectionThread- Start, Obj = %08X\n", pCChannel);

   // Set the thread in the detached state so that it is cleaned up when it exits
	pthread_detach(pthread_self());

   // Check that we are still connected
   if (pCChannel->m_state == State_Connected)
   {
      // Receive and process channel data
      while (!doneReceivingData)
      {
         DbgTrace(2, "CChannel::connectionThread- Receive Loop, Obj = %08X\n", pCChannel);

         // Receive the ReqDataPktHdr. Note, if we add other packet types and if the
         // packet types have different header lengths, then we will need to modify
         // this code to first receive the packet type and then receive the rest
         // of the header based on type.
         while (1)
         {
            bytesReceived = recv(pCChannel->m_socket,
                                 reqDataPktHdr,
                                 sizeof(ReqDataPktHdrTemplate) - 1,
                                 MSG_WAITALL);
            if (bytesReceived == SOCKET_ERROR
                && errno == EINTR)
            {
               continue;
            }
            break;
         }
         if (bytesReceived != SOCKET_ERROR)
         {
            // Check if the connection was terminated
            if (bytesReceived == (sizeof(ReqDataPktHdrTemplate) - 1))
            {
               // Get the payload length
               if (ChannelProto::getReqIdAndPayloadLength(reqDataPktHdr,
                                                          sizeof(reqDataPktHdr),
                                                          &reqId,
                                                          &payloadLength))
               {
                  // Procced based on the packet type
                  switch (ChannelProto::getPktType(*reqDataPktHdr))
                  {
                     case ChannelProto::ReqDataCarrierPacketType:

                        DbgTrace(2, "CChannel::connectionThread- Processing Request Data Packet, Obj = %08X\n", pCChannel);

                        // Allocate a buffer big enough to receive the payload
                        pRecvBuff = (char*) malloc(payloadLength);
                        if (pRecvBuff != NULL)
                        {
                           // Buffer allocated, receive the request payload.
                           while (1)
                           {
                              bytesReceived = recv(pCChannel->m_socket,
                                                   pRecvBuff,
                                                   payloadLength,
                                                   MSG_WAITALL);
                              if (bytesReceived == SOCKET_ERROR
                                  && errno == EINTR)
                              {
                                 continue;
                              }
                              break;
                           }
                           if (bytesReceived != SOCKET_ERROR)
                           {
                              // Verify that we received all of the payload
                              if (bytesReceived == payloadLength)
                              {
                                 // Received all of the payload data
                                 totalPayloadBytesReceived += bytesReceived;
   
                                 // Acquire exclusive access to the CChannel object
                                 pthread_mutex_lock(&pCChannel->m_mutex);
   
                                 // Find the appropriate ClientReq object in the ClientReqMap using
                                 // the reqId present in the Req Data Packet Header.
                                 iter = pCChannel->m_rcMap.find(reqId);
                                 if (iter != pCChannel->m_rcMap.end())
                                 {
                                    // Object was found in the map, use it to process the
                                    // request payload.
                                    pClientReq = iter->second;
                                    pClientReq->processServerData(pRecvBuff,
                                                                  bytesReceived);
   
                                    // Forget about the receive buffer
                                    pRecvBuff = NULL;
                                 }
                                 else
                                 {
                                    DbgTrace(0, "CChannel::connectionThread- Error, did not find object in map, Obj = %08X\n", pCChannel);
                                 }
   
                                 // Release exclusive access to the CChannel object
                                 pthread_mutex_unlock(&pCChannel->m_mutex);
                              }
                              else
                              {
                                 DbgTrace(1, "CChannel::connectionThread- Connection aborted prematurely, Obj = %08X\n", pCChannel);
                                 //DbgTrace(0, "CChannel::connectionThread- 1Connection aborted prematurely, Obj = %08X\n", pCChannel);
                                 //DbgTrace(0, "CChannel::connectionThread- 1bytesReceived = %d\n", bytesReceived);
                                 //DbgTrace(0, "CChannel::connectionThread- 1payloadLength = %d\n", payloadLength);
                                 //DbgTrace(0, "CChannel::connectionThread- 1errno = %d\n", errno);
                                 //printf("bytesReceived = %d, payloadLength = %d\n", bytesReceived, payloadLength); 
                                 //printf("CChannel::connectionThread- 1Connection aborted prematurely, Obj = %08X\n", pCChannel);
                                 doneReceivingData = true;
                              }
                           }
                           else
                           {
                              DbgTrace(1, "CChannel::connectionThread- Connection aborted prematurely, Obj = %08X\n", pCChannel);
                              //DbgTrace(0, "CChannel::connectionThread- 2Connection aborted prematurely, Obj = %08X\n", pCChannel);
                              //DbgTrace(0, "CChannel::connectionThread- 2errno = %d\n", errno);
                              //printf("Socket error = %d\n", errno);
                              //printf("CChannel::connectionThread- 2Connection aborted prematurely, Obj = %08X\n", pCChannel);
                              doneReceivingData = true;
                           }
   
                           // Free receive buffer if necessary
                           if (pRecvBuff)
                              delete[] pRecvBuff;
                        }
                        else
                        {
                           DbgTrace(0, "CChannel::connectionThread- Unable to allocate receive buffer, Obj = %08X\n", pCChannel);
                           doneReceivingData = true;
                        }
                        break;

                     case ChannelProto::ReqErrorCarrierPacketType:

                        DbgTrace(1, "CChannel::connectionThread- Processing Request Error Packet, Obj = %08X\n", pCChannel);

                        // Acquire exclusive access to the CChannel object
                        pthread_mutex_lock(&pCChannel->m_mutex);
   
                        // Find the appropriate ClientReq object in the ClientReqMap using
                        // the reqId present in the Req Data Packet Header.
                        iter = pCChannel->m_rcMap.find(reqId);
                        if (iter != pCChannel->m_rcMap.end())
                        {
                           // Object was found in the map, use it to process the Request Error.
                           pClientReq = iter->second;
                           pClientReq->processError();
                        }
                        else
                        {
                           DbgTrace(0, "CChannel::connectionThread- Error, did not find object in map, Obj = %08X\n", pCChannel);
                        }
   
                        // Release exclusive access to the CChannel object
                        pthread_mutex_unlock(&pCChannel->m_mutex);
                        break;

                     default:

                        DbgTrace(0, "CChannel::connectionThread- Unknown Packet Type, Obj = %08X\n", pCChannel);
                        doneReceivingData = true;
                        break;
                  }
               }
               else
               {
                  DbgTrace(1, "CChannel::connectionThread- Unable to obtain payload length, Obj = %08X\n", pCChannel);
                  doneReceivingData = true;
               }
            }
            else
            {
               DbgTrace(1, "CChannel::connectionThread- The channel connection was terminated, Obj = %08X\n", pCChannel);
               //DbgTrace(0, "CChannel::connectionThread- 3The channel connection was terminated, Obj = %08X\n", pCChannel);
               //DbgTrace(0, "CChannel::connectionThread- 3bytesReceived = %d\n", bytesReceived);
               //DbgTrace(0, "CChannel::connectionThread- 3expected = %d\n", ReqDataPktHdrTemplate.length());
               //DbgTrace(0, "CChannel::connectionThread- 3errno = %d\n", errno);
               //printf("bytesReceived = %d, expected = %d\n", bytesReceived, ReqDataPktHdrTemplate.length()); 
               //printf("CChannel::connectionThread- 3The channel connection was terminated, Obj = %08X\n", pCChannel);
               doneReceivingData = true;
            }
         }
         else
         {
            DbgTrace(1, "CChannel::connectionThread- The channel connection was aborted, Obj = %08X\n", pCChannel);
            //DbgTrace(0, "CChannel::connectionThread- 4The channel connection was aborted, Obj = %08X\n", pCChannel);
            //DbgTrace(0, "CChannel::connectionThread- 4errno = %d\n", errno);
            //printf("Socket error = %d\n", errno);
            //printf("CChannel::connectionThread- 4The channel connection was aborted, Obj = %08X\n", pCChannel);
            doneReceivingData = true;
         }
      }
   }

   // Acquire exclusive access to the CChannel object
   pthread_mutex_lock(&pCChannel->m_mutex);

   // Try to change the CChannel state to disconnected
   if (pCChannel->m_state == State_Connected)
      pCChannel->m_state = State_Disconnected;

   // Deliver error notifications to all of the ClientReqs
   // still associated with the CChannel.
   if (!pCChannel->m_rcMap.empty())
   {
      iter = pCChannel->m_rcMap.begin();
      while (iter != pCChannel->m_rcMap.end())
      {
         // Object was found in the map, deliver error notification
         // to it.
         pClientReq = iter->second;
         pClientReq->processError();

         // Move on to the next element in the map
         iter ++;
      }
   }

   // Release exclusive access to the CChannel object
   pthread_mutex_unlock(&pCChannel->m_mutex);

   // Free SmartCChannel
   delete pSmartCChannel;

   DbgTrace(1, "CChannel::connectionThread- End\n", 0);

   // Exit
	pthread_exit(NULL);

	return 0;	// never-reached!

}  /*-- CChannel::connectionThread() --*/


//++=======================================================================
void
CChannel::closeChannel(void)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{

   DbgTrace(1, "CChannel::closeChannel- Start, Obj = %08X\n", this);

   // Acquire CChannel mutex
   pthread_mutex_lock(&m_mutex);

   // Switch the socket state to closed
   m_state = State_Closed;

   // Check if we must close the socket
   if (m_socket != INVALID_SOCKET)
   {
      // Socket needs to be closed, this will
      // release the channel connection thread
      // if it is active.
      shutdown(m_socket, SHUT_RDWR);
      struct linger linger_opt = {1, 15};
      setsockopt(m_socket, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(linger_opt));
      closesocket(m_socket);
      m_socket = INVALID_SOCKET;
   }

   // Release CChannel mutex
   pthread_mutex_unlock(&m_mutex);

   DbgTrace(1, "CChannel::closeChannel- End\n", 0);

}  /*-- CChannel::closeChannel() --*/


//++=======================================================================
bool
CChannel::ok(void)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   bool retStatus;

   DbgTrace(1, "CChannel::ok- Start, Obj = %08X\n", this);

   // Return true if connected
   if (m_state == State_Connected)
      retStatus = true;
   else
      retStatus = false;

   DbgTrace(1, "CChannel::ok- End, retStatus = %08X\n", retStatus);

   return retStatus;

}  /*-- CChannel::ok() --*/


//++=======================================================================
uint32_t
CChannel::allocReqId(void)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   uint32_t allocatedId;

   DbgTrace(1, "CChannel::allocReqId- Start, Obj = %08X\n", this);

   // Perform atomic operation
   allocatedId = InterlockedIncrement(&m_reqIdAllocator);

   DbgTrace(1, "CChannel::allocReqId- End, allocatedId = %08X\n", allocatedId);

   return allocatedId;

}  /*-- CChannel::allocReqId() --*/


//++=======================================================================
int
CChannel::submitReq(
   uint32_t reqId,
   ClientReq &clientReq,
   char *pClientData,
   int clientDataLen)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   int            retStatus = -1;
   char           reqDataPktHdr[sizeof(ReqDataPktHdrTemplate) - 1];
   struct msghdr  sendmsgHdr = {0};
   struct iovec   ioVectors[2];
   unsigned long  bytesSent;
   unsigned long  totalBytesSent = 0;
   unsigned long  bytesToSend = sizeof(reqDataPktHdr) + clientDataLen;

   DbgTrace(1, "CChannel::submitReq- Start, Obj = %08X\n", this);

   // Acquire exclusive access to the channel object
   pthread_mutex_lock(&m_mutex);

   // Verify that the channel is connected
   if (m_state == State_Connected)
   {
      // Insert the specified ClientReq into the ClientReqMap and forward the client
      // data to the server.
      RCIterBoolPair insertResult;
      insertResult = m_rcMap.insert(make_pair(reqId, &clientReq));
      if (!insertResult.second)
      {
         // Insertion failed
         DbgTrace(0, "CChannel::submitReq- Unable to insert ClientReq into ClientReqMap, Obj = %08X\n", this);
      }
      else
      {
         // Insertion succeded, now send the request to the server.
         //
         // Build ReqDataHeader
         if (ChannelProto::buildReqDataPktHdr(reqId,
                                              clientDataLen,
                                              reqDataPktHdr) == 0)
         {
            // Packet header was built, now sent it along with the client data to
            // the server.
            ioVectors[0].iov_base = reqDataPktHdr;
            ioVectors[0].iov_len = sizeof(reqDataPktHdr);
            ioVectors[1].iov_base = (char*) pClientData;
            ioVectors[1].iov_len = clientDataLen;
            sendmsgHdr.msg_iov = ioVectors;
            sendmsgHdr.msg_iovlen = 2;
            while (1)
            {
               bytesSent = sendmsg(m_socket,
                                   &sendmsgHdr,
                                   MSG_NOSIGNAL);
               if (bytesSent == SOCKET_ERROR)
               {
                  // Check if we were interrupted during the transfer
                  if (errno == EINTR)
                  {
                     // Just try again
                     continue;
                  }

                  // An unrecoverable error was encountered during the send operation,
                  // assume there was a communication failure. Close the socket to make
                  // sure that the connectionThread cleans up.
                  DbgTrace(0, "CChannel::submitReq- sendmsg error, errno = %d\n", errno);
                  m_state = State_Disconnected;
                  shutdown(m_socket, SHUT_RDWR);
                  struct linger linger_opt = {1, 15};
                  setsockopt(m_socket, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(linger_opt));
                  closesocket(m_socket);
                  m_socket = INVALID_SOCKET;
                  break;
               }
               else
               {
                  // Account for the bytes sent
                  totalBytesSent += bytesSent;

                  // Check if we are done sending all of the data
                  if (totalBytesSent >= bytesToSend)
                  {
                     // We are done
                     break;
                  }
                  else
                  {
                     // Adjust the ioVector structure to send data not yet sent
                     if (totalBytesSent >= sizeof(reqDataPktHdr))
                     {
                        // The packet header was sent, use only one ioVector.
                        int   clientDataAlreadySent = totalBytesSent - sizeof(reqDataPktHdr);
                        ioVectors[0].iov_base = (char*) pClientData + clientDataAlreadySent;
                        ioVectors[0].iov_len = clientDataLen - clientDataAlreadySent;
                        sendmsgHdr.msg_iov = ioVectors;
                        sendmsgHdr.msg_iovlen = 1;
                     }
                     else
                     {
                        // Not all of the packet header was sent, use two ioVectors.
                        ioVectors[0].iov_base = (char*) reqDataPktHdr + totalBytesSent;
                        ioVectors[0].iov_len = sizeof(reqDataPktHdr) - totalBytesSent;
                        ioVectors[1].iov_base = (char*) pClientData;
                        ioVectors[1].iov_len = clientDataLen;
                        sendmsgHdr.msg_iov = ioVectors;
                        sendmsgHdr.msg_iovlen = 2;
                     }
                  }
               }
            }

            // Return success even if the send failed to allow things to be cleaned up
            // by the connectionThread routine.
            retStatus = 0;
         }
         else
         {
            DbgTrace(0, "CChannel::submitReq- Error building Req Data Pkt Header, Obj = %08X\n", this);

            // Remove ClientReq from the ClientReqMap
            RCMapIter iter = m_rcMap.find(reqId);
            if (iter != m_rcMap.end())
            {
               // Object was found in the map, remove it.
               m_rcMap.erase(iter);
            }
            else
            {
               DbgTrace(0, "CChannel::submitReq- Error, did not find object in map to remove\n", 0);
            }
         }
      }
   }
   else
   {
      DbgTrace(1, "CChannel::submitReq- Channel not connected, state = %08X\n", m_state);
   }

   // Release exclusive access to the channel object
   pthread_mutex_unlock(&m_mutex);

   DbgTrace(1, "CChannel::submitReq- End, retStatus = %08X\n", retStatus);

   return retStatus;

}  /*-- CChannel::submitReq() --*/


//++=======================================================================
void
CChannel::removeReq(
   uint32_t reqId)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   DbgTrace(1, "CChannel::removeReq- Start, Obj = %08X\n", this);

   // Acquire exclusive access to the channel object
   pthread_mutex_lock(&m_mutex);

   // Try to find the ClientReq in the ClientReqMap using the reqId
   RCMapIter iter = m_rcMap.find(reqId);
   if (iter != m_rcMap.end())
   {
      // Object was found in the map, remove it.
      m_rcMap.erase(iter);
   }
   else
   {
      DbgTrace(0, "CChannel::removeReq- Error, did not find object in map\n", 0);
   }

   // Release exclusive access to the channel object
   pthread_mutex_unlock(&m_mutex);

   DbgTrace(1, "CChannel::removeReq- End\n", 0);

}  /*-- CChannel::removeReq() --*/


//++=======================================================================
int
CChannel::sendData(
   uint32_t reqId,
   char *pClientData,
   int clientDataLen)
//
//  Arguments: 
//
//  Returns:   
//
//  Abstract:  
//
//  Notes:
//
// L2
//=======================================================================--
{
   int            retStatus = -1;
   char           reqDataPktHdr[sizeof(ReqDataPktHdrTemplate) - 1];
   struct msghdr  sendmsgHdr = {0};
   struct iovec   ioVectors[2];
   unsigned long  bytesSent;
   unsigned long  totalBytesSent = 0;
   unsigned long  bytesToSend = sizeof(reqDataPktHdr) + clientDataLen;

   DbgTrace(1, "CChannel::sendData- Start, Obj = %08X\n", this);

   // Acquire exclusive access to the channel object
   pthread_mutex_lock(&m_mutex);

   // Verify that the channel is connected
   if (m_state == State_Connected)
   {
      // Build ReqDataHeader
      if (ChannelProto::buildReqDataPktHdr(reqId,
                                           clientDataLen,
                                           reqDataPktHdr) == 0)
      {
         // Packet header was built, now sent it along with the client data to
         // the server.
         ioVectors[0].iov_base = reqDataPktHdr;
         ioVectors[0].iov_len = sizeof(reqDataPktHdr);
         ioVectors[1].iov_base = (char*) pClientData;
         ioVectors[1].iov_len = clientDataLen;
         sendmsgHdr.msg_iov = ioVectors;
         sendmsgHdr.msg_iovlen = 2;
         while (1)
         {
            bytesSent = sendmsg(m_socket,
                                &sendmsgHdr,
                                MSG_NOSIGNAL);
            if (bytesSent == SOCKET_ERROR)
            {
               // Check if we were interrupted during the transfer
               if (errno == EINTR)
               {
                  // Just try again
                  continue;
               }

               // An unrecoverable error was encountered during the send operation,
               // assume there was a communication failure. Close the socket to make
               // sure that the connectionThread cleans up.
               DbgTrace(0, "CChannel::sendData- sendmsg error, errno = %d\n", errno);
               m_state = State_Disconnected;
               shutdown(m_socket, SHUT_RDWR);
               struct linger linger_opt = {1, 15};
               setsockopt(m_socket, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(linger_opt));
               closesocket(m_socket);
               m_socket = INVALID_SOCKET;
               break;
            }
            else
            {
               // Account for the bytes sent
               totalBytesSent += bytesSent;

               // Check if we are done sending all of the data
               if (totalBytesSent >= bytesToSend)
               {
                  // We are done
                  break;
               }
               else
               {
                  // Adjust the ioVector structure to send data not yet sent
                  if (totalBytesSent >= sizeof(reqDataPktHdr))
                  {
                     // The packet header was sent, use only one ioVector.
                     int   clientDataAlreadySent = totalBytesSent - sizeof(reqDataPktHdr);
                     ioVectors[0].iov_base = (char*) pClientData + clientDataAlreadySent;
                     ioVectors[0].iov_len = clientDataLen - clientDataAlreadySent;
                     sendmsgHdr.msg_iov = ioVectors;
                     sendmsgHdr.msg_iovlen = 1;
                  }
                  else
                  {
                     // Not all of the packet header was sent, use two ioVectors.
                     ioVectors[0].iov_base = (char*) reqDataPktHdr + totalBytesSent;
                     ioVectors[0].iov_len = sizeof(reqDataPktHdr) - totalBytesSent;
                     ioVectors[1].iov_base = (char*) pClientData;
                     ioVectors[1].iov_len = clientDataLen;
                     sendmsgHdr.msg_iov = ioVectors;
                     sendmsgHdr.msg_iovlen = 2;
                  }
               }
            }
         }

         // Return success even if the send failed to allow things to be cleaned up
         // by the connectionThread routine.
         retStatus = 0;
      }
      else
      {
         DbgTrace(0, "CChannel::sendData- Error building Req Data Pkt Header, Obj = %08X\n", this);
      }
   }
   else
   {
      DbgTrace(1, "CChannel::sendData- Channel not connected, state = %08X\n", m_state);
   }

   // Release exclusive access to the channel object
   pthread_mutex_unlock(&m_mutex);

   DbgTrace(1, "CChannel::sendData- End, retStatus = %08X\n", retStatus);

   return retStatus;

}  /*-- CChannel::sendData() --*/


//=========================================================================
//=========================================================================