//-----------------------------------------------------------------------------
// Copyright (c) 1994,1995 Southeastern Universities Research Association,
//                         Continuous Electron Beam Accelerator Facility
//
// This software was developed under a United States Government license
// described in the NOTICE file included as part of this distribution.
//
//-----------------------------------------------------------------------------
//
// Description:
//      Implementation of rsvcClient Class
//
// Author:  Jie Chen
//
// Revision History:
//   rsvcClient.cc,v
// Revision 1.8  1998/03/19  18:30:56  chen
// add monitorEntry capability to server table
//
// Revision 1.6  1998/02/25  19:52:55  chen
// keep alive option on
//
// Revision 1.5  1998/02/24  14:13:39  chen
// remove errno check on read_n
//
// Revision 1.4  1998/02/20  19:37:28  chen
// reset errno after using it (hpux) sometimes not resetting it
//
// Revision 1.3  1998/02/06  15:12:32  chen
// add connection management to the name server
//
// Revision 1.2  1998/01/22  20:04:06  chen
// to make it work with cdevReactor
//
// Revision 1.1  1998/01/22  17:08:02  akers
// Addition of new NameServer
//
//
//
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/types.h>

#ifdef _WIN32
#include <WinSock2.h>
#include <fcntl.h>
#include <io.h>

#define RSVC_ERRNO  WSAGetLastError()  
#define EADDRINUSE  WSAEADDRINUSE
#define EINPROGRESS WSAEINPROGRESS
#define SYSINTR     WSAEINTR
#define EISCONN     WSAEISCONN
#define ETIMEDOUT   WSAETIMEDOUT
#define EWOULDBLOCK WSAEWOULDBLOCK

#define RSVC_SET_ERRNO(e) WSASetLastError(e)

#define SOCKOPT_VALUE_PTR (const char *)

#else   /* _WIN32 */
#include <sys/socket.h>
#include <unistd.h>
#include <netinet/in.h>
#include <netdb.h>

#define RSVC_ERRNO errno
#define SYSINTR    EINTR
#define RSVC_SET_ERRNO(e) errno=e

#ifndef solaris
#define SOCKOPT_VALUE_PTR (void *)
#else
#define SOCKOPT_VALUE_PTR (char *)
#endif

#endif

#if defined(AIX) || defined(__linux)
	typedef unsigned int SOCKOPT_SIZE_PARM;
#else
	typedef int SOCKOPT_SIZE_PARM;
#endif

#include <rsvcErr.h>
#include "rsvcClient.h"

rsvcClient::rsvcClient (void)
:connected_ (0), tcp_fd_ (-1), server_host_ (0), server_port_ (0),
 lock_ (0), reqid_ (1234), cbkid_ (93), discCbkList_ (), cmdCbkList_ (),
 monitorCbkTable_ (1000, 3)
{
#ifdef _TRACE_OBJECTS
  printf ("Create rsvcClient Class Object\n");
#endif

#ifdef _WIN32
  WORD wVersion;
  WSADATA wsaData;

  wVersion = MAKEWORD (2, 0); /* WinSock 2.0 */
  if (WSAStartup (wVersion, &wsaData) != 0) {
    WSACleanup ();
    fprintf (stderr, "rsvcClient: Cannot initialize win socket\n");
  }
#endif

  cbuflen_ = 4096;
  cbuffer_ = new char[cbuflen_];
  if (!cbuffer_) {
    fprintf (stderr, "rsvcClient Fatal: malloc error\n");
    exit (1);
  }
}

rsvcClient::~rsvcClient (void)
{
#ifdef _TRACE_OBJECTS
  printf ("Delete rsvcClient Class Object\n");
#endif  

  if (server_host_) 
    delete []server_host_;

  if (connected_)
    disconnect ();

  delete []cbuffer_;

#ifdef _WIN32
  WSACleanup ();
#endif
}

int
rsvcClient::connect (char* host, unsigned short port,
		     double seconds)
{
  if (connected_)
    return RSVC_SUCCESS;

  struct hostent* phe = 0;
  struct sockaddr_in sin;

  memset (&sin, 0, sizeof (sin));
  sin.sin_family = AF_INET;
  sin.sin_port = htons (port);

  if (phe = gethostbyname (host)) 
    memcpy (&sin.sin_addr, phe->h_addr, phe->h_length);
  else {
    fprintf (stderr, "RSVC Client: gethostbyname %s error\n", host);
    return RSVC_ERROR;
  }

  tcp_fd_ = socket (AF_INET, SOCK_STREAM, 0);

  if (tcp_fd_ < 0) {
    fprintf (stderr, "RSVC Client: cannot create tcp socket\n");
    return RSVC_ERROR;
  }

  // if timeout is zero we do simple blocked connect
  int status;
  if (seconds < 0.00001) {
    if ((status = ::connect (tcp_fd_, (struct sockaddr *)&sin,
			     sizeof (sin))) < 0) {
      fprintf (stderr, "RSVC Client: cannot connect to server %s@%d\n",
	       host, port);
      return RSVC_ERROR;
    }
    else {
      connected_ = 1;

      // remember the server information
      if (server_host_)
	delete server_host_;
      server_host_ = new char[strlen (host) + 1];
      strcpy (server_host_, host);
      server_port_ = port;

      // set socket option keep alive
      int alive = 1;
      setsockopt (tcp_fd_, SOL_SOCKET, SO_KEEPALIVE, SOCKOPT_VALUE_PTR &alive,
		  (SOCKOPT_SIZE_PARM)sizeof (alive));
      return RSVC_SUCCESS;
    }
  }
  else {
    struct timeval timeout;

    timeout.tv_sec = (int)seconds;
    timeout.tv_usec = (int) ((seconds - timeout.tv_sec)*1000000);
    if (connect_nonblock (tcp_fd_, (struct sockaddr *)&sin,
			  sizeof (sin), &timeout) != RSVC_SUCCESS) {
      close (tcp_fd_);
      tcp_fd_ = -1;
      return RSVC_ERROR;
    }
    connected_ = 1;

    // remember the server information
    if (server_host_)
      delete server_host_;
    server_host_ = new char[strlen (host) + 1];
    strcpy (server_host_, host);
    server_port_ = port;

    // set socket option keep alive
    int alive = 1;
    setsockopt (tcp_fd_, SOL_SOCKET, SO_KEEPALIVE, SOCKOPT_VALUE_PTR &alive,
		(SOCKOPT_SIZE_PARM)sizeof (alive));
    return RSVC_SUCCESS;
  }
}

int
rsvcClient::connect_nonblock (int sockfd, struct sockaddr* addr,
			     size_t addrlen, struct timeval* timeout)
{
  int flags, n, error;
  int len;
  fd_set rset, wset;

#ifndef _WIN32
  // set non block mode
  flags = fcntl (sockfd, F_GETFL, 0);
  fcntl (sockfd, F_SETFL, flags | O_NONBLOCK);
#endif

  error = 0;

  if ((n = ::connect (sockfd, addr, addrlen)) < 0) {
    if (errno != EINPROGRESS && errno != EWOULDBLOCK) 
      return RSVC_ERROR;
  }

  if (n == 0) 
    goto connect_done;       // connection complete

  FD_ZERO (&rset);
  FD_SET  (sockfd, &rset);
  memcpy (&wset, &rset, sizeof (rset));

#ifdef _SELECT_USES_INT_
  if ((n = select (sockfd + 1, (int *)&rset, (int *)&wset, (int *)0, timeout)) == 0) 
#else
  if ((n = select (sockfd + 1, &rset, &wset, 0, timeout)) == 0) 
#endif    
  {
    // timeout
    RSVC_SET_ERRNO (ETIMEDOUT);
    return RSVC_ERROR;
  }
  if (FD_ISSET (sockfd, &rset) || FD_ISSET (sockfd, &wset)) {
    len = sizeof (error);
    if (getsockopt (sockfd, SOL_SOCKET, SO_ERROR, (char *)&error, (SOCKOPT_SIZE_PARM *)&len) < 0) 
      return RSVC_ERROR;   // solaris pending error
  }
  else 
    return RSVC_ERROR;
  
connect_done:
#ifndef _WIN32
  // restore old flag
  fcntl (sockfd, F_SETFL, flags);
#endif

  if (error) {
    RSVC_SET_ERRNO(error);
    return RSVC_ERROR;
  }
  return RSVC_SUCCESS;
}

int
rsvcClient::disconnect (void)
{
  if (connected_) {
    shutdown (tcp_fd_, 0);
    close (tcp_fd_);
    tcp_fd_ = -1;
    connected_ = 0;

    // remove all callbacks
    cleanupCallbacks ();
  }
  return RSVC_SUCCESS;
}

int
rsvcClient::connected (void)
{
  return connected_;
}

int
rsvcClient::disconnectCallback (rsvcCbkFunc func, void* arg)
{
  rsvcCallback* cbk = new rsvcCallback (func, arg);

  rsvcSlistIterator ite (discCbkList_);
  rsvcCallback* tcbk = 0;
  int           error = 0;

  for (ite.init (); !ite; ++ite) {
    tcbk = (rsvcCallback *) ite ();
    if (*tcbk == *cbk) {
      error = 1;
      break;
    }
  }
  if (!error) {
    discCbkList_.add ((void *)cbk);
    return RSVC_SUCCESS;
  }
  delete cbk;
  return RSVC_ERROR;
}


void
rsvcClient::cleanupCallbacks (void)
{
  // delete everything from disconnect callback list
  rsvcSlistIterator ite (discCbkList_);
  rsvcCallback*     cbk = 0;

  for (ite.init (); !ite; ++ite) {
    cbk = (rsvcCallback *) ite ();
    delete cbk;
  }
  discCbkList_.deleteAllValues ();

  // delete all from command callback list
  rsvcSlistIterator cite (cmdCbkList_);
  rsvcCbk*          scbk = 0;
  
  for (cite.init (); !cite; ++cite) {
    scbk = (rsvcCbk *) cite ();
    cbk = (rsvcCallback *)scbk->userptr ();
    delete cbk;
    delete scbk;
  }
  cmdCbkList_.deleteAllValues ();

  // delete all from monitor table
  rsvcHashIterator hite (monitorCbkTable_);

  for (hite.init (); !hite; ++hite) {
    scbk = (rsvcCbk *) hite ();
    cbk = (rsvcCallback *)scbk->userptr ();
    delete cbk;
    delete scbk;  
  }
  monitorCbkTable_.deleteAllValues ();
}

void
rsvcClient::callAllDiscCbks (void)
{
  rsvcSlistIterator ite (discCbkList_);
  rsvcCallback* cbk = 0;
  rsvcCbkFunc   func = 0;
  void *arg = 0;

  for (ite.init (); !ite; ++ite) {
    cbk = (rsvcCallback *)ite ();
    func = cbk->cbkFunc ();
    arg = cbk->userarg ();
    (*func)(RSVC_DISCONNECTED, arg, (rsvcData *)0);
  }
}  


int
rsvcClient::getFd (void) const
{
  return tcp_fd_;
}

int
rsvcClient::pendIO (double seconds)
{
  // first prevent recursice calling from a callback
  if (lock_)
    return RSVC_INVALID;

  if (tcp_fd_ == -1)
    return RSVC_BADIO;

  int nfound = 0;
  fd_set       readfd;
  struct timeval tv;

  if (seconds <= 0.000001) {
    tv.tv_sec = 0;
    tv.tv_usec = 0;
  }
  else {
    tv.tv_sec = (int)seconds;
    tv.tv_usec = (int)((seconds - tv.tv_sec)*1000000);
  }
    
  FD_ZERO (&readfd);
  FD_SET  (tcp_fd_, &readfd); 

  do {
#ifndef __hpux
    nfound = select (tcp_fd_ + 1, &readfd, 0, 0, &tv);
#else

#ifdef _SELECT_USES_INT_
    nfound = select (tcp_fd_ + 1, (int *)&readfd, (int *)0, (int *)0,
		     &tv);
#else
    nfound = select (tcp_fd_ + 1, &readfd, 0, 0, &tv);
#endif

#endif
  }while (nfound == -1 && errno == SYSINTR);

  if (nfound == -1) { // something wrong with tcp_fd_ 
    fprintf (stderr, "rsvcClient: Bad tcp fd %d\n", tcp_fd_);
    tcp_fd_ = -1;
    return RSVC_BADIO;
  }
  else if (nfound == 0) // timeout
    return RSVC_TIMEOUT;
  else {
    if (FD_ISSET (tcp_fd_, &readfd)) {
      if (handle_input (tcp_fd_) < 0) { // bad io
	handle_close (tcp_fd_);
	return RSVC_IOERROR;
      }
    }
  }
  return RSVC_SUCCESS;
}

int
rsvcClient::pendIO (void)
{
  if (lock_)
    return RSVC_INVALID;

  if (tcp_fd_ == -1)
    return RSVC_BADIO;

  int nfound = 0;
  fd_set             readfd;  

  FD_ZERO (&readfd);
  FD_SET  (tcp_fd_, &readfd);

  do {
#ifndef __hpux
    nfound = select (tcp_fd_ + 1, &readfd, 0, 0, 0);
#else

#ifdef _SELECT_USES_INT_
    nfound = select (tcp_fd_ + 1, (int *)&readfd, (int *)0, (int *)0,
		     (struct timeval *)0);
#else
    nfound = select (tcp_fd_ + 1, &readfd, 0, 0, 0);
#endif

#endif
  }while (nfound == -1 && errno == EINTR);

  if (nfound == -1) { // something wrong with tcp_fd_
    fprintf (stderr, "rsvcClient: bad tcp fd %d\n", tcp_fd_);
    tcp_fd_ = -1;
    return RSVC_BADIO;
  }
  else if (nfound == 0) // timeout
    return RSVC_TIMEOUT;
  else {
    if (FD_ISSET (tcp_fd_, &readfd)) {
      if (handle_input (tcp_fd_) < 0) { // bad io
	handle_close (tcp_fd_);
	return RSVC_IOERROR;
      }
    }
  }
  return RSVC_SUCCESS;
}

int
rsvcClient::handle_input (int)
{
  int n;
  rsvcNetData ndata;
  
  // lock thus event processing function to prevent a recursive call
  rsvcClientLocker locker (this);

  // read header information
  char header[32];
  n = rsvcClient::read_n (tcp_fd_, header, rsvcNetData::headerLen ());
  if (n != rsvcNetData::headerLen ()) {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: Receiving netdata header error\n");
#endif
    printf ("rsvcClient: Server is dead\n");
    return -1;
  }

  size_t datasize;
  if (rsvcNetData::readHeader (header, n, &datasize) != RSVC_SUCCESS) {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: reading netdata header error\n");
#endif
    return -1;
  }

  if (datasize > cbuflen_) {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: incoming data size %d > %d, resize\n",
	    datasize, cbuflen_);
#endif
    delete []cbuffer_;
    cbuflen_ = 2*datasize;
    cbuffer_ = new char[cbuflen_];
    if (!cbuffer_) {
      fprintf (stderr, "rsvcClientIO: malloc error in hanlde_input\n");
      exit (1);
    }
  }

  // now really try to read data
  n = rsvcClient::read_n (tcp_fd_, cbuffer_, datasize);
  if (n != datasize) {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: Received %d bytes != %d bytes\n",
	    n, datasize);
#endif
    return -1;
  }
  
  // convert data
  if (ndata.streamIn (cbuffer_, datasize) != RSVC_SUCCESS) {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: stream in netdata error\n");
#endif
    return -1;
  }

  return processData (ndata);
}

int
rsvcClient::handle_close (int)
{
  shutdown (tcp_fd_, 0);
  close (tcp_fd_);
  tcp_fd_ = -1;
  connected_ = 0;

  // call all disconnect callback
  callAllDiscCbks ();

  // cleanup all callbacks
  cleanupCallbacks ();

  return 0;
}


int
rsvcClient::processData (rsvcNetData& ndata)
{
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  int       op   = cbk.opcode ();
  int       status = 0;

  if (op < RSVC_MONITOR_ON) {
    // all simple transaction callback
    status = cmdCbkFromServer ((int)cbk.cbkstatus (), data, cbk);
  }
  else if (op < RSVC_OP_UNKNOWN){
    // all monitor callbacks
    status = monitorCbkFromServer ((int)cbk.cbkstatus (), data, cbk);
  }
  else {
    fprintf (stderr, "rsvcClient: Invalid operation code\n");
    status = -1;
  }
  return status;
}

int
rsvcClient::cmdCbkFromServer (int status, rsvcData& data, rsvcCbk& cbk)
{
  rsvcSlistIterator ite (cmdCbkList_);
  rsvcCbk* tcbk = 0;
  rsvcCallback* rcbk = 0;
  int found = 0;

  for (ite.init (); !ite; ++ite) {
    tcbk = (rsvcCbk *) ite ();
    if (tcbk->match (&cbk)) {
      found = 1;
      rcbk = (rsvcCallback *)tcbk->userptr ();
      (*rcbk->cbkFunc() )(status, rcbk->userarg (), &data);
    }
  }
  if (found && status != RSVC_INCOMPLETE) {
    cmdCbkList_.remove ((void *)tcbk);
    delete rcbk;
    delete tcbk;
  }
  else {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: Callback from server does not match any\n");
#endif
  }

  return 0;
}

int
rsvcClient::monitorCbkFromServer (int status, rsvcData& data,
				  rsvcCbk& cbk)
{
  rsvcHSlist& list = monitorCbkTable_.bucketRef (&cbk);
  rsvcHSlistIterator ite (list);

  rsvcCbk* tcbk = 0;
  rsvcCallback* rcbk = 0;
  int found = 0;

  for (ite.init (); !ite; ++ite) {
    tcbk = (rsvcCbk *) ite ();
    if (tcbk->match (&cbk)) {
      found = 1;
      rcbk = (rsvcCallback *)tcbk->userptr ();
      (*rcbk->cbkFunc())(status, rcbk->userarg (), &data);
    }
  }
  if (found && cbk.cbkstatus () != RSVC_SUCCESS 
      && cbk.cbkstatus () != RSVC_INCOMPLETE) {
    // if callback is bad or finished remove it from the list
    list.remove (tcbk);
    delete rcbk;
    delete tcbk;
  }
  else if (cbk.cbkstatus () != RSVC_SUCCESS) {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: monitorCallback from server does not match any\n");
    printf ("rsvcClient: cbk has %d %d %d %d %d\n",
	    cbk.opcode (), cbk.reqid (), cbk.cbkid (), cbk.clientid (),
	    cbk.cbkstatus());
#endif
  }
  
  return 0;
}
  

int
rsvcClient::createMemDbase (char* name, rsvcData& data,
			    rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;
  
  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else 
    status = commandCallback ((int)RSVC_CREATE_MEMTABLE, data,
			      func, arg);
  return status;
}


int
rsvcClient::openDatabase (char* name, rsvcData& data,
			  rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;
  
  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else 
    status = commandCallback ((int)RSVC_OPEN_DBASE, data,
			      func, arg);
  return status;
}

int
rsvcClient::insertValue (char* name, rsvcData& data,
			 rsvcCbkFunc func, void* arg, int overwrite)
{
  int status = RSVC_SUCCESS;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS) 
    status = RSVC_ERROR;
  else {
    if (!overwrite)
      status = commandCallback ((int)RSVC_INSERT, data,
				func, arg);
    else
      status = commandCallback ((int)RSVC_OVERWRITE, data,
				func, arg);
  }
  return status;
}

int
rsvcClient::getValue (char* name, rsvcData& data,
		      rsvcCbkFunc func, void* arg)
{
 int status = RSVC_SUCCESS;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else
    status = commandCallback ((int)RSVC_GET, data,
			      func, arg);
  return status;
}


int
rsvcClient::delValue (char* name, rsvcData& data,
		      rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else
    status = commandCallback ((int)RSVC_DEL, data,
			      func, arg);
  return status;
}

int
rsvcClient::setValue (char* name, rsvcData& data,
		      rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else
    status = commandCallback ((int)RSVC_SET, data,
			      func, arg);
  return status;
}


int
rsvcClient::commandCallback (int opcode, rsvcData& data,
			     rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;
  size_t nbytes = 0;
  size_t nw = 0;

  rsvcCallback* cbk = new rsvcCallback (func, arg);
  rsvcCbk*      cbkid = new rsvcCbk (opcode, cbkid_++,
				     reqid_++, tcp_fd_, 0,
				     RSVC_SUCCESS, (void *)cbk);

  // create netdata
  rsvcNetData   netdata (data, *cbkid);
    
  if ((nbytes = streamData (netdata)) <= 0)
    status = RSVC_ERROR;
  else {
    // block write
    if ((nw = rsvcClient::write_n (tcp_fd_, cbuffer_, nbytes)) != nbytes) {
#ifdef _RSVC_DEBUG
      printf ("rsvcClient: wrote %d nbytes and expect to write %d bytes\n",
	      nw, nbytes);
#endif
      status = RSVC_ERROR;
    }
    else 
      // register callback and so on
      cmdCbkList_.add ((void *)cbkid);
  }
  if (status != RSVC_SUCCESS) {
    delete cbkid;
    delete cbk;
  }
    
  return status;
}


int
rsvcClient::monitorValue (char* name, rsvcData& data,
			  rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else
    status = monitorCallback ((int)RSVC_MONITOR_ON, data,
			      func, arg);
  return status;
}

int
rsvcClient::monitorOffValue (char* name, rsvcData& data, 
			     rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else
    status = monitorOffCallback ((int)RSVC_MONITOR_OFF, data,
				 func, arg);
  return status;
}

int
rsvcClient::monitorIncomingEntries (char* name, rsvcData& data,
				    rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;
  
  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else
    status = monitorCallback ((int)RSVC_MONITOR_ENTRIES, data,
			      func, arg);

  return status;
}

int
rsvcClient::monitorOffIncomingEntries (char* name, rsvcData& data,
				       rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;
  
  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  else
    status = monitorOffCallback ((int)RSVC_MONITOR_OFFENTRIES, data,
				 func, arg);

  return status;
}


int
rsvcClient::monitorCallback (int opcode, rsvcData& data,
			     rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;
  size_t nbytes = 0;
  size_t nw = 0;

  // this callback is in the table already
  if (monitorCbkInTable (func, arg) != 0) {
    printf ("rsvcClient: monitor callback function already in use\n");
    return RSVC_INVALIDARG;
  }

  rsvcCallback* cbk = new rsvcCallback (func, arg);
  rsvcCbk*      cbkid = new rsvcCbk (opcode, cbkid_++,
				     reqid_++, tcp_fd_, 0,
				     RSVC_SUCCESS, (void *)cbk);


  // create netdata
  rsvcNetData   netdata (data, *cbkid);
    
  if ((nbytes = streamData (netdata)) <= 0)
    status = RSVC_ERROR;
  else {
    // block write
    if ((nw = rsvcClient::write_n (tcp_fd_, cbuffer_, nbytes)) != nbytes) {
#ifdef _RSVC_DEBUG
      printf ("rsvcClient: wrote %d nbytes and expect to write %d bytes\n",
	      nw, nbytes);
#endif
      status = RSVC_ERROR;
    }
    else 
      // register callback and so on
      monitorCbkTable_.add (cbkid);
  }
  
  if (status != RSVC_SUCCESS) {
    delete cbkid;
    delete cbk;
  }
  return status;
}

int
rsvcClient::monitorOffCallback (int opcode, rsvcData& data,
				rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;
  size_t nbytes = 0;
  size_t nw = 0;
  rsvcCbk* cbk = 0;


  if ((cbk = monitorCbkInTable (func, arg)) == 0)
    return RSVC_INVALIDARG;

  // create netdata
  // change cbk opcode into monitor off
  cbk->opcode (opcode);
  rsvcNetData   netdata (data, *cbk);
    
  if ((nbytes = streamData (netdata)) <= 0)
    status = RSVC_ERROR;
  else {
    // block write
    if ((nw = rsvcClient::write_n (tcp_fd_, cbuffer_, nbytes)) != nbytes) {
#ifdef _RSVC_DEBUG
      printf ("rsvcClient: wrote %d nbytes and expect to write %d bytes\n",
	      nw, nbytes);
#endif
      status = RSVC_ERROR;
    }
  }

  return status;
}

rsvcCbk*
rsvcClient::monitorCbkInTable (rsvcCbkFunc func, void* arg)
{
  rsvcHashIterator ite (monitorCbkTable_);
  rsvcCbk*         tcbk = 0;
  rsvcCallback*    usrcbk0 = 0;
  int              found = 0;
  rsvcCallback     tmp (func, arg);

  for (ite.init (); !ite; ++ite) {
    tcbk = (rsvcCbk *) ite ();
    usrcbk0 = (rsvcCallback *)tcbk->userptr ();
    if (*usrcbk0 == tmp) {
      found = 1;
      break;
    }
  }
  if (found) 
    return tcbk;
  return 0;
}

int
rsvcClient::monitorAttr (char* name, char* attr, rsvcData& data,
			 rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;

  
  if (data.insert (RSVC_MONITOR_TAG, attr) != RSVC_SUCCESS)
    status = RSVC_ERROR;

  if (status == RSVC_SUCCESS) 
    status = monitorCallback ((int)RSVC_MONITOR_ONATTR, data,
			      func, arg);
  return status;  
}

int
rsvcClient::monitorOffAttr (char* name, char* attr, rsvcData& data,
			    rsvcCbkFunc func, void* arg)
{
  int status = RSVC_SUCCESS;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  
  if (data.insert (RSVC_MONITOR_TAG, attr) != RSVC_SUCCESS)
    status = RSVC_ERROR;

  if (status == RSVC_SUCCESS) 
    status = monitorOffCallback ((int)RSVC_MONITOR_OFFATTR, data,
				 func, arg);    
  return status;
}

int
rsvcClient::query (char* name, char* qmsg,
		   rsvcCbkFunc func, void *arg)
{
  int status = RSVC_SUCCESS;
  rsvcData data;

  if (data.insert (RSVC_TABLE_NAME, name) != RSVC_SUCCESS)
    status = RSVC_ERROR;
  
  if (data.insert (RSVC_QUERY_TAG, qmsg) != RSVC_SUCCESS)
    status = RSVC_ERROR;

  if (status == RSVC_SUCCESS) 
    status = commandCallback ((int)RSVC_QUERY, data,
			      func, arg);
  return status;  
}  

void
rsvcClient::shutdownServer (void)
{
  rsvcData data;
  
  commandCallback ((int)RSVC_SERVER_EXIT, data, 0, 0);
}

int
rsvcClient::streamData (rsvcNetData& data)
{
  size_t datasize = data.streamSize ();

  if (datasize > cbuflen_) {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: Streamed data size %d is exceeding %d, resize\n",
	    datasize, cbuflen_);
#endif
    delete []cbuffer_;
    cbuflen_ = 2*datasize;
    cbuffer_ = new char[cbuflen_];
    if (!cbuffer_) {
      fprintf (stderr, "rsvcClient: malloc error on stream data\n");
      exit (1);
    }
  }
  if (data.streamOut (cbuffer_, datasize) != RSVC_SUCCESS) {
#ifdef _RSVC_DEBUG
    printf ("rsvcClient: rsvcNetData stream out error\n");
#endif
    return -1;
  }
  return datasize;
}

int
rsvcClient::write_n (int fd, const char* buffer, size_t len)
{
  size_t bytes_written;
  int    n;

  for (bytes_written = 0; bytes_written < len; bytes_written += n)
    {
#ifdef _WIN32
      n = ::send (fd,  buffer + bytes_written, len - bytes_written, 0);
#else
      n = ::write (fd, buffer + bytes_written, len - bytes_written);
#endif
      if (n == -1) {
	return -1;
      }
    }
  return bytes_written;
}

int
rsvcClient::read_n (int fd, char* buffer, size_t len)
{
  size_t bytes_read;
  int    n = 0;

  for (bytes_read = 0; bytes_read < len; bytes_read += n)
    {
#ifdef _WIN32
      n = ::recv (fd, buffer + bytes_read, len - bytes_read, 0);
#else
      n = ::read (fd, buffer + bytes_read, len - bytes_read);
#endif
      if (n == -1) 
	return -1;
      else if (n == 0)
        break;
    }
  return bytes_read;
}

void
rsvcClient::lock (void)
{
  lock_ = 1;
}

void
rsvcClient::unlock (void)
{
  lock_ = 0;
}


//============================================================
//    Implemenation of rsvcClientLocker Class
//============================================================
rsvcClientLocker::rsvcClientLocker (rsvcClient* client)
  :cl_ (client)
{
  cl_->lock ();
}

rsvcClientLocker::~rsvcClientLocker (void)
{
  cl_->unlock ();
}
  

    
  
  
