pvvmud/common/lib/srvcli/communicate.C

/*
 * PVVMUD a 3D MUD
 * Copyright (C) 1998-1999  Programvareverkstedet (pvv@pvv.org)
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 */
#include "pvvmud.H"
#include <iostream.h>
#include "communicate.H"

// Debug switches
#define LOGMESSAGES  

CBadProtocolException::CBadProtocolException()
  :CException("Bad protocol exception"){

}

class CWriteQueueItem: public CObject {
  CMessage * m_message;
  int m_delete;

public:
  CWriteQueueItem(CMessage * message,int del){
    m_message = message;
    m_delete = del;
  }

  CMessage * getMessage(){
    return m_message;
  }

  int getDelete(){
    return m_delete;
  }
};

CCommunicate::CCommunicate(CSocket * socket,CTimeKeeper * timeKeeper){

  m_socket = socket;
  stream = new CStream(m_socket->getFileDescriptor());

  msgWriteQueue = new CObjectQueue();

  // Add notification when ready for read or write
  m_timeKeeper = timeKeeper;
  if (m_timeKeeper != NULL){
    m_timeKeeper->addFd(socket->getFileDescriptor(),this,
                        TIMEKEEPER_IN);
  }

  currReadMsg = NULL;
  currWriteMsg = NULL;

  msgList = new CMessageList();
}

CCommunicate::~CCommunicate(){
  if (m_timeKeeper != NULL) m_timeKeeper->rmFd(m_socket->getFileDescriptor());
  if (stream != NULL) delete stream;
  if (m_socket != NULL) delete m_socket;
  if (msgList != NULL){
    msgList->deleteAll();
    delete msgList;
  }
}

CSocket * CCommunicate::getSocket(){
  return m_socket;
}

void CCommunicate::quit(){
  cdebug << "Quit communicate not impl.\n";
}

int CCommunicate::timeKeeperFD(int event){
  try {
    if (event & TIMEKEEPER_IN)  readyToRead();
    if (event & TIMEKEEPER_OUT) readyToWrite();
    return TRUE;
  } catch (CEmptyStreamException * e){
    quit();
    delete e;
    return FALSE;
  } catch (CException * e){
    cdebug << *e << "\n";
    quit();
    delete e;
    return FALSE;
  }
}

void CCommunicate::readyToRead(){
  int blocking = TRUE;
  
  do {
    if (currReadMsg == NULL){
      try {
        int id = stream->peepByte();  
        if ((currReadMsg = msgList->createMsg(id)) == NULL){
          cdebug << "Unknown message id : " << id << "\n";
          throw new CBadProtocolException();
        }
        currReadMsg->setManager(getManager());
      } catch (CEmptyStreamException * e){
        cdebug << *e << "\n";
        delete e;
        return;
      }
    }


    int result = currReadMsg->recive(stream);

    switch (result){
    case MESSAGE_OK:
#ifdef LOGMESSAGES
      cdebug << "Recive(\""<< getName() << "\"): " 
           << currReadMsg->getName() << "\n";
#endif
      currReadMsg->executeMsg(); 
      delete currReadMsg;
      currReadMsg = NULL;
      blocking = FALSE;
      break;
    case MESSAGE_BLOCKING:
      blocking = TRUE;
      break;
    }

  } while (!blocking); // Quit if blocking

}

void CCommunicate::readyToWrite(){
  int blocking = TRUE;

  do {

    // If no message in progres get message from queue
    if (currWriteMsg == NULL){
      CWriteQueueItem * queueItem  = (CWriteQueueItem*)msgWriteQueue->getFIFO();
      // Return if queue is empty
      if (queueItem == NULL){
        timeKeeperSetEvents(TIMEKEEPER_IN);
        return;
      }
      currWriteMsg = queueItem->getMessage(); 
      currWriteMsg->prepareSend();
      currWriteDelete = queueItem->getDelete();
      delete queueItem;
    }

    int result = currWriteMsg->send(stream);
    switch (result){
    case MESSAGE_OK:
#ifdef LOGMESSAGES
      cdebug << "Send(\""<< getName() << "\"): " 
           << currWriteMsg->getName() << "\n";
#endif
      if (currWriteDelete) delete currWriteMsg;
      currWriteMsg = NULL;
      blocking = FALSE;
      break;
    case MESSAGE_BLOCKING:
      timeKeeperSetEvents(TIMEKEEPER_IN | TIMEKEEPER_OUT);
      blocking = TRUE;
      break;
    }

  } while (!blocking); // Quit if blocking 

}

void CCommunicate::sendMessage(CMessage * message,int del){

//  cdebug << "Add to queue : " << message->getName() 
//       << " queue size " << msgWriteQueue->getNumItems() << "\n";

  // Add message to write queue
  msgWriteQueue->add(new CWriteQueueItem(message,del));

  // Try to send new message
  try {
    readyToWrite(); 
  } catch (CException * e){
    cdebug << *e << "\n";
    quit();
    delete e;
  }

}

void CCommunicate::setManager(CManager * manager){
  this->manager = manager;
}

CManager * CCommunicate::getManager(){
  return manager;
}

void CCommunicate::addMsg(MsgCreateFunc_t msgCreateFunc){
  msgList->addMsg(msgCreateFunc);
}