corbaconnection.cpp

java开源的企业总线.xmlBlaster

/*----------------------------------------------------------------------------
Name:      CorbaConnection.cpp
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Helper to connect to xmlBlaster: for now a simplified version
           without caching and without failsafe mode.
Author:    <Michele Laghi> [e-mail]
-----------------------------------------------------------------------------*/
/*
#ifdef _WINDOWS
#pragma warning(disable:4786)
#endif
*/
#include <client/protocol/corba/CorbaConnection.h>
#include <util/Constants.h>
#include <sys/types.h>
#ifdef _WINDOWS
#  include <winsock.h>
#else
#  if defined(__APPLE__) || defined(__FreeBSD__) || defined(__NetBSD__) || defined(__hpux__)
#    include <netinet/in.h>
#    include <sys/types.h>   /* Needed for __FreeBSD__ */
#  endif
#  include <sys/socket.h>
#  include <netdb.h>
#  include <arpa/inet.h>   // inet_addr()
#  include <unistd.h>      // gethostname()
#endif

#include <util/XmlBlasterException.h>
#include <util/Global.h>
#include <client/protocol/corba/CorbaDriver.h>

void closeSocket(int fd) {
#ifdef _WINDOWS
   closesocket(fd);
#else
   (void)close(fd);
#endif
}

namespace org {
 namespace xmlBlaster {
  namespace client {
   namespace protocol {
    namespace corba {

using namespace std;
using namespace org::xmlBlaster::util;
using namespace org::xmlBlaster::util::qos;
using namespace org::xmlBlaster::util::key;

CorbaConnection::CorbaConnection(Global& global, CORBA::ORB_ptr orb)
  : orb_(0),
    poa_(0),
    /* loginQos_(), */
    connectReturnQos_((ConnectReturnQos*)0), 
    global_(global), 
    log_(global.getLog("org.xmlBlaster.client.protocol.corba")),
    msgKeyFactory_(global),
    msgQosFactory_(global)
{
  //global_.getProperty().loadPropertyFile();
  log_.info(me(), "Initializing CORBA ORB");
  if (log_.call()) log_.call(me(), "CorbaConnection constructor ...");
//  if (numOfSessions_ == 0) {
     if (orb) orb_ = orb;
     else {
        int args                 = global_.getArgs();
        const char * const* argc = global_.getArgc();
        orb_ = CORBA::ORB_init(args, const_cast<char **>(argc)); //, "XmlBlaster-C++-Client");
     }

//  numOfSessions_++;
  nameServerControl_ = 0;
  numLogins_         = 0;
  xmlBlaster_        = 0;
  authServer_        = 0; // initAuthenticationService();
  callback_          = 0;
  defaultCallback_   = 0;
  sessionId_         = "";
  xmlBlasterIOR_     = "";
}


CorbaConnection::~CorbaConnection() 
{
   if (log_.call()) log_.call(me(), "destructor");

   delete nameServerControl_;
   
   if (log_.trace()) log_.trace(me(), "destructor: invoking shutdown");
   shutdown();
   
   if (log_.trace()) log_.trace(me(), "destructor: invoking shutdownCb");
   shutdownCb();
   if (log_.trace()) log_.trace(me(), "destructor: deleting the defaultCallback");
   delete defaultCallback_;

   if (log_.trace()) log_.trace(me(), "destructor: releasing the orb");
   if (!CORBA::is_nil(orb_)) CORBA::release(orb_);
   if (log_.trace()) log_.trace(me(), "destructor: releasing the poa");
   if (!CORBA::is_nil(poa_)) CORBA::release(poa_);
   orb_ = 0;
   poa_ = 0;
}

string CorbaConnection::getAddress() const
{
   return xmlBlasterIOR_;
}

string CorbaConnection::getCbAddress() const
{
   return callbackIOR_;
}

void CorbaConnection::initNamingService() 
{
  if (log_.call()) log_.call(me(), "initNamingService() ...");
  if (orb_ == 0) log_.panic(me(), "orb==null, internal problem");
  if (nameServerControl_ == 0)
     nameServerControl_ = new NameServerControl(orb_);
}

   
void CorbaConnection::initAuthenticationService() 
{
  if (log_.call()) log_.call(me(), "initAuthenticationService() ...");
  if (!CORBA::is_nil(authServer_))
     return;

  // 1) check if argument -IOR at program startup is given
  string authServerIOR = /* -dispatch/connection/plugin/ior/iorString IOR string is directly given */
     global_.getProperty().getStringProperty("dispatch/callback/plugin/ior/iorString","");
  if (authServerIOR != "") {
     CORBA::Object_var
        obj = orb_->string_to_object(authServerIOR.c_str());
     authServer_ = authenticateIdl::AuthServer::_narrow(obj.in());
     log_.info(me(),"Accessing xmlBlaster using your given IOR string");
     return;
  }
  if (log_.trace()) log_.trace(me(), "No -dispatch/connection/plugin/ior/iorString ...");

  string authServerIORFile =
     global_.getProperty().getStringProperty("dispatch/connection/plugin/ior/iorFile","");
  // -dispatch/connection/plugin/ior/iorFile IOR string is given through a file
  if (authServerIORFile != "") {
     ifstream in(authServerIORFile.c_str());
     if ((!in) /* && (log_.PANIC) */ )
        log_.panic(me(), "Could not open the file");
     in >> authServerIOR;
     in.close();
     CORBA::Object_var
        obj = orb_->string_to_object(authServerIOR.c_str());
     authServer_ = authenticateIdl::AuthServer::_narrow(obj.in());
     string msg  = "Accessing xmlBlaster using your given IOR file ";
     msg += authServerIORFile;
     log_.info(me(), msg);
     return;
  }
  if (log_.trace()) log_.trace(me(), "No -dispatch/connection/plugin/ior/iorFile ...");

  // 3) Using builtin http IOR download ...
  {
     char myHostName[126];
     strcpy(myHostName, "localhost");
     gethostname(myHostName, 125);
     string iorHost = global_.getProperty().getStringProperty("bootstrapHostname",myHostName);
     // Port may be a name from /etc/services: "xmlBlaster 3412/tcp"
     string iorPortStr = global_.getProperty().getStringProperty("bootstrapPort","3412"); // default bootstrapPort=3412 (xmlblaster)
     if (log_.trace()) log_.trace(me(), "Trying -bootstrapHostname=" + iorHost + " and -bootstrapPort=" + iorPortStr + " ...");
     struct sockaddr_in xmlBlasterAddr;
     memset((char *)&xmlBlasterAddr, 0, sizeof(xmlBlasterAddr));
     xmlBlasterAddr.sin_family=AF_INET;
     struct hostent *hostP = gethostbyname(iorHost.c_str());
     struct servent *portP = getservbyname(iorPortStr.c_str(), "tcp");
     string authServerIOR;
     authServerIOR.reserve(520);
     if (hostP != NULL) {
        xmlBlasterAddr.sin_addr.s_addr = ((struct in_addr *)(hostP->h_addr))->s_addr; //inet_addr("192.168.1.2");
        if (portP != NULL)
           xmlBlasterAddr.sin_port = portP->s_port;
        else
           xmlBlasterAddr.sin_port = htons(global_.getProperty().getIntProperty("bootstrapPort",3412));
        int s = socket(AF_INET, SOCK_STREAM, 0);
        if (s != -1) {
           if (::connect(s, (struct sockaddr *)&xmlBlasterAddr, sizeof(xmlBlasterAddr)) != -1) {
              string req="GET /AuthenticationService.ior HTTP/1.0\r\n \n";
              int numSent = send(s, req.c_str(), req.size(), 0);
              if (numSent < (int)req.size()) {
                 log_.error(me(), "Problems sending request '" + req + "'");
              }
              else {
                 log_.trace(me(), "Sent IOR request '" + req + "'");
              }
              int numRead;
              char buf[10];
              while ((numRead = recv(s, buf, 10, 0)) > 0) {
                authServerIOR.append(buf, numRead); 
              }
              if (log_.dump()) log_.dump(me(), "Received IOR data: '" + authServerIOR + "'");
              size_t pos = authServerIOR.find("IOR:");
//              if (pos > 0)
              if (pos != authServerIOR.npos) authServerIOR = authServerIOR.substr(pos);
              else {
                 throw serverIdl::XmlBlasterException("communication.noConnection", 
                                                      "client", me().c_str(), "en",
                                                      "can't access authentication Service", "", "", "", "", 
                                                      "", "");
              }
              if (log_.trace()) log_.trace(me(), "Received IOR data: '" + authServerIOR + "'");
           }
           else {
              log_.warn(me(), "Connecting to -bootstrapHostname=" + iorHost + " failed"); // errno
           }
           ::shutdown(s, 2); // SHUT_RDWR
           ::closeSocket(s); // Added because of handle leak reported by James Cazier
        }
     }
     if (!authServerIOR.empty()) {
        CORBA::Object_var obj = orb_->string_to_object(authServerIOR.c_str());
        if (!CORBA::is_nil(obj.in())) {
           if (!CORBA::is_nil(authServer_)) {
              CORBA::release(authServer_);
              authServer_ = 0;
           }
           authServer_ = authenticateIdl::AuthServer::_narrow(obj.in());
           string msg  = "Accessing xmlBlaster using -bootstrapHostname "+iorHost;
           log_.info(me(), msg);
           return;
        }
     }
  }
  if (log_.trace()) log_.trace(me(), "No -bootstrapHostname and -bootstrapPort ...");


  // 4) asking Name Service CORBA compliant
  bool useNameService=global_.getProperty().getBoolProperty("dispatch/connection/plugin/ior/useNameService",true);
  // -dispatch/connection/plugin/ior/useNameService default is to ask the naming service

  string text = "Can't access xmlBlaster Authentication Service";
  text += ", is the server running and ready?\n - try to specify ";
  text += "'-dispatch/connection/plugin/ior/iorFile <fileName>' if server is running on same host\n";
  text += " - try to specify '-bootstrapHostname <hostName>  -bootstrapPort 3412' to ";
  text += "locate xmlBlaster\n  - or contact your ";
  text += "system administrator to start a naming service";

  if (useNameService) {
     try {
        if (!nameServerControl_) initNamingService();

        string contextId = global_.getProperty().getStringProperty("NameService.context.id", "xmlBlaster");
        string contextKind = global_.getProperty().getStringProperty("NameService.context.kind", "MOM");
        string clusterId = global_.getProperty().getStringProperty("NameService.node.id", global_.getStrippedId());
        string clusterKind = global_.getProperty().getStringProperty("NameService.node.kind", "MOM");

        CORBA::Object_var obj = nameServerControl_->resolve(contextId, contextKind);
        CosNaming::NamingContext_var relativeContext_obj = CosNaming::NamingContext::_narrow(obj.in());
        NameServerControl relativeContext(relativeContext_obj);
        log_.info(me(), "Retrieved NameService context " + contextId + "." + contextKind);

         authenticateIdl::AuthServer_var authServerFirst;
         string tmpId = "";           // for logging only
         string tmpServerName = "";   // for logging only
         string firstServerName = ""; // for logging only
         int countServerFound = 0;    // for logging only
         string serverNameList = "";  // for logging only
         try {
            authServer_ = authenticateIdl::AuthServer::_narrow(relativeContext.resolve(clusterId, clusterKind));
         }
         catch (XmlBlasterException ex) {
            log_.info(me(), "Narrow AuthServer failed: " + ex.toString());
         }

         /*============================
          TestGet -ORBInitRef NameService=`cat /tmp/ns.ior` -trace true -call true
         =============================*/

         if ( CORBA::is_nil(authServer_) ) {
            if (log_.trace()) log_.trace(me(), "Query NameServer to find a suitable xmlBlaster server for '" +
                     NameServerControl::getString(contextId, contextKind)+"/"+NameServerControl::getString(clusterId, clusterKind) +
                     "' failed, is nil");
            CosNaming::BindingList_var bl;
            CosNaming::BindingIterator_var bi;
            CosNaming::NamingContext_var tmp = relativeContext.getNamingService();
            tmp->list(0, bl, bi);

            // process the remaining bindings if an iterator exists:
            if (CORBA::is_nil(authServer_) && !CORBA::is_nil(bi.in())) {
               int i = 0;
               CORBA::Boolean more;
               do {
                  more = bi->next_n(1, bl);
                  if (bl->length() != 1) {
                     if (log_.trace()) log_.trace(me(), "NameService entry id is nil");
                     break;
                  }
                  CORBA::ULong index = 0;
                  string id = lexical_cast<std::string>(bl[index].binding_name[0].id);
                  string kind = lexical_cast<std::string>(bl[index].binding_name[0].kind);
                  if (log_.trace()) log_.trace(me(), "id=" + id + " kind=" + kind);

                  tmpId = id;
                  countServerFound++;
                  tmpServerName = NameServerControl::getString(contextId, contextKind)+"/"+NameServerControl::getString(id, kind);
                  if (i>0) serverNameList += ", ";
                  i++;
                  serverNameList += tmpServerName;

                  if (clusterId == id && clusterKind == kind) {
                     try {
                        if (log_.trace()) log_.trace(me(), "Trying to resolve NameService entry '"+NameServerControl::getString(id, kind)+"'");
                        authServer_ = authenticateIdl::AuthServer::_narrow(relativeContext.resolve(id, kind));
                        if (! CORBA::is_nil(authServer_))
                           break; // found a matching server
                        else
                           log_.warn(me(), "Connecting to NameService entry '"+tmpServerName+"' failed, is_nil");