cMsgWrapper.cc

Go to the documentation of this file.

// to do
 
//   subscribe must lock list until done ?
 
 
 
/*---------------------------------------------------------------------------*
*  Copyright (c) 2005        Southeastern Universities Research Association, *
*                            Thomas Jefferson National Accelerator Facility  *
*                                                                            *
*    This software was developed under a United States Government license    *
*    described in the NOTICE file included as part of this distribution.     *
*                                                                            *
*    E.Wolin, 25-Feb-2005, Jefferson Lab                                     *
*                                                                            *
*    Authors: Elliott Wolin                                                  *
*             wolin@jlab.org                    Jefferson Lab, MS-6B         *
*             Phone: (757) 269-7365             12000 Jefferson Ave.         *
*             Fax:   (757) 269-5519             Newport News, VA 23606       *
*                                                                            *
*----------------------------------------------------------------------------*/
 
 
#include <cMsg.hxx>
#include <cMsgPrivate.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <vector>
#include <sstream>
 
using namespace std;
using namespace cmsg;
 
 
 
//-----------------------------------------------------------------------------
//  local data and static functions
//-----------------------------------------------------------------------------
 
 
namespace cmsg {
 
 
typedef struct {
  cMsgCallback *cb;      
  void *userArg;         
} dispatcherStruct;
 
 
typedef struct {
  void  *domainId;      
  void  *handle;        
  string subject;       
  string type;          
  dispatcherStruct *d;  
} subscrStruct;
 
 
static vector<subscrStruct*> subscrVec;
 
 
static pthread_mutex_t subscrMutex = PTHREAD_MUTEX_INITIALIZER;
 
 
//-----------------------------------------------------------------------------
 
 
static void callbackDispatcher(void *msg, void *userArg) {
  dispatcherStruct *ds = (dispatcherStruct*)userArg;
  ds->cb->callback(new cMsgMessage(msg),ds->userArg);
}
 
 
//-----------------------------------------------------------------------------
 
 
 /*
static bool subscriptionExists(void *domainId, const string &subject, const string &type, 
                               cMsgCallback *cb, void *userArg) {
 
  bool itExists = false;
 
 
  // search list for matching subscription
  pthread_mutex_lock(&subscrMutex);
  for(unsigned int i=0; i<subscrVec.size(); i++) {
    if( (subscrVec[i]->domainId   == domainId) &&
        (subscrVec[i]->subject    == subject) &&
        (subscrVec[i]->type       == type) &&
        (subscrVec[i]->d->cb      == cb) &&
        (subscrVec[i]->d->userArg == userArg)
        ) {
      itExists=true;
      break;
    }
  }
  pthread_mutex_unlock(&subscrMutex);
 
 
  // done searching
  return(itExists);
}
 */
 
//-----------------------------------------------------------------------------
 
 
static void addSubscription(void *domainId, const string &subject, const string &type,
                            dispatcherStruct *d, void *handle) {
 
  subscrStruct *s = new subscrStruct();
 
  s->domainId=domainId;
  s->subject=subject;
  s->type=type;
  s->d=d;
  s->handle=handle;
 
  pthread_mutex_lock(&subscrMutex);
  subscrVec.push_back(s);
  pthread_mutex_unlock(&subscrMutex);
 
  return;
}
 
 
//-----------------------------------------------------------------------------
 
 
static bool deleteSubscription(void *domainId, void *handle) {
 
  bool deleted = false;
 
  pthread_mutex_lock(&subscrMutex);
  vector<subscrStruct*>::iterator iter;
  for(iter=subscrVec.begin(); iter!=subscrVec.end(); iter++) {
    if(((*iter)->domainId==domainId)&&((*iter)->handle==handle)) {
      delete((*iter)->d);
      delete(*iter);
      subscrVec.erase(iter);
      deleted=true;
      break;
    }
  }
  pthread_mutex_unlock(&subscrMutex);
 
  return(deleted);
}
 
 
//-----------------------------------------------------------------------------
 
 
static void deleteSubscriptions(void *domainId) {
 
  pthread_mutex_lock(&subscrMutex);
  vector<subscrStruct*>::iterator iter;
  for(iter=subscrVec.begin(); iter!=subscrVec.end(); iter++) {
    if((*iter)->domainId==domainId) {
      delete((*iter)->d);
      delete(*iter);
      subscrVec.erase(iter);
    }
  }
  pthread_mutex_unlock(&subscrMutex);
 
  return;
}
 
 
//-----------------------------------------------------------------------------
 
} // namespace cmsg
 
 
 
//-----------------------------------------------------------------------------
//  cMsgException methods
//-----------------------------------------------------------------------------
 
 
cMsgException::cMsgException(void) : descr(""), returnCode(0) {}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgException::cMsgException(const string &c) : descr(c), returnCode(0) {}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgException::cMsgException(const string &c, int code) : descr(c), returnCode(code) {}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgException::cMsgException(const cMsgException &e) : descr(e.descr),  returnCode(e.returnCode) {}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgException::~cMsgException(void) throw() {
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgException::toString(void) const throw() {
  stringstream ss;
  ss << "?cMsgException returnCode = " << returnCode << "    descr = " << descr << ends;
  return(ss.str());
}
 
 
//-----------------------------------------------------------------------------
 
 
const char *cMsgException::what(void) const throw() {
  return(toString().c_str());
}
 
 
//-----------------------------------------------------------------------------
//  cMsgMessage methods
//-----------------------------------------------------------------------------
 
 
cMsgMessage::cMsgMessage(void) throw(cMsgException) {
    
  myMsgPointer=cMsgCreateMessage();
  if(myMsgPointer==NULL) {
    throw(cMsgException("?cMsgMessage constructor...unable to create message",CMSG_ERROR));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage::cMsgMessage(const cMsgMessage &msg) throw(cMsgException) {
 
  myMsgPointer=cMsgCopyMessage(msg.myMsgPointer);
  if(myMsgPointer==NULL) {
    throw(cMsgException("?cMsgMessage copy constructor...unable to create message",CMSG_ERROR));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage::cMsgMessage(void *msgPointer) throw(cMsgException) {
 
  myMsgPointer=msgPointer;
  if(myMsgPointer==NULL) {
    throw(cMsgException("?cMsgMessage pointer constructor...unable to create message",CMSG_ERROR));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage::~cMsgMessage(void) {
  if(myMsgPointer!=NULL)cMsgFreeMessage(&myMsgPointer);
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getSubject(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetSubject(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setSubject(const string &subject) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetSubject(myMsgPointer,subject.c_str()))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getType(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetType(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
  
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setType(const string &type) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetType(myMsgPointer,type.c_str()))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getText(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetText(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setText(const string &text) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetText(myMsgPointer,text.c_str()))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setByteArrayLength(int length) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetByteArrayLength(myMsgPointer,length)!=CMSG_OK)) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::resetByteArrayLength() {
  cMsgResetByteArrayLength(myMsgPointer);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgMessage::getByteArrayLength(void) {
 
  int i;
  cMsgGetByteArrayLength(myMsgPointer,&i);
  return(i);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgMessage::getByteArrayLengthFull(void) {
  int i;
  cMsgGetByteArrayLengthFull(myMsgPointer,&i);
  return(i);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setByteArrayOffset(int offset) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetByteArrayOffset(myMsgPointer,offset)!=CMSG_OK)) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgMessage::getByteArrayOffset(void) {
 
  int i;
  cMsgGetByteArrayOffset(myMsgPointer,&i);
  return(i);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setByteArray(char *array, int length) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetByteArray(myMsgPointer,array, length)!=CMSG_OK)) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setByteArrayNoCopy(char* array, int length) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetByteArrayNoCopy(myMsgPointer,array,length)!=CMSG_OK)) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
char* cMsgMessage::getByteArray(void) {
 
  char *p;
  cMsgGetByteArray(myMsgPointer,&p);
  return(p);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgMessage::getByteArrayEndian(void) {
 
  int endian;
  cMsgGetByteArrayEndian(myMsgPointer,&endian);
  return(endian);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setByteArrayEndian(int endian) throw(cMsgException) {
  int stat;
 
  if((stat=cMsgSetByteArrayEndian(myMsgPointer,endian))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  return;
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsgMessage::needToSwap(void) const throw(cMsgException) {
 
  int flag,stat;
 
  if((stat=cMsgNeedToSwap(myMsgPointer,&flag))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  return(flag==1);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgMessage::getUserInt(void) const throw(cMsgException) {
 
  int i;
 
  int stat;
  if((stat=cMsgGetUserInt(myMsgPointer,&i))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(i);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setUserInt(int i) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetUserInt(myMsgPointer,i))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
struct timespec cMsgMessage::getUserTime(void) const throw(cMsgException) {
 
  struct timespec t;
 
  int stat;
  if((stat=cMsgGetUserTime(myMsgPointer,&t))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(t);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setUserTime(const struct timespec &userTime) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetUserTime(myMsgPointer, &userTime))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgMessage::getVersion(void) const throw(cMsgException) {
 
  int version;
 
  int stat;
  if((stat=cMsgGetVersion(myMsgPointer, &version))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(version);
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage *cMsgMessage::copy(void) const throw(cMsgException) {
 
  void *newPointer = cMsgCopyMessage(myMsgPointer);
  return(new cMsgMessage(newPointer));
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getDomain(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetDomain(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getReceiver(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetReceiver(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getReceiverHost(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetReceiverHost(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getSender(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetSender(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getSenderHost(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetSenderHost(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
struct timespec cMsgMessage::getReceiverTime(void) const throw(cMsgException) {
 
  struct timespec t;
 
  int stat;
  if((stat=cMsgGetReceiverTime(myMsgPointer,&t))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(t);
}
 
 
//-----------------------------------------------------------------------------
 
 
struct timespec cMsgMessage::getSenderTime(void) const throw(cMsgException) {
 
  struct timespec t;
 
  int stat;
  if((stat=cMsgGetSenderTime(myMsgPointer,&t))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(t);
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsgMessage::isGetRequest(void) const throw(cMsgException) {
  
  int b;
 
  int stat;
  if((stat=cMsgGetGetRequest(myMsgPointer,&b))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(b);
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsgMessage::isGetResponse(void) const throw(cMsgException) {
  
  int b;
 
  int stat;
  if((stat=cMsgGetGetResponse(myMsgPointer,&b))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(b);
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsgMessage::isNullGetResponse(void) const throw(cMsgException) {
  
  int b;
 
  int stat;
  if((stat=cMsgGetNullGetResponse(myMsgPointer,&b))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(b);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::makeNullResponse(const cMsgMessage &msg) throw(cMsgException) {
 
  cMsgMessage_t *t = (cMsgMessage_t*)myMsgPointer;
  cMsgMessage_t *m = (cMsgMessage_t*)msg.myMsgPointer;
 
  t->sysMsgId    = m->sysMsgId;
  t->senderToken = m->senderToken;
  t->info        = CMSG_IS_GET_RESPONSE | CMSG_IS_NULL_GET_RESPONSE;
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::makeNullResponse(const cMsgMessage *msg) throw(cMsgException) {
 
  cMsgMessage_t *t = (cMsgMessage_t*)myMsgPointer;
  cMsgMessage_t *m = (cMsgMessage_t*)msg->myMsgPointer;
 
  t->sysMsgId    = m->sysMsgId;
  t->senderToken = m->senderToken;
  t->info        = CMSG_IS_GET_RESPONSE | CMSG_IS_NULL_GET_RESPONSE;
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::makeResponse(const cMsgMessage &msg) throw(cMsgException) {
 
  cMsgMessage_t *t = (cMsgMessage_t*)myMsgPointer;
  cMsgMessage_t *m = (cMsgMessage_t*)msg.myMsgPointer;
 
  t->sysMsgId    = m->sysMsgId;
  t->senderToken = m->senderToken;
  t->info        = CMSG_IS_GET_RESPONSE;
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::makeResponse(const cMsgMessage *msg) throw(cMsgException) {
 
  cMsgMessage_t *t = (cMsgMessage_t*)myMsgPointer;
  cMsgMessage_t *m = (cMsgMessage_t*)msg->myMsgPointer;
 
  t->sysMsgId    = m->sysMsgId;
  t->senderToken = m->senderToken;
  t->info        = CMSG_IS_GET_RESPONSE;
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage *cMsgMessage::nullResponse(void) const throw(cMsgException) {
 
  void *newMsgPointer;
  if((newMsgPointer=cMsgCreateNullResponseMessage(myMsgPointer))==NULL) {
    throw(cMsgException("?cMsgMessage::nullResponse...unable to create message",CMSG_ERROR));
  }
 
  return(new cMsgMessage(newMsgPointer));
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage *cMsgMessage::response(void) const throw(cMsgException) {
 
  void *newMsgPointer;
  if((newMsgPointer=cMsgCreateResponseMessage(myMsgPointer))==NULL) {
    throw(cMsgException("?cMsgMessage::response...unable to create message",CMSG_ERROR));
  }
 
  return(new cMsgMessage(newMsgPointer));
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setGetResponse(bool b) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetGetResponse(myMsgPointer,b))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setNullGetResponse(bool b) throw(cMsgException) {
 
  int stat;
  if((stat=cMsgSetNullGetResponse(myMsgPointer,b))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::toString(void) const throw(cMsgException) {
 
  char *cs;
 
  int stat;
  if((stat=cMsgToString(myMsgPointer,&cs))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  string s(cs);
  free(cs);
  return(s);
           
}
 
 
//-----------------------------------------------------------------------------
//   message context accessor functions
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getSubscriptionDomain(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetSubscriptionDomain(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getSubscriptionSubject(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetSubscriptionSubject(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getSubscriptionType(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetSubscriptionType(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsgMessage::getSubscriptionUDL(void) const throw(cMsgException) {
 
  const char *s;
 
  int stat;
  if((stat=cMsgGetSubscriptionUDL(myMsgPointer,&s))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  };
 
  if(s==NULL) {
    return("null");
  } else {
    return(string(s));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgMessage::getSubscriptionCueSize(void) const throw(cMsgException) {
 
  int i;
 
  int stat;
  if((stat=cMsgGetSubscriptionCueSize(myMsgPointer,&i))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(i);
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsgMessage::getReliableSend(void) const throw(cMsgException) {
 
  int i;
 
  int stat;
  if((stat=cMsgGetReliableSend(myMsgPointer,&i))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(i == 0 ? false : true);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgMessage::setReliableSend(bool b) throw(cMsgException) {
 
  int i = b ? 1 : 0;
  
  int stat;
  if((stat=cMsgSetReliableSend(myMsgPointer,i))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
// cMsgSubscriptionConfig methods
//-----------------------------------------------------------------------------
 
 
cMsgSubscriptionConfig::cMsgSubscriptionConfig(void) {
  config = cMsgSubscribeConfigCreate();
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgSubscriptionConfig::~cMsgSubscriptionConfig(void) {
  cMsgSubscribeConfigDestroy(config);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgSubscriptionConfig::getMaxCueSize(void) const {
  int size;
  cMsgSubscribeGetMaxCueSize(config,&size);
  return(size);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgSubscriptionConfig::setMaxCueSize(int size) {
  cMsgSubscribeSetMaxCueSize(config,size);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgSubscriptionConfig::getSkipSize(void) const {
  int size;
  cMsgSubscribeGetSkipSize(config,&size);
  return(size);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgSubscriptionConfig::setSkipSize(int size) {
  cMsgSubscribeSetSkipSize(config,size);
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsgSubscriptionConfig::getMaySkip(void) const {
  int maySkip;
  cMsgSubscribeGetMaySkip(config,&maySkip);
  return((maySkip==0)?false:true);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgSubscriptionConfig::setMaySkip(bool maySkip) {
  cMsgSubscribeSetMaySkip(config,(maySkip)?1:0);
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsgSubscriptionConfig::getMustSerialize(void) const {
  int maySerialize;
  cMsgSubscribeGetMustSerialize(config,&maySerialize);
  return((maySerialize==0)?false:true);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgSubscriptionConfig::setMustSerialize(bool mustSerialize) {
  cMsgSubscribeSetMustSerialize(config,(mustSerialize)?1:0);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgSubscriptionConfig::getMaxThreads(void) const {
  int max;
  cMsgSubscribeGetMaxThreads(config,&max);
  return(max);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgSubscriptionConfig::setMaxThreads(int max) {
  cMsgSubscribeSetMaxThreads(config,max);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsgSubscriptionConfig::getMessagesPerThread(void) const {
  int mpt;
  cMsgSubscribeGetMessagesPerThread(config,&mpt);
  return(mpt);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgSubscriptionConfig::setMessagesPerThread(int mpt) {
  cMsgSubscribeSetMessagesPerThread(config,mpt);
}
 
 
//-----------------------------------------------------------------------------
 
 
size_t cMsgSubscriptionConfig::getStackSize(void) const {
  size_t size;
  cMsgSubscribeGetStackSize(config,&size);
  return(size);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsgSubscriptionConfig::setStackSize(size_t size) {
  cMsgSubscribeSetStackSize(config,size);
}
 
 
//-----------------------------------------------------------------------------
//  cMsg methods
//-----------------------------------------------------------------------------
 
 
cMsg::cMsg(const string &UDL, const string &name, const string &descr)
  : myUDL(UDL), myName(name), myDescr(descr), initialized(false) {
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsg::~cMsg(void) {
  cMsg::disconnect();
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::connect(void) throw(cMsgException) {
    int stat;
   
    // if we're already initialized (called connect once), we want to reconnect
    if(initialized) {
        if((stat=cMsgReconnect(myDomainId))!=CMSG_OK) {
            throw(cMsgException(cMsgPerror(stat),stat));
        }
        return;
    }
 
 
    if((stat=cMsgConnect(myUDL.c_str(),myName.c_str(),myDescr.c_str(),&myDomainId))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
    initialized=true;
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::disconnect(void) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
  cMsgDisconnect(&myDomainId);
 
  deleteSubscriptions(myDomainId);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::send(cMsgMessage &msg) throw(cMsgException) {
    
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat;
  if((stat=cMsgSend(myDomainId,msg.myMsgPointer))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::send(cMsgMessage *msg) throw(cMsgException) {
  cMsg::send(*msg);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsg::syncSend(cMsgMessage &msg, const struct timespec *timeout) throw(cMsgException) {
    
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int response;
 
  int stat;
  if((stat=cMsgSyncSend(myDomainId,msg.myMsgPointer,timeout,&response))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(response);
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsg::syncSend(cMsgMessage *msg, const struct timespec *timeout) throw(cMsgException) {
  return(cMsg::syncSend(*msg, timeout));
}
 
 
//-----------------------------------------------------------------------------
 
 
void *cMsg::subscribe(const string &subject, const string &type, cMsgCallback *cb, void *userArg,
                      const cMsgSubscriptionConfig *cfg) throw(cMsgException) {
    
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat;
  void *handle;
 
 
  // create and fill dispatcher struct
  dispatcherStruct *d = new dispatcherStruct();
  d->cb=cb;
  d->userArg=userArg;
 
 
  // subscribe and get handle
  stat=cMsgSubscribe(myDomainId,
                     (subject.size()<=0)?NULL:subject.c_str(),
                     (type.size()<=0)?NULL:type.c_str(),
                     callbackDispatcher,
                     (void*)d,
                     (cfg==NULL)?NULL:(cfg->config),
                     &handle);
 
 
  // check if subscription accepted
  if(stat!=CMSG_OK) {
    delete(d);
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
 
  // add this subscription to internal list
  addSubscription(myDomainId,subject,type,d,handle);
 
 
  return(handle);
}
 
 
//-----------------------------------------------------------------------------
 
 
void *cMsg::subscribe(const string &subject, const string &type, cMsgCallback &cb, void *userArg,
                      const cMsgSubscriptionConfig *cfg) throw(cMsgException) {
  return(cMsg::subscribe(subject, type, &cb, userArg, cfg));
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::unsubscribe(void *handle) throw(cMsgException) {
 
    if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
    int stat;
 
 
  // remove subscription from internal list
    if(!deleteSubscription(myDomainId,handle)) {
        throw(cMsgException(cMsgPerror(CMSG_BAD_ARGUMENT),CMSG_BAD_ARGUMENT));
    }
 
 
  // unsubscribe
    if((stat=cMsgUnSubscribe(myDomainId,handle))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
 
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::subscriptionPause(void *handle) throw(cMsgException) {
 
    if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
    int stat;
 
    // pause
    if((stat=cMsgSubscriptionPause(myDomainId,handle))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
 
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::subscriptionResume(void *handle) throw(cMsgException) {
 
    if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
    int stat;
 
    // resume
    if((stat=cMsgSubscriptionResume(myDomainId,handle))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
 
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::subscriptionQueueClear(void *handle) throw(cMsgException) {
 
    if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
    int stat;
 
    // clear Q
    if((stat=cMsgSubscriptionQueueClear(myDomainId,handle))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
 
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsg::subscriptionQueueCount(void *handle) throw(cMsgException) {
 
    if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
    int stat, count;
 
    // get Q count
    if((stat=cMsgSubscriptionQueueCount(myDomainId,handle,&count))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
    return count;
 
}
 
 
//-----------------------------------------------------------------------------
 
 
int cMsg::subscriptionMessagesTotal(void *handle) throw(cMsgException) {
 
    if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
    int stat, count;
 
    // total # of messages callback received
    if((stat=cMsgSubscriptionMessagesTotal(myDomainId,handle,&count))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
    return count;
 
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsg::subscriptionQueueIsFull(void *handle) throw(cMsgException) {
 
    if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
    int stat, val;
 
    // Q is full?
    if((stat=cMsgSubscriptionQueueIsFull(myDomainId,handle,&val))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
    return (val == 0 ? false : true);
 
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage *cMsg::sendAndGet(cMsgMessage &sendMsg, const struct timespec *timeout) throw(cMsgException) {
    
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  void *replyPtr;
 
  int stat;
  if((stat=cMsgSendAndGet(myDomainId,sendMsg.myMsgPointer,timeout,&replyPtr))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  return(new cMsgMessage(replyPtr));
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage *cMsg::sendAndGet(cMsgMessage *sendMsg, const struct timespec *timeout) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  void *replyPtr;
 
  int stat;
  if((stat=cMsgSendAndGet(myDomainId,sendMsg->myMsgPointer,timeout,&replyPtr))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  return(new cMsgMessage(replyPtr));
}
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage *cMsg::subscribeAndGet(const string &subject, const string &type, const struct timespec *timeout) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  void *replyPtr;
 
  int stat;
  if((stat=cMsgSubscribeAndGet(myDomainId,subject.c_str(),type.c_str(),timeout,&replyPtr))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  return(new cMsgMessage(replyPtr));
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::flush(const struct timespec *timeout) throw(cMsgException) {
    
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat;
  if((stat=cMsgFlush(myDomainId, timeout))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::start(void) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat;
  if((stat=cMsgReceiveStart(myDomainId))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::stop(void) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat;
  if((stat=cMsgReceiveStop(myDomainId))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsg::getDescription(void) const {
  return(myDescr);
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsg::getName(void) const {
  return(myName);
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsg::getUDL(void) const {
    return(myUDL);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::setUDL(const string &udl) throw(cMsgException) {
 
    int stat;
    if((stat=cMsgSetUDL(myDomainId,udl.c_str()))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    }
}
 
 
//-----------------------------------------------------------------------------
 
 
string cMsg::getCurrentUDL(void) const throw(cMsgException) {
    const char *s;
 
    int stat;
    if((stat=cMsgGetCurrentUDL(myDomainId,&s))!=CMSG_OK) {
        throw(cMsgException(cMsgPerror(stat),stat));
    };
 
    if(s==NULL) {
        return("null");
    } else {
        return(string(s));
    }
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsg::isConnected(void) const throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat,connected;
  if((stat=cMsgGetConnectState(myDomainId,&connected))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(connected==1);
}
 
 
//-----------------------------------------------------------------------------
 
 
bool cMsg::isReceiving(void) const throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat,receiving;
  if((stat=cMsgGetReceiveState(myDomainId,&receiving))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
  return(receiving==1);
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::setShutdownHandler(cMsgShutdownHandler *handler, void* userArg) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat;
  if((stat=cMsgSetShutdownHandler(myDomainId,handler,userArg))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::shutdownClients(const string &client, int flag) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat;
  if((stat=cMsgShutdownClients(myDomainId,client.c_str(),flag))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
void cMsg::shutdownServers(const string &server, int flag) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
 
  int stat;
  if((stat=cMsgShutdownServers(myDomainId,server.c_str(),flag))!=CMSG_OK) {
    throw(cMsgException(cMsgPerror(stat),stat));
  }
}
 
 
//-----------------------------------------------------------------------------
 
 
cMsgMessage *cMsg::monitor(const string &monString) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
  void *m = cMsgCreateMessage();
  if(m==NULL)throw(cMsgException("?cMsgMessage constructor...unable to create message",CMSG_ERROR));
 
  int stat;
  if((stat=cMsgMonitor(myDomainId,monString.c_str(),&m))!=CMSG_OK) {
    cMsgFreeMessage(&m);
    throw(cMsgException(cMsgPerror(stat),stat));
  }
 
  return(new cMsgMessage(m));
}
 
 
//-----------------------------------------------------------------------------
 
 
 void cMsg::setMonitoringString(const string &monString) throw(cMsgException) {
 
  if(!initialized)throw(cMsgException(cMsgPerror(CMSG_NOT_INITIALIZED),CMSG_NOT_INITIALIZED));
 
  int stat;
  if((stat=cMsgMonitor(myDomainId,monString.c_str(),NULL))!=CMSG_OK)
    throw(cMsgException(cMsgPerror(stat),stat));
 
  return;
}
 
 
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------