#include <pthread.h>
#include <map>
#include <set>
#include <zlib.h>

#include "SocketInfo.h"
#include "Messages.h"
#include "NewConnections.h"
#include "LogFile.h"
#include "ZLibMalloc.h"
#include "ReplyToClient.h"
#include "GlobalConfigFile.h"
#include "ThreadMonitor.h"
#include "ZLibStats.h"

#include "CommandDispatcher.h"


static const unsigned maxBufferSize = 500000;

const std::string ExternalRequest::messageIdString = "message_id";
const std::string ExternalRequest::commandString = "command";
const std::string ExternalRequest::emptyString;

TclList ExternalRequest::debugDump() const
{
  TclList result = Request::debugDump();
  result<<TclList("message_id", getResponseMessageId().getValue())
        <<TclList("_properties", getProperties());
  return result;
}

/////////////////////////////////////////////////////////////////////
// A buffer stores the bits and pieces of a message as we reassemble
// and decode the message.  There are different buffer types as there
// are different ways to transmit the data.  The buffer represents
// all of the data that this module knows that is specific to each
// connection.
/////////////////////////////////////////////////////////////////////

class Buffer : public FixedMalloc
{
protected:
  SocketInfo * const _socket;
public:
  Buffer(SocketInfo *socket) :
    _socket(socket)
  {
  }
  virtual ExternalRequest *getMessage() = 0;
  virtual void addNewData(std::string newData) = 0;
  virtual bool getAborted() const = 0;
  virtual std::string copyBuffer() const = 0;
  virtual ~Buffer();
};

Buffer::~Buffer()
{
}

/////////////////////////////////////////////////////////////////////
// ZLibBuffer is the preferred way to communicate.  The entire input
// stream is compressed with streaming zlib.  (We don't reset on each
// new message like some programs.)
// A string is transferred by sending the length of the string,
// followed by the body of the string.  The length of the string is
// an integer, in the standard intel byte order.  The body of the 
// string then needs no quoting and can be variable length.
// Send a name value pair by sending the name as a string followed by
// the value as a string.  The name cannot be the empty string.
// A message is one ore more name-value pairs.
// A message is terminated with the integer 0.  This is not ambiguous
// because a name-value pair cannot have the empty string as a name.
// A message cannot have 0 name values pairs.  A message that starts
// with a 0 is reserved for future use.
/////////////////////////////////////////////////////////////////////

class ZLibBuffer : public Buffer
{
private:
  bool _aborted;
  z_stream_s _zlibBuffer;
  // holds the uncompressed, but otherwise uninterpreted data stream.
  std::string _buffer;
  // this is amount of data that we have already parsed from the buffer.
  int _bufferOffset;
  // These hold the request that we are in the process of decoding.
  ExternalRequest *_request;
  unsigned _stringLength;
  std::string _name;
  enum 
    { rsNone,           // No fields yet.  Looking for the first field.
      rsGetNameLength,  // Looking for the start of the next field.
      rsGetNameBody,    // _stringLength is the length of the name.
      rsGetValueLength, 
      rsGetValueBody,   // _stringLength is the length of the value.
      rsComplete        // _request is finished.
    } _requestState;
  void abort();
  bool getStringLength();
  bool getStringBody(std::string &output);
  void checkForProgress();
public:
  ZLibBuffer(SocketInfo *socket);
  ExternalRequest *getMessage();
  void addNewData(std::string newData);
  bool getAborted() const;
  std::string copyBuffer() const;
  ~ZLibBuffer();
};

ZLibBuffer::ZLibBuffer(SocketInfo *socket) :
  Buffer(socket),
  _aborted(false),
  _bufferOffset(0),
  _request(NULL),
  _requestState(rsNone)
{ // zlibMalloc / zlibFree are optional.  The defaults would work.
  // These routines do extra logging and possibly some performance
  // enhancements.
  _zlibBuffer.zalloc = zlibMalloc; //Z_NULL;
  _zlibBuffer.zfree = zlibFree; //Z_NULL;
  _zlibBuffer.opaque = Z_NULL;
  int result = inflateInit(&_zlibBuffer);
  _aborted = result != Z_OK;
  if (_aborted)
    {
      TclList msg;
      msg<<"CommandDispatcher.C"
         <<"inflateInit failed"
         <<(_zlibBuffer.msg?_zlibBuffer.msg:"NULL");
      LogFile::primary().sendString(msg, socket);
    }
}

ZLibBuffer::~ZLibBuffer()
{
  if (_request)
    {
      delete _request;
    }
  inflateEnd(&_zlibBuffer);
}

void ZLibBuffer::abort()
{
  _aborted = true;
  _buffer.erase();
  _bufferOffset = 0;
  if (_request)
    {
      delete _request;
      _request = NULL;
    }
  _requestState = rsNone;
  _name.erase();
}

bool ZLibBuffer::getStringLength()
{ // The size is always 32 bits in the standard intel byte order.
  // That seems to work in g++ for 32 and 64 bit processors.
  if (_buffer.size() - _bufferOffset < sizeof(int))
    {
      return false;
    }
  _stringLength = *(int *)(_buffer.data() + _bufferOffset);
  _bufferOffset += sizeof(int);
  if ((_stringLength < 0) || (_stringLength > maxBufferSize))
    {
      TclList msg;
      msg<<"CommandDispatcher.C"
         <<"Invalid string length"
         <<_stringLength
         <<std::string(_buffer, 0, _bufferOffset)
         <<std::string(_buffer, _bufferOffset, 250);
      LogFile::primary().sendString(msg, _socket);
      abort();
      _stringLength = 0;
      return false;
    }
  return true;
}

bool ZLibBuffer::getStringBody(std::string &output)
{
  if (_buffer.size() - _bufferOffset < _stringLength)
    {
      return false;
    }
  output.assign(_buffer, _bufferOffset, _stringLength);
  _bufferOffset += _stringLength;
  return true;
}

static const std::string s_checkForProgress = "checkForProgress()";
static const std::string s_eraseCommand = "erase command";

void ZLibBuffer::checkForProgress()
{ // This takes in an uncompressed stream of bytes, and builds messages.
  ThreadMonitor::SetState newState(s_checkForProgress);
  while (!_aborted)
    {
      switch (_requestState)
        {
        case rsNone:
          {
            if (!getStringLength())
              {
                return;
              }
            if (!_stringLength)
              { // A key cannot have a length of 0.  This usually marks
                // the end of a record, but we can't have a
                // record with 0 name-value pairs.  This was done on
                // purpose.  A record starting with the integer 0 is reserved
                // for future use.
                _aborted = true;
                return;
              }
            _requestState = rsGetNameBody;
            break;  
          }
        case rsGetNameLength:
          {
            if (!getStringLength())
              {
                return;
              }
            if (!_stringLength)
              { // End of message
                _requestState = rsComplete;
                ThreadMonitor::find().setState(s_eraseCommand);
                _buffer.erase(0, _bufferOffset);
                _bufferOffset = 0;
                return;
              }
            _requestState = rsGetNameBody;
            // Fall through!
          }
        case rsGetNameBody:
          {
            if (!getStringBody(_name))
              {
                return;
              }
            _requestState = rsGetValueLength;
            // Fall through!
          }
        case rsGetValueLength:
          {
            if (!getStringLength())
              {
                return;
              }
            _requestState = rsGetValueBody;
            // Fall through!
          }
        case rsGetValueBody:
          {
            std::string value;
            if (!getStringBody(value))
              {
                return;
              }
            if (!_request)
              {
                _request = new ExternalRequest(_socket);
              }
            _request->getProperties()[_name] = value;
            _name.erase();
            _requestState = rsGetNameLength;
            break;
          }
        case rsComplete:
          {
            return;
          }
        }
    }
}

ExternalRequest *ZLibBuffer::getMessage()
{
  if (_requestState == rsComplete)
    {
      ExternalRequest *result = _request;
      _request = NULL;
      _requestState = rsNone;
      checkForProgress();
      return result;
    }
  return NULL;
}

static const std::string s_addNewData = "addNewData()";

void ZLibBuffer::addNewData(std::string newData)
{ // This handles the decompression, then passes the result on to
  // checkForProgress().
  ThreadMonitor::SetState setState(s_addNewData);
  ZLibStats::beforeDecompression(newData.size());
  _zlibBuffer.next_in = (Bytef *)newData.data();
  _zlibBuffer.avail_in = newData.size();
  bool outputBufferFull = false;
  while ((_zlibBuffer.avail_in || outputBufferFull) && !_aborted)
    {
      const int bufferLength = 1024;
      char buffer[bufferLength];
      _zlibBuffer.next_out = (Bytef *)buffer;
      _zlibBuffer.avail_out = bufferLength;
      _zlibBuffer.total_out = 0;
      int result = inflate(&_zlibBuffer, Z_SYNC_FLUSH);
      // Note:  Z_BUF_ERROR is unavoidable in some cases.  This means that you
      // provided zlib with no new input, and it had no output for you.  This
      // happens when we are finished reading the input to zlib, the last
      // output from zlib perfectly filled the output buffer, and zlib has
      // no more data for us.
      if (!((result == Z_OK) || (outputBufferFull && (result == Z_BUF_ERROR))))
        {
          TclList msg;
          msg<<"CommandDispatcher.C"
             <<"Error on inflate"
             <<(_zlibBuffer.msg?_zlibBuffer.msg:"NULL")
             <<result;
          LogFile::primary().sendString(msg, _socket);
          abort();
        }
      else
        {
          if (_zlibBuffer.total_out)
            {
              ZLibStats::afterDecompression(_zlibBuffer.total_out);
              _buffer.append(buffer, _zlibBuffer.total_out);
              checkForProgress();
            }
          outputBufferFull = !_zlibBuffer.avail_out;
        }
    }
}

bool ZLibBuffer::getAborted() const
{
  return _aborted;
}

std::string ZLibBuffer::copyBuffer() const
{ // We cannot correctly extract anything meaningful from our buffer.  We
  // assume all data is compressed, so we immediately try to decompress it.
  // if it's in a different format, we've jut screwed it up.  In any case we
  // can't recover it.  If someone wants to switch from this mode to another
  // mode, they will need some type of synchronization.  They will have to
  // send the request, wait until they know it has been satisfied, then send
  // new messages in the new format only after that confirmation.
  //
  // Note:  The default mode was explicitly designed not to have this 
  // limitation.  You can start in the default mode, send the command to
  // switch to zlib, then immediately send more requests, in the zlib format.
  return "";
}

/////////////////////////////////////////////////////////////////////
// TextBuffer is a very simple way to send commands to the server.
// This is primarily intended for debugging using telnet.  However,
// this is also the default listener.  If you want to use zlib, or
// something else, you send a simple text command to switch to that 
// mode.
// In the TextBuffer mode, each message is terminated by a carriage
// return followed by a new line.  Each message should be encoded
// like the payload of a url.  Each name and value is rawurlencoded.
// Put an = between the name and value.  Put an & between each
// adjacent pair of names and values.  Include at least one name-
// value pair.  All other symbols are reserved for future use.
// (For example, a properly encoded XML document could not be
// confused for something encoded this way.)
/////////////////////////////////////////////////////////////////////


class TextBuffer : public Buffer
{
  private:
  bool _aborted;
  std::string _buffer;
  TextBuffer &operator =(TextBuffer const &) const;  // Purposely undefined.
  TextBuffer(TextBuffer const &);  // Purposely undefined.
public:
  TextBuffer(SocketInfo *socket) :
    Buffer(socket),
    _aborted(false)
  {
  }
  ExternalRequest *getMessage()
  {
    int messageBreak = _buffer.find("\r\n");
    if (messageBreak < 0)
      {        
        return NULL;
      }
    std::string currentMessageString = _buffer.substr(0, messageBreak);
    _buffer.erase(0, messageBreak + 2);
    ExternalRequest *currentMessage = new ExternalRequest(_socket);
    parseUrlRequest(currentMessage->getProperties(), 
                    currentMessageString);
    return currentMessage;
  }
  void addNewData(std::string newData)
  {
    if (!_aborted)
      {
        _buffer += newData;
        if (_buffer.size() > maxBufferSize)
          {
            _aborted = true;
            _buffer.clear();
          }
      }
  }
  bool getAborted() const
  {
    return _aborted;
  }
  std::string copyBuffer() const
  {
    return _buffer;
  }
};

/////////////////////////////////////////////////////////////////////
// CommandDispatcher::TimerCallbackList
// This is required so we can get a little bit of status back.
// The rule is that we tweek the dead man timer each time a new
// connection is made, and each time the connection sends us a valid
// message.  After a timeout we kill the connection.
/////////////////////////////////////////////////////////////////////

void CommandDispatcher::TimerCallbackList::successfulDispatch(Request *request)
{
  _timer->touchConnection(request->getSocketInfo());
}

CommandDispatcher::TimerCallbackList::TimerCallbackList(DeadManTimer *timer) :
  _timer(timer)
{
}

/////////////////////////////////////////////////////////////////////
// ListenForCommandRequest
// This allows us to send a message to ourself to register a new
// listener.  This avoids locks and makes the common case of
// dispatching a message simple and effecient.  We will only register
// a handful of commands, so we aren't as worried about the cost of
// that.
/////////////////////////////////////////////////////////////////////

class ListenForCommandRequest : public Request
{
private:
  std::string _commandName;
  RequestListener *_listener;
  int _callbackIdForNewCommand;
  bool _lock;
  bool _immuneToLock;
public:
  ListenForCommandRequest(std::string commandName, 
                          RequestListener *listener, 
                          int callbackId,
                          bool lock,
                          bool immuneToLock) :
    Request(NULL),
    _commandName(commandName),
    _listener(listener),
    _callbackIdForNewCommand(callbackId),
    _lock(lock),
    _immuneToLock(immuneToLock)
  {
  }
  std::string getCommandName()
  {
    return _commandName;
  }
  RequestListener *getListener()
  {
    return _listener;
  }
  int getCallbackIdForNewCommand()
  {
    return _callbackIdForNewCommand;
  }
  bool getLock()
  {
    return _lock;
  }
  bool getImmuneToLock()
  {
    return _immuneToLock;
  }
};

/////////////////////////////////////////////////////////////////////
// CallbackList
/////////////////////////////////////////////////////////////////////

void CallbackList::successfulDispatch(Request *)
{
}

void CallbackList::dumpMessage(ExternalRequest *request, 
                               std::string const &reason)
{
  TclList forLog;
  forLog<<FLF<<reason;
  for (PropertyList::const_iterator it = request->getProperties().begin();
       it != request->getProperties().end();
       it++)
    {
      forLog<<it->first
            <<it->second;
    }
  LogFile::primary().sendString(forLog, request->getSocketInfo());
}

static const std::string s_dispatchMessage = "dispatchMessage";

void CallbackList::dispatchMessage(ExternalRequest *request)
{
  ThreadMonitor::SetState setState(s_dispatchMessage);
  SocketInfo *const socket = request->getSocketInfo();
  if (_dumpAll)
    {
      dumpMessage(request, "dump_all");
    }
  std::queue< ExternalRequest * > *queue = 
    getProperty(_locked, socket);
  std::string command = request->getCommand();
  CallbackInfo *callback = getProperty(_commandList, command);
  if (!callback)
    {
      if (_dumpUnknownCommands)
        dumpMessage(request, "UNKNOWN_COMMAND");
      if (_requireLogin && !_initialized.count(socket))
        DeleteSocketThread::deleteSocket(socket);
      delete request;
    }
  else if (queue && !callback->immuneToLock)
    { // This socket is locked.  Queue up the request for later.
      queue->push(request);
    }
  else
    {
      ThreadMonitor::find().increment(command);
      if (callback->lock)
        { // Create an empty queue to lock this.  There should be no
          // way to lock something that is already locked, but if it
          // is already locked, this is a no-op.
          _locked[socket];
          // For simplicity we assume that the same commands which lock the
          // queue also initialize the socket.  In both cases the normal use
          // case is that we log in before we do anything else.
          _initialized.insert(socket);
        }
      if (_requireLogin && 
          !(callback->immuneToLock || _initialized.count(socket)))
        {
          DeleteSocketThread::deleteSocket(socket);
          delete request;
        }
      else
        {
          successfulDispatch(request);
          RequestListener *listener = 
            callback->requestListener;
          request->callbackId = 
            callback->callbackId;
          listener->newRequest(request);
        }
    }
}

Request *CallbackList::createNewCommandRequest(std::string commandName,
                                               RequestListener *listener,
                                               int callbackId,
                                               bool lock,
                                               bool immuneToLock)
{
  return new ListenForCommandRequest(commandName, listener, callbackId, 
                                     lock, immuneToLock);
}

void CallbackList::addNewCommand(Request *request)
{
  ListenForCommandRequest *r =
    dynamic_cast<ListenForCommandRequest *>(request);
  if (_commandList.count(r->getCommandName()))
  {
    TclList msg;
    msg <<FLF<<"duplicate command"<<r->getCommandName();
    LogFile::primary().sendString(msg);
    LogFile::primary().scheduleShutdown();
  }
  CallbackInfo &info = _commandList[r->getCommandName()];
  info.requestListener = r->getListener();
  info.callbackId = r->getCallbackIdForNewCommand();
  info.lock = r->getLock();
  info.immuneToLock = r->getImmuneToLock();
}

void CallbackList::unlock(SocketInfo *socket)
{ // We don't every expect an unlock without a lock.  That case would
  // work, but we don't optimize for it.
  std::queue< ExternalRequest * > queue = _locked[socket];
  _locked.erase(socket);
  if (_dumpAll && !queue.empty())
    {
      TclList msg;
      msg<<__FILE__<<__LINE__<<__FUNCTION__
         <<"dump_all"<<"repeating"<<queue.size();
      LogFile::primary().sendString(msg, socket);
    }
  while (!queue.empty())
    {
      dispatchMessage(queue.front());
      queue.pop();
    }
}

void CallbackList::remove(SocketInfo *socket)
{ // We assume, for the sake of optimization that most of the time
  // the socket will not be locked.  Of course, this still works
  // and does a reasonable, when the socket is locked.
  std::queue< ExternalRequest * > *queue = getProperty(_locked, socket);
  if (queue)
    {
      while (!queue->empty())
        {
          delete queue->front();
          queue->pop();
        }
      _locked.erase(socket);
    }
  _initialized.erase(socket);
}

CallbackList::CallbackList()
{ // By default logging in is optional.  That's how the AX server worked.  Now
  // we have the option of a more traditional server.  The first command has
  // to be a login or you will get disconnected.
  _requireLogin = false;
  // getConfigItem: dump_all=1 means send a copy of every message in or out
  // getConfigItem:   of the system to the log file.  This can be useful in
  // getConfigItem:   debugging, but it makes really big log files very
  // getConfigItem:   quickly.  This setting is shared by several libraries.
  _dumpAll = getConfigItem("dump_all") == "1";
  // getConfigItem: dump_unknown_commands=1 sends a lot of details to the log
  // getConfigItem:   file any time someone sends a command that we don't know
  // getConfigItem:   how to handle.  This was aimed strictly at debugging.
  _dumpUnknownCommands = getConfigItem("dump_unknown_commands") == "1";
}

CallbackList::~CallbackList()
{ 
  for (std::map< SocketInfo *, std::queue< ExternalRequest * > >::iterator it =
         _locked.begin();
       it != _locked.end();
       it++)
    {
      std::queue< ExternalRequest * > &queue = it->second;
      while (!queue.empty())
        {
          delete queue.front();
          queue.pop();
        }
    }
}

/////////////////////////////////////////////////////////////////////
// CommandDispatcher
/////////////////////////////////////////////////////////////////////

void CommandDispatcher::newRequest(Request *request)
{ // This function catches stuff we want to happen immediately.  In particular,
  // the client might send a request to change input modes followed immediately
  // a request in the new input mode.  So we must change modes immediately.
  switch (request->callbackId)
    {
    case mtSetInput:
      {
        ExternalRequest *r = dynamic_cast<ExternalRequest *>(request);
        SocketInfo *socket = r->getSocketInfo();
        std::string mode = r->getProperty("mode", "");
        if (mode == "zlib")
          {
            Buffer *&buffer = _allBuffers[socket];
            Buffer *newBuffer = new ZLibBuffer(socket);
            if (buffer)
              {
                newBuffer->addNewData(buffer->copyBuffer());
                delete buffer;
              }
            buffer = newBuffer;
          }
        else 
          {
            TclList msg;
            msg<<"CommandDispatcher.C"
               <<"Unknown input mode"
               <<mode;
            LogFile::primary().sendString(msg, socket);
            DeleteSocketThread::deleteSocket(socket);
          }
        break;
      }
    case mtProxyInfo:
      {
        ExternalRequest *r = dynamic_cast<ExternalRequest *>(request);
        SocketInfo *socket = r->getSocketInfo();
        std::string original = socket->remoteAddr();
        std::string replacement = r->getProperty("remote");
        std::string auth = r->getProperty("auth");
        TclList msg;
        msg<<"CommandDispatcher.C"
           <<"mtProxyInfo"
           <<"original"
           <<original
           <<"replacement"
           <<replacement
           <<"auth"
           <<auth;
        std::string error;
        if (replacement.empty())
          {
            error = "Invalid replacement";
          }
        else if (auth.empty())
          {
            error = "Invalid authorization";
          }
        if (!error.empty())
          {
            msg<<"error"<<error;
            LogFile::primary().sendString(msg, socket);
            DeleteSocketThread::deleteSocket(socket);
          }
        else
          {
            socket->setRemoteAddr(replacement);
            LogFile::primary().sendString(msg, socket);
          }
      }
    }
  delete request;
}

void CommandDispatcher::threadFunction()
{
  while (true)
    {
      _incomingRequests.waitForRequest();
      while (Request *current = _incomingRequests.getRequest())
        {
          switch (current->callbackId)
            {
            case mtNewCommand:
              {
                _callbackList.addNewCommand(current);
                break;
              }
            case mtUnlock:
              {
                _callbackList.unlock(current->getSocketInfo());
                break;
              }
            case mtNewInput:
              { 
                ThreadMonitor::SetState setState("mtNewInput");
                NewInputRequest *r =
                  dynamic_cast<NewInputRequest *>(current);
                SocketInfo *socket = r->getSocketInfo();
                if (r->atEOF())
                  {
                    if (_showEOF)
                      {
                        LogFile::primary().quoteAndSend("CommandDispatcher.C",
                                                        "EOF on read",
                                                        socket);
                      }
                    DeleteSocketThread::deleteSocket(socket);
                  }
                else if (r->newSocket())
                  {
                    _deadManTimer.touchConnection(socket);
                  }
                else
                  {
                    Buffer *&buffer = _allBuffers[socket];
                    if (!buffer)
                      {
                        buffer = new TextBuffer(socket);
                      }
                    buffer->addNewData(r->getNewInput());
                    if (buffer->getAborted())
                      {
                        LogFile::primary().
                          quoteAndSend("CommandDispatcher.C",
                                       "Buffer aborted adding new data",
                                       socket);
                        DeleteSocketThread::deleteSocket(socket);
                      }
                    else
                      {
                        while (ExternalRequest *outgoing = 
                               buffer->getMessage())
                          {
                            _callbackList.dispatchMessage(outgoing);
                          }
                        if (buffer->getAborted())
                          {
                            LogFile::primary().
                              quoteAndSend("CommandDispatcher.C",
                                           "Buffer aborted while retrieving message",
                                           socket);
                            DeleteSocketThread::deleteSocket(socket);
                          }
                      }
                  }
                break;
              }
            case mtQuit:
              {
                delete current;
                return;
              }
            case DeleteSocketThread::callbackId:
              {
                SocketInfo *socket = current->getSocketInfo();
                Buffer *buffer = _allBuffers[socket];
                if (buffer)
                  {
                    delete buffer;
                  }
                _allBuffers.erase(socket);
                _callbackList.remove(socket);
                break;
              }
            }
          delete current;
        }
    }
}

void CommandDispatcher::listenForCommand(std::string commandName,
                                         RequestListener *listener,
                                         int callbackId,
                                         bool lock,
                                         bool immuneToLock)
{
  Request *request = new ListenForCommandRequest(commandName,
                                                 listener,
                                                 callbackId,
                                                 lock,
                                                 immuneToLock);
  request->callbackId = mtNewCommand;
  _incomingRequests.newRequest(request);
}

void CommandDispatcher::unlock(SocketInfo *socket)
{
  Request *request = new Request(socket);
  request->callbackId = mtUnlock;
  _incomingRequests.newRequest(request);
}

CommandDispatcher::CommandDispatcher(std::string const &name) :
  ThreadClass(name),
  _incomingRequests(getName()),
  _callbackList(&_deadManTimer),
  _showEOF(true)
{
  listenForCommand("set_input", this, mtSetInput, false, true);
  listenForCommand("proxy_info", this, mtProxyInfo, false, true);
  connectToReplyToClient(this);

  startThread();
}

CommandDispatcher::~CommandDispatcher()
{
  Request *r = new Request(NULL);
  r->callbackId = mtQuit;
  _incomingRequests.newRequest(r);
}
        
// Don't copy this!  It works, but it's unnecessarily complicated.
// Just make a static variable in getInstance and initialize it in
// place.  G++ will make sure the initializer is called exactly
// once, the first time we get to that line, automatically thread
// safe.  MultiCast::getInstance() is an example of the better
// way to do it.  I didn't use that trick on old code because I
// didn't know about it!
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
CommandDispatcher *CommandDispatcher::_instance = NULL;

CommandDispatcher *CommandDispatcher::getInstance()
{
  if (!_instance)
    {
      pthread_mutex_lock(&mutex);
      if (!_instance)
        {
          _instance = new CommandDispatcher;
        }
      pthread_mutex_unlock(&mutex);
    }
  return _instance;
}