corbaconnection.cpp

java开源的企业总线.xmlBlaster

* Note: You don't need to free anything
* @param A vector with MessageUnit
* @return A vector of strings each is a publish return QoS. 
*/
vector<std::string> 
CorbaConnection::publishArr(const vector<util::MessageUnit> &msgVec) 
{
  if (log_.call()) log_.call(me(), "publishArr() ...");

  if (CORBA::is_nil(xmlBlaster_)) {
     string txt = "no auth.Server, you must login first";
     throw serverIdl::XmlBlasterException("communication.noConnection", 
                                          "client", me().c_str(), "en",
                                          txt.c_str(), "", "", "", "", "", "");
  }

  try {
     serverIdl::MessageUnitArr_var msgUnitArr = new serverIdl::MessageUnitArr;
     copyToCorba(msgUnitArr, msgVec);
     serverIdl::XmlTypeArr_var retArr = xmlBlaster_->publishArr(msgUnitArr);
     vector<std::string> vecArr;
     for (CORBA::ULong ii=0; ii<retArr->length(); ii++) {
        vecArr.push_back(corbaWStringToString(retArr[ii].inout()));
        //vecArr.push_back(static_cast<char *>(retArr[ii].inout()));
     }
     return vecArr;
  }
  catch(serverIdl::XmlBlasterException &e) {
     if (log_.trace()) log_.trace(me(), "XmlBlasterException: "
                                + string(e.message) );
     throw e;
  }
}

/**
* @deprecated Please use the STL vector variant
*/
serverIdl::XmlTypeArr* 
CorbaConnection::publishArr(const serverIdl::MessageUnitArr& msgUnitArr)
{
  if (log_.call()) log_.call(me(), "publishArr() ...");

  if (CORBA::is_nil(xmlBlaster_)) {
     string txt = "no auth.Server, you must login first";
     throw serverIdl::XmlBlasterException("communication.noConnection", 
                                          "client", me().c_str(), "en",
                                          txt.c_str(), "", "", "", "", "", "");
  }

  try {
     return xmlBlaster_->publishArr(msgUnitArr);
  }
  catch(serverIdl::XmlBlasterException &e) {
     if (log_.trace()) log_.trace(me(), "XmlBlasterException: "
                                + string(e.message) );
     throw e;
  }
  return 0;
}

/**
* Publish a bulk of messages without ACK. 
* <br />
* This method has a common interface which is not CORBA depending. 
* <br />
* Note: You don't need to free anything
* @param The MessageUnit array as a STL vector
*/
void 
CorbaConnection::publishOneway(const vector<util::MessageUnit>& msgVec)
{
  if (log_.call()) log_.call(me(), "publishOneway() ...");

  if (CORBA::is_nil(xmlBlaster_)) {
     string txt = "no auth.Server, you must login first";
     throw serverIdl::XmlBlasterException("communication.noConnection", 
                                          "client", me().c_str(), "en",
                                          txt.c_str(), "", "", "", "", "", "");
  }

  try {
     serverIdl::MessageUnitArr_var msgUnitArr = new serverIdl::MessageUnitArr;
     copyToCorba(msgUnitArr, msgVec);
     xmlBlaster_->publishOneway(msgUnitArr);
  }
  catch (const exception& e) {
     log_.error(me(), string("Exception caught in publishOneway, it is not transferred to client: ") + e.what());
  }
  catch(...) {
     log_.error(me(), "Exception caught in publishOneway, it is not transferred to client");
  }
}

/*
* Please use the STL based variant
* @param The MessageUnit array as a CORBA datatype
* @deprecated Use the vector<util::MessageUnit> variant
*/
void 
CorbaConnection::publishOneway(const serverIdl::MessageUnitArr& msgUnitArr)
{
  if (log_.call()) log_.call(me(), "publishOneway() ...");

  if (CORBA::is_nil(xmlBlaster_)) {
     string txt = "no auth.Server, you must login first";
     throw serverIdl::XmlBlasterException("communication.noConnection", 
                                          "client", me().c_str(), "en",
                                          txt.c_str(), "", "", "", "", "", "");
  }

  try {
     xmlBlaster_->publishOneway(msgUnitArr);
  }
  catch (const exception& e) {
     log_.error(me(), string("Exception caught in publishOneway, it is not transferred to client: ") + e.what());
  }
  catch(...) {
     log_.error(me(), "Exception caught in publishOneway, it is not transferred to client");
  }
}

/**
* This method has a common interface which is not CORBA depending. 
* <br />
* Note: You don't need to free anything
*/
vector<std::string> 
CorbaConnection::erase(const string &xmlKey, const string &qos) 
{
  if (log_.call()) log_.call(me(), "erase() ...");
  if (log_.dump()) {
     log_.dump(me(), string("erase: the key: ") + xmlKey);
     log_.dump(me(), string("erase: the qos: ") + qos);
  }

  if (CORBA::is_nil(xmlBlaster_)) {
     string txt = "no auth.Server, you must login first";
     throw serverIdl::XmlBlasterException("communication.noConnection", 
                                          "client", me().c_str(), "en",
                                          txt.c_str(), "", "", "", "", "", "");
  }

  try {
     serverIdl::XmlTypeArr_var retArr = xmlBlaster_->erase(toCorbaWString(xmlKey), toCorbaWString(qos));
     vector<std::string> vecArr;
     for (CORBA::ULong ii=0; ii<retArr->length(); ii++) {
        vecArr.push_back(corbaWStringToString(retArr[ii]));
        //vecArr.push_back(static_cast<const char *>(retArr[ii]));
     }
     return vecArr;
  }
  catch(serverIdl::XmlBlasterException e) {
     throw e;
  }
}


/**
* Access messages the synchronous way. 
* <br />
* Note: You don't need to free anything
* @return The STL MessageUnit vector, its a copy so if you have the variable on the
*         stack it will free itself
*/
vector<util::MessageUnit>
CorbaConnection::get(const string &xmlKey, const string &qos) 
{

  serverIdl::MessageUnitArr_var units;
  if (log_.call()) log_.call(me(), "get() ...");
  if (log_.dump()) {
     log_.dump(me(), string("get: the key: ") + xmlKey);
     log_.dump(me(), string("get: the qos: ") + qos);
  }

  if (CORBA::is_nil(xmlBlaster_)) {
     string txt = "no auth.Server, you must login first";
     throw serverIdl::XmlBlasterException("communication.noConnection", 
                                          "client", me().c_str(), "en",
                                          txt.c_str(), "", "", "", "", "", "");
  }

  try {
     units = xmlBlaster_->get(toCorbaWString(xmlKey), toCorbaWString(qos));
     /*
     string subId = xmlBlaster_->subscribe(xmlKey.c_str(),
                                           qos.c_str());
     log_.info(me(),"New Entry in Cache created (subId="+subId+")");
     */
     vector<util::MessageUnit> msgVec;
     copyFromCorba(msgVec, units);
     return msgVec;
  }
  catch(serverIdl::XmlBlasterException &e) {
     throw e;
  }
}

string 
CorbaConnection::ping(const string &qos) 
{
  if (log_.call()) log_.call(me(), "ping(" + qos + ") ...");

  if (CORBA::is_nil(xmlBlaster_)) {
     string txt = "no auth.Server, you must login first";
     throw serverIdl::XmlBlasterException("communication.noConnection", 
                                          "client", me().c_str(), "en",
                                          txt.c_str(), "", "", "", "", "", "");
  }

  try {
     CORBA::String_var ret = xmlBlaster_->ping("");
     return static_cast<char *>(ret);
  }
  catch(serverIdl::XmlBlasterException &e) {
     throw e;
  }
}

/**
* Transform a util::MessageUnit to the corba variant
*/
void 
CorbaConnection::copyToCorba(serverIdl::MessageUnit &dest,
                             const util::MessageUnit &src) const 
{
  dest.xmlKey = toCorbaWString(src.getKey().toXml());
  serverIdl::ContentType content(src.getContentLen(),
                                 src.getContentLen(),
                                 (CORBA::Octet*)src.getContent(),
                                 false); // our src does memory management itself
  dest.content = content;  // dest.content and content point to same memory? memory leak?
  dest.qos = toCorbaWString(src.getQos().toXml());
}

   
/**
* Transform STL vector to corba messageUnit array variant. 
*/
void 
CorbaConnection::copyToCorba(serverIdl::MessageUnitArr_var &units,
                             const vector<util::MessageUnit> &msgVec) const 
{
  unsigned int len = msgVec.size();
  units->length(len);
  for (CORBA::ULong ii=0; ii<len; ii++) {
     util::MessageUnit src = msgVec[ii];
     serverIdl::MessageUnit dest;
     copyToCorba(dest, src);
     units[ii] = dest;
  }
}

/**
* Transform corba messageUnit array to vector variant. 
* @param units Is not const as [] operator does not like it
*/
void 
CorbaConnection::copyFromCorba(vector<util::MessageUnit> &msgVec,
                               serverIdl::MessageUnitArr_var &units)
{
  unsigned int len = units->length();
  msgVec.reserve(len);
  for (CORBA::ULong ii=0; ii<len; ii++) {
     const serverIdl::MessageUnit &msgUnit = static_cast<const serverIdl::MessageUnit>(units[ii]);
     unsigned long len = static_cast<unsigned long>(msgUnit.content.length());
     const unsigned char * blob = static_cast<const unsigned char *>(&msgUnit.content[0]);
     if (log_.trace()) log_.trace(me(), "copyFromCorba() '" + string((const char *)blob) + "' len=" + lexical_cast<std::string>(len));
     MsgKeyData key = msgKeyFactory_.readObject(corbaWStringToString(msgUnit.xmlKey));
     MsgQosData qos = msgQosFactory_.readObject(corbaWStringToString(msgUnit.qos)); 
     const util::MessageUnit msg(key, len, blob, qos);
     msgVec.push_back(msg);
  }
}

/**
* Transform corba messageUnit array to vector variant. 
void CorbaConnection::copyFromCorba(util::MessageUnit &msgUnitUtil, serverIdl::MessageUnitArr_var &msgUnit) {
  string key(units[ii].xmlKey);
  unsigned long len = static_cast<unsigned long>(units[ii].content.length());
  const unsigned char * blob = static_cast<const unsigned char *>(&units[ii].content[0]);
  //unsigned char *blob = (unsigned char *)&units[ii].content[0];
  string qos(units[ii].qos);
  msgUnitUtil.setKey(key);
  msgUnitUtil.setContentLen(len);
  msgUnitUtil.setContent(blob);
  msgUnitUtil.setQos(qos);
}
*/

std::string CorbaConnection::usage() 
{
   std::string text = string("");
   //text += string("\n");
   text += string("\nThe CORBA plugin configuration:");
   text += string("\n   -bootstrapHostname <host>");
   text += string("\n                       The host where to find xmlBlaster [localhost]");
   text += string("\n   -bootstrapPort     <port>");
   text += string("\n                       The bootstrap port where xmlBlaster publishes its IOR [3412]");
   text += string("\n   -dispatch/connection/plugin/ior/iorString <IOR:00...>");
   text += string("\n                       The IOR string of the xmlBlaster-authentication server.");
   text += string("\n   -dispatch/connection/plugin/ior/iorFile <file>");
   text += string("\n                       A file with the xmlBlaster-authentication server IOR.");
   text += string("\n   -dispatch/connection/plugin/ior/useNameService <true/false>");
   text += string("\n                       Try to access xmlBlaster through a naming service [true]");
   text += string("\n");
   return text;
}

// CORBA::ORB_ptr CorbaConnection::orb_           = 0;
// unsigned short CorbaConnection::numOfSessions_ = 0;
// PortableServer::POA_ptr CorbaConnection::poa_  = 0;

}}}}} // end of namespace