//-----------------------------------------------------------------------------
// 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 rsvcIO Class
//
// Author:  Jie Chen
//
// Revision History:
//   rsvcIO.cc,v
// Revision 1.8  1998/03/31  18:40:21  chen
// add monitorEntryOff
//
// Revision 1.7  1998/03/19  18:31:00  chen
// add monitorEntry capability to server table
//
// Revision 1.6  1998/02/26  19:10:52  chen
// minor changes
//
// Revision 1.5  1998/02/25  19:53:12  chen
// Put more debug information
//
// Revision 1.4  1998/02/24  19:49:34  chen
// turn on some dbug printf
//
// Revision 1.3  1998/02/20  19:30:22  chen
// Fix a bug on blocked write by resetting errno
//
// Revision 1.2  1998/01/22  20:04:07  chen
// to make it work with cdevReactor
//
// Revision 1.1  1998/01/22  17:08:14  akers
// Addition of new NameServer
//
//
//
#include <rsvcDataStoreMem.h>
#include <rsvcAcceptor.h>
#include "rsvcIO.h"

#ifdef _WIN32
#include <sys/timeb.h>
#endif

extern int rsvc_finished;

rsvcIO::rsvcIO (cdevReactor & r, rsvcAcceptor* acceptor,
		rsvcDataStoreTable& table)
:cdevEventHandler (), sockStream_ (), reactor_ (r), acceptor_ (acceptor),
 clientHost_ (0), dataQueue_ (), wmsg_ (), storeTable_ (table),
 wmaskSet_ (0)
{
#ifdef _TRACE_OBJECTS
  printf ("    Create rsvcIO Class Object\n");
#endif

  // create output buffer
  dbuffer_ = new char[1024];
  dbuflen_ = 1024;
  wmsg_.attach (dbuffer_, dbuflen_, 0);

  // create input buffer
  ibuffer_ = new char[1024];
  ibuflen_ = 1024;
}

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

  handleClose ();

  sockStream_.close ();

  if (clientHost_)
    delete []clientHost_;

  if (dbuffer_)
    delete []dbuffer_;

  if (ibuffer_)
    delete []ibuffer_;

  // clean out unfinished data items in the data queue
  rsvcNetData* data = 0;
  while (!dataQueue_.isEmpty ()) {
    data = (rsvcNetData *)dataQueue_.dequeue ();
    delete data;
  }
}

rsvcIO::operator cdevSocketStream & (void)
{
  return sockStream_;
}

int
rsvcIO::peerAddr (const cdevInetAddr & addr)
{
  if (clientHost_)
    delete []clientHost_;

  clientHost_ = new char[strlen (addr.getHostName ()) + 1];
  strcpy (clientHost_, addr.getHostName ());
  return RSVC_SUCCESS;
}

int
rsvcIO::getHandle (void) const
{
  return sockStream_.getHandle ();
}

int
rsvcIO::setSockOption (int level, int option, void *optval, int optlen) const
{
  return sockStream_.setOption (level, option, optval, optlen);
}

int
rsvcIO::sendToPeer (rsvcNetData* data)
{
#ifdef _CDEV_MANAGE_SERVERS
  if (dataQueue_.count () > 50) {
    fprintf (stderr, "Fatal: Client from %s at port %d is not responding\n",
	     clientHost_, getHandle ());
    fprintf (stderr, "Fatal: Server close down this connection %d\n", getHandle ());
    shutdownIO ();

    return 0;
  }
#endif
    
  // put this data to the end of queue
  dataQueue_.enqueue ((void *)data);

  // if write mask is set, the socket is write blocked.
  // I do not want to call write directly
  if (!wmaskSet_)
    handleOutput ();
  return RSVC_SUCCESS;
}

int
rsvcIO::handleClose ( void )
{
  // ask all data store to handle close
  storeTable_.handleClose ((void *)this);

  // check all callback lists and remove all related callbacks
  storeTable_.monitorOff ((void *)this);
  
  // remove this client from client list
  if (!acceptor_->listLock_)
    acceptor_->remove (this);  

  // get current time to print out
#ifdef _WIN32
  struct _timeb tv;
  _ftime (&tv);
  long ts = (long)tv.time;
#else  
  struct timeval tv;
  gettimeofday (&tv, 0);
  long ts = (long)tv.tv_sec;
#endif
  printf ("At %s rsvcIO: client %s disconnects at port %d from server\n",
	  ctime (&ts), clientHost_, getHandle ());

  // deallocate all the resources since this one is created dynamically.
  // Reactor will only remove it from its array and calls this handleClose
  // delete this;



  return 0;
}

int
rsvcIO::writeStreamBuffer (void)
{
  int nbytes = 0;

  while (wmsg_.size () > 0) {
#ifdef _RSVC_DEBUG
    printf ("rsvcIO: Trying to write %u bytes to output socket %d\n", 
	    wmsg_.size (), getHandle ());
#endif
    nbytes = sockStream_.send (wmsg_.actionPtr (), wmsg_.size ());
    if (nbytes <= 0) {
      if (errno == EWOULDBLOCK || errno == EAGAIN) {
//#ifdef _RSVC_DEBUG
	printf ("rsvcIO: Unable to write on to the output socket %d, break out\n",
		getHandle ());
//#endif
	// reset errno to prevent the system to ckeck these errno again
	errno = 0;
	return 0;
      }
      else {
//#ifdef _RSVC_DEBUG
	printf ("rsvcIO: Writing to a socket %d error %d\n", getHandle(), errno);
//#endif
	errno = 0;
	return -1;
      }
    }
    else {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: Wrote %d bytes to socket %d in the rsvcIO\n", nbytes,
	      getHandle ());
#endif
      wmsg_.actionPtr (nbytes);  // move pointer ahead
    }
  }
  // reset the msg buffer
  wmsg_.reset ();
  return 1;
}


int
rsvcIO::handleOutput ( void )
{
  // check whether write mask is set or not, if yes, remove the mask
  // without calling handleClose
  int wstatus;
  if (wmaskSet_ == 1) {
    setMask(READ_MASK);
    wmaskSet_ = 0;
  }

  if (wmsg_.active ()) {
    if ((wstatus = writeStreamBuffer ()) == 0) {
      // unable to write, turn on write mask
      setMask(WRITE_MASK);
      wmaskSet_ = 1;
      return 0;
    }
    else if (wstatus == -1)                     // write error
      return -1;
    else {                                      // success, remove from queue
      rsvcNetData* data = (rsvcNetData *)dataQueue_.dequeue ();
      if (data == 0) {
	fprintf (stderr, "Fatal: data queue cannot be empty here\n");
      }
      else
	delete data;
    }
  }
  
  while (!dataQueue_.isEmpty ()) {
    // take the first netdata and convert to stream msg
    rsvcNetData* data = (rsvcNetData *)dataQueue_.front ();
    size_t datasize = data->streamSize ();

    // resize buffer if necessary
    if (datasize > dbuflen_) {
#ifdef _RSVC_DEBUG
      printf (" data size is greater that %d resize databuffer to %d\n",
	      2*datasize);
#endif
      delete []dbuffer_;
      dbuflen_ = 2*datasize;
      dbuffer_ = new char[dbuflen_];
    }
    // stream network data
    if (data->streamOut (dbuffer_, datasize) != RSVC_SUCCESS) {
      fprintf (stderr, "FATAL: Stream error for network data\n");
      data = (rsvcNetData *)dataQueue_.dequeue ();
      delete data;
    }
    else {
      wmsg_.attach (dbuffer_, datasize, 0);
	
      if ((wstatus = writeStreamBuffer ()) == 0) {
	// unable to write, turn on write mask
	setMask(WRITE_MASK);
	wmaskSet_ = 1;
	return 0;
      }
      else if (wstatus == -1)
	return -1;
      else {
	rsvcNetData* data = (rsvcNetData *)dataQueue_.dequeue ();
	delete data;
      }
    }
  }
  return 0;
}

int
rsvcIO::handleInput ( void )
{
  rsvcNetData recvdata;
  int         n = 0;
  int         err = 0;

  // first read header length information
  char header[32];
  n = sockStream_.recv_n (header, rsvcNetData::headerLen ());
  if (n != rsvcNetData::headerLen ()) {
    //#ifdef _RSVC_DEBUG
    printf ("rsvcIO:Receiving net data header error:expect %d bytes got %d\n",
	    rsvcNetData::headerLen (), n);
    //#endif

    shutdownIO ();
    // above shutdown will trigger handleClose, no need to
    // return -1 to trigger handleClose again
    return 0;
  }
  
  size_t datasize;
  if (rsvcNetData::readHeader (header, n, &datasize) != RSVC_SUCCESS) {
    //#ifdef _RSVC_DEBUG
    printf ("rsvcIO: Reading rsvcNetData Header error in rsvcIO handleInput\n");
    //#endif
    shutdownIO ();
    // above shutdown will trigger handleClose, no need to
    // return -1 to trigger handleClose again
    return 0;    
  }
  
  if (datasize > ibuflen_) {
#ifdef _RSVC_DEBUG
    printf ("rsvcIO: Input data size %d > buffer size %d, resize buffer size to %d\n",
	    datasize, ibuflen_, 2*datasize);
#endif
    delete []ibuffer_;
    ibuflen_ = 2*datasize;
    ibuffer_ = new char[ibuflen_];
  }
  
  n = sockStream_.recv_n (ibuffer_, datasize);
  if (n != datasize) {
    //#ifdef _RSVC_DEBUG
    printf ("rsvcIO: Received %d bytes, expecting %d bytes\n", n, datasize);
    //#endif
    shutdownIO ();
    // above shutdown will trigger handleClose, no need to
    // return -1 to trigger handleClose again
    return 0;
  }

#ifdef _RSVC_DEBUG
  printf ("rsvcIO: Received data with size %d from port %d\n", datasize,
	  getHandle ());
#endif
  if (recvdata.streamIn (ibuffer_, datasize) != RSVC_SUCCESS) {
    //#ifdef _RSVC_DEBUG
    printf ("rsvcIO: stream data in error\n");
    //#endif
    shutdownIO ();
    // above shutdown will trigger handleClose, no need to
    // return -1 to trigger handleClose again
    return 0;
  }
  
  // do something with recvdata here
  return processData (recvdata);
}

int
rsvcIO::processData (rsvcNetData& ndata)
{
  int status = 0;
  rsvcCbk& cbk = ndata.cbk ();
  // assign socket id to the cbk object
  cbk.userptr ((void *)this);

  switch (cbk.opcode ()) {
  case RSVC_CREATE_MEMTABLE:
    status = createMemTable (ndata);
    break;
  case RSVC_OPEN_DBASE:
    status = openTable (ndata);
    break;
  case RSVC_INSERT:
    status = insertValue (ndata, 0);
    break;
  case RSVC_OVERWRITE:
    status = insertValue (ndata, 1);
    break;
  case RSVC_GET:
    status = getValue (ndata);
    break;
  case RSVC_DEL:
    status = delValue (ndata);
    break;
  case RSVC_SET:
    status = setValue (ndata);
    break;
  case RSVC_MONITOR_ON:
    status = monitorValue (ndata);
    break;
  case RSVC_MONITOR_OFF:
    status = monitorOffValue (ndata);
    break;
  case RSVC_MONITOR_ONATTR:
    status = monitorAttr (ndata);
    break;
  case RSVC_MONITOR_OFFATTR:
    status = monitorOffAttr (ndata);
    break;
  case RSVC_MONITOR_ENTRIES:
    status = monitorEntries (ndata);
    break;
  case RSVC_MONITOR_OFFENTRIES:
    status = monitorOffEntries (ndata);
    break;
  case RSVC_QUERY:
    status = query (ndata);
    break;
  case RSVC_SERVER_EXIT:
    rsvc_finished = 1;
    break;
  default:
    status = -1;
    break;
  }
  return status;
}
    

int
rsvcIO::createMemTable (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) != 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is already here\n", tablename);
#endif
      cbk.cbkstatus (RSVC_INVALIDOP);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else {
      datastore = new rsvcDataStoreMem (tablename);
      storeTable_.add (datastore);
      
      // now create data store table definition
      datastore->createDatabase (ndata, retdata, &num);
    }
  }
  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}


int
rsvcIO::openTable (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      datastore = rsvcDataStore::createDataStore (tablename);
      if (!datastore) {
	cbk.cbkstatus (RSVC_INVALIDOP);
	retdata[0] = new rsvcNetData (cbk);
	num = 1;
      }
      else {
#ifdef _RSVC_DEBUG
	printf ("rsvcIO: add database \"%s\" to the index table\n", tablename);
#endif
	storeTable_.add (datastore);
	// now open database
	datastore->openDatabase (ndata, retdata, &num);
      }
    }
    else {
      // now open database
      datastore->openDatabase (ndata, retdata, &num);
    }
  }
  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}

int
rsvcIO::insertValue (rsvcNetData& ndata, int overwrite)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->putValue (ndata, retdata, &num, overwrite);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}

int
rsvcIO::getValue (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->getValue (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}


int
rsvcIO::delValue (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->delValue (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}


int
rsvcIO::setValue (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->setValue (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}


int
rsvcIO::monitorValue (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->monitorValue (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}


int
rsvcIO::monitorEntries (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->monitorIncomingEntries (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}

int
rsvcIO::monitorAttr (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->monitorAttr (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}

int
rsvcIO::monitorOffValue (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->monitorOffValue (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}

int
rsvcIO::monitorOffEntries (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->monitorOffIncomingEntries (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}


int
rsvcIO::monitorOffAttr (rsvcNetData& ndata)
{
  // get real data and callback information
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[1024];
  size_t       num = 1024;
  int          i = 0;

  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->monitorOffAttr (ndata, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);

  return 0;
}

int
rsvcIO::query (rsvcNetData& ndata)
{
  rsvcData& data = ndata.data ();
  rsvcCbk&  cbk  = ndata.cbk ();
  rsvcNetData* retdata[10240];
  size_t       num = 10240;
  int          i = 0;


  rsvcDataStore* datastore = 0;
  // get name of the table
  char tablename[128];
  char qmsg[1024];
  if (data.get ("table", tablename, sizeof (tablename)) != RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }
  else if (data.get (rsvcServerConfig::queryTag (), qmsg, sizeof (qmsg)) !=
	   RSVC_SUCCESS) {
    cbk.cbkstatus (RSVC_ERROR);
    // construct return message
    retdata[0] = new rsvcNetData (cbk);
    num = 1;
  }    
  else {
    if ((datastore = storeTable_.find (tablename)) == 0) {
#ifdef _RSVC_DEBUG
      printf ("rsvcIO: table name %s is not found\n", tablename);
#endif
      cbk.cbkstatus (RSVC_NOTFOUND);
      retdata[0] = new rsvcNetData (cbk);
      num = 1;
    }
    else 
      datastore->query (ndata, qmsg, retdata, &num);
  }

  for (i = 0; i < num; i++) 
    sendToPeer (retdata[i]);
  
  return 0;
}  


void
rsvcIO::shutdownIO (void)
{
  reactor_.removeHandler (this);
}

void
rsvcIO::enable (int num)
{
  sockStream_.setFlags (num);
}

void
rsvcIO::disable (int num)
{
  sockStream_.setFlags (sockStream_.getFlags()^num);
}

