corbaconnection.cpp

java开源的企业总线.xmlBlaster

                     }
                     catch (const CORBA::Exception &exc) {
                        log_.warn(me(), "Connecting to NameService entry '"+tmpServerName+"' failed: " + to_string(exc));
                     }
                  }

                  if (CORBA::is_nil(authServerFirst.in())) {
                     if (log_.trace()) log_.trace(me(), "Remember the first server");
                     try {
                        firstServerName = tmpServerName;
                        if (log_.trace()) log_.trace(me(), "Remember the first reachable xmlBlaster server from NameService entry '"+firstServerName+"'");
                        authServerFirst = authenticateIdl::AuthServer::_narrow(relativeContext.resolve(id, kind));
                     }
                     catch (const CORBA::Exception &exc) {
                        log_.warn(me(), "Connecting to NameService entry '"+tmpServerName+"' failed: " + to_string(exc));
                     }
                  }
               } while ( more );
            }
            bi->destroy();  // Clean up server side iteration resources
         }

         if (CORBA::is_nil(authServer_)) {
            if (!CORBA::is_nil(authServerFirst.in())) {
               if (countServerFound > 1) {
                  string str = string("Can't choose one of ") + lexical_cast<std::string>(countServerFound) +
                                 " avalailable server in CORBA NameService: " + serverNameList +
                                 ". Please choose one with e.g. -NameService.node.id " + tmpId;
                  log_.warn(me(), str);
                  throw XmlBlasterException("communication.noConnection", "client", me(), "en", str);
               }
               log_.info(me(), "Choosing only available server '" + firstServerName + "' in CORBA NameService");
               this->authServer_ = authenticateIdl::AuthServer::_duplicate(authServerFirst.in());
               return;
            }
            else {
               log_.trace(me(), "No usable xmlBlaster server found in NameService: " + serverNameList);
               throw XmlBlasterException("communication.noConnection", "client", me(), "en", text);
            }
         }

         log_.info(me(), "Accessing xmlBlaster using CORBA naming service entry '" +
                           NameServerControl::getString(contextId, contextKind) +
                           "/" + NameServerControl::getString(clusterId, clusterKind));

         return;
      }
      catch(serverIdl::XmlBlasterException &e ) {
         log_.trace(me() + ".NoAuthService", text);
         throw CorbaDriver::convertFromCorbaException(e);
      }
   } // if (useNameService)

   if (log_.trace()) log_.trace(me(), "No -dispatch/connection/plugin/ior/useNameService ...");
   throw XmlBlasterException("communication.noConnection", "client", me(), "en", text);
} // initAuthenticationService() 

void CorbaConnection::createCallbackServer(POA_clientIdl::BlasterCallback *implObj) 
{
  if (implObj) {
     if (log_.trace()) log_.trace(me(), "Trying resolve_initial_references ...");
     CORBA::Object_var obj = orb_->resolve_initial_references("RootPOA");
     if (log_.trace()) log_.trace(me(), "Trying narrowing POA ...");
     poa_ = PortableServer::POA::_narrow(obj.in());
     PortableServer::POAManager_var poa_mgr = poa_->the_POAManager();
     // _this() incarnates with the servant ...
     callback_ = implObj->_this();
     if (log_.trace()) log_.trace(me(), "Trying object_to_string POA ...");
     CORBA::String_var tmp = orb_->object_to_string(callback_);
     callbackIOR_ = tmp;
     if (log_.trace()) log_.trace(me(), "Trying activate POA ...");
     poa_mgr->activate();
#if  defined(XMLBLASTER_MICO) && defined(ORB_IS_THREAD_SAFE)
     // - multi threaded mico 2.3.11 sometimes blocked forever in work_pending()
     // - omniORB doesn't need perform_work() either but it doesn't harm
#else
     // - TAO seems to need it (callback messages won't arrive without)
     if (log_.trace()) log_.trace(me(), "Trying orb.work_pending ...");
     while (orb_->work_pending()) {
        if (log_.trace()) log_.trace(me(), "Entering perform_work ...");
        orb_->perform_work();
     }
     if (log_.trace()) log_.trace(me(), "Trying work_pending POA done ...");
#endif
     return;
     // add exception handling here !!!!!
  }
  return;
}

ConnectReturnQosRef CorbaConnection::connect(const ConnectQosRef& connectQos)
{
   if ( !CORBA::is_nil(xmlBlaster_)) {
      string msg = "You are already logged in, returning cached handle";
      msg += " on xmlBlaster";
      log_.warn(me(), msg);
      return connectReturnQos_;
   }

   loginName_ = connectQos->getUserId();
   if (log_.call()) log_.call(me(),"connect(" + loginName_ + ") ...");
   try {
      if (CORBA::is_nil(authServer_)) initAuthenticationService();
      ConnectQos help = *connectQos; // since it is a const
      string reqQos = help.toXml();
      if (log_.trace()) log_.trace(me(), string("connect req: ") + reqQos);
      // If using wstring in xmlBlaster.idl:
      //CORBA::WString_var ws1 = CORBA::wstring_dup(toWstring(reqQos).c_str());
      //CORBA::WString_var ws2 = authServer_->connect(ws1);
      //string retQos = toString(wstring(ws2));
      // or
      string retQos = corbaWStringToString(authServer_->connect(toCorbaWString(reqQos)));
      //string retQos = authServer_->connect(reqQos.c_str());
      if (log_.trace()) log_.trace(me(), string("connect ret: ") + retQos);
      ConnectQosFactory factory(global_);
      if (log_.dump()) log_.dump(me(), "connect: the connect return qos before parsing: " + retQos);
      connectReturnQos_ = factory.readObject(retQos);
      sessionId_ = connectReturnQos_->getSecretSessionId();
      xmlBlasterIOR_ = connectReturnQos_->getServerRef().getAddress();

      CORBA::Object_var obj = orb_->string_to_object(xmlBlasterIOR_.c_str());
      xmlBlaster_ = serverIdl::Server::_narrow(obj.in());

      numLogins_++;
      if (log_.trace()) log_.trace(me(),"Success, connect for "+loginName_);
      return connectReturnQos_;
   }
   catch(const XmlBlasterException &e) {
      string msg = "Connect failed for ";
      msg +=  loginName_; //  + ", numLogins=" + numLogins_;
      if (log_.trace()) log_.trace(me(), msg);
      throw e;
   }
}

bool CorbaConnection::shutdown()
{
   bool ret = false;
   if (!CORBA::is_nil(xmlBlaster_)) {
      CORBA::release(xmlBlaster_);
      xmlBlaster_ = NULL;
      ret = true;
   }
   if (!CORBA::is_nil(authServer_)) {
      CORBA::release(authServer_);
      authServer_ = NULL;
      ret = true;
   }
   return ret;
}

bool CorbaConnection::shutdownCb()
{
   if (!CORBA::is_nil(callback_)) {
      CORBA::release(callback_);
      callback_ = NULL;
      return true;
   }
   return false;
}

bool CorbaConnection::disconnect(const string& qos)
{
   if (log_.call()) log_.call(me(), "disconnect() ...");
   if (log_.dump()) log_.dump(me(), string("disconnect: the qos: ") + qos);

   try {
      if (!CORBA::is_nil(authServer_)) {
         if (sessionId_=="") authServer_->logout(xmlBlaster_);
         else authServer_->disconnect(sessionId_.c_str(), toCorbaWString(qos));
      }
      shutdown();
      return true;
   }
   catch (...) {
   }
   shutdown();
   return false;
}

/**
* Subscribe a message. 
* <br />
* Note: You don't need to free anything
* @return The xml based QoS
*/
string 
CorbaConnection::subscribe(const string &xmlKey, const string &qos) 
{
   if (log_.call()) log_.call(me(), "subscribe() ...");
   if (log_.dump()) {
      log_.dump(me(), string("subscribe: the key: ") + xmlKey);
      log_.dump(me(), string("subscribe: 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 {
     return corbaWStringToString(xmlBlaster_->subscribe(toCorbaWString(xmlKey), toCorbaWString(qos)));
     //CORBA::String_var ret = toString(xmlBlaster_->subscribe(xmlKey.c_str(), qos.c_str()));
     //return static_cast<const char *>(ret);
  } catch(serverIdl::XmlBlasterException &e) {
     throw e;
  }
  //return "";
}


vector<std::string> CorbaConnection::unSubscribe(const string &xmlKey,
                                  const string &qos) 
{
  if (log_.call()) log_.call(me(), "unSubscribe() ...");
  if (log_.dump()) {
     log_.dump(me(), string("unSubscribe: the key: ") + xmlKey);
     log_.dump(me(), string("unSubscribe: 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_->unSubscribe(toCorbaWString(xmlKey), toCorbaWString(qos));
     
     vector<std::string> vecArr;
     for (CORBA::ULong ii=0; ii<retArr->length(); ii++) {
        vecArr.push_back(corbaWStringToString(retArr[ii].inout()));
     }
     return vecArr;
  }
  catch(serverIdl::XmlBlasterException e) {
     throw e;
  }
}

/**
* publish a message. 
* <br />
* This method has a common interface which is not CORBA depending. 
* <br />
* Note: You don't need to free anything
* @return The xml based QoS
*/
string CorbaConnection::publish(const util::MessageUnit &msgUnitUtil) {
  if (log_.trace()) log_.trace(me(), "Publishing the STL way ...");
  if (log_.dump()) {
     log_.dump(me(), string("publish: the msgUnit: ") + msgUnitUtil.toXml());
  }

  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::MessageUnit msgUnit;
     // serverIdl::MessageUnit_var msgUnit;
     copyToCorba(msgUnit, msgUnitUtil);
     return corbaWStringToString(xmlBlaster_->publish(msgUnit));
     //CORBA::String_var ret = xmlBlaster_->publish(msgUnit);
     //return static_cast<char *>(ret);
  }
  catch(serverIdl::XmlBlasterException &e) {
     string msg = "XmlBlasterException: ";
     msg += e.message;
     if (log_.trace()) log_.trace(me(), msg);
     throw e;
  }
//        catch(CORBA::Exception &ex1) {
//       throw serverIdl::XmlBlasterException(me().c_str(),to_string(ex1));
//        }
}

   /**
    * @deprecated Please use the util::MessageUnit variant
    */
string 
CorbaConnection::publish(const serverIdl::MessageUnit &msgUnit) 
{
  if (log_.trace()) log_.trace(me(), "Publishing ...");

  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 corbaWStringToString(xmlBlaster_->publish(msgUnit));
     //CORBA::String_var ret = xmlBlaster_->publish(msgUnit);
     //return static_cast<char *>(ret);
  }
  catch(serverIdl::XmlBlasterException &e) {
     string msg = "XmlBlasterException: ";
     msg += e.message;
     if (log_.trace()) log_.trace(me(), msg);
     throw e;
  }
//        catch(CORBA::Exception &ex1) {
//       throw serverIdl::XmlBlasterException(me().c_str(),to_string(ex1));
//        }
}

/**
* Publish a bulk of messages. 
* <br />
* This method has a common interface which is not CORBA depending. 
* <br />