giopbidir.cc

编译工具

  }
  g = new GIOP_C(this,s);
  g->impl(impl);
  g->initialise(ior,key,keysize,calldesc);
  g->giopStreamList::insert(s->clients);
  return g;
}

////////////////////////////////////////////////////////////////////////
void
BiDirServerRope::decrRefCount() {
  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,0);

  omni_tracedmutex_lock sync(*omniTransportLock);
  pd_refcount--;
  OMNIORB_ASSERT(pd_refcount >=0);

  if (pd_refcount) return;

  // Normally pd_strands always contain 1 strand unless the bidirectional
  // strand has been shutdown.
  if (RopeLink::is_empty(pd_strands) && !pd_nwaiting) {
    RopeLink::remove();
    delete this;
  }
  else {
    RopeLink* p = pd_strands.next;
    for (; p != &pd_strands; p = p->next) {
      giopStrand* g = (giopStrand*)p;
      if (g->state() != giopStrand::DYING) {
	if (omniORB::trace(30)) {
	  omniORB::logger l;
	  l << "Bi-directional rope is no longer referenced; strand "
	    << (void*)g << " is a candidate for scavenging.\n";
	}
	g->startIdleCounter();
      }
    }
  }
}


////////////////////////////////////////////////////////////////////////
void
BiDirServerRope::realIncrRefCount() {

  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,1);

  OMNIORB_ASSERT(pd_refcount >= 0);

  pd_refcount++;
}

////////////////////////////////////////////////////////////////////////
BiDirClientRope::BiDirClientRope(const giopAddressList& addrlist,
				 const omnivector<CORBA::ULong>& preferred) :
  giopRope(addrlist,preferred)
{
  pd_maxStrands = 1;
  pd_oneCallPerConnection = 0;
}

////////////////////////////////////////////////////////////////////////
IOP_C*
BiDirClientRope::acquireClient(const omniIOR* ior,
			       const CORBA::Octet* key,
			       CORBA::ULong keysize,
			       omniCallDescriptor* calldesc) {

  GIOP_C* giop_c = (GIOP_C*) giopRope::acquireClient(ior,key,keysize,calldesc);

  // Bidirectional is only supported in GIOP 1.2 and above
  GIOP::Version v = ior->getIORInfo()->version();
  if (!(v.major > 1 || v.minor >= 2)) {
    if (omniORB::trace(20)) {
      omniORB::logger log;
      log << "Bidirectional client using normal connection because "
	  << "it is only GIOP " << (int)v.major << "." << (int)v.minor << "\n";
    }
    return giop_c;
  }

  omni_tracedmutex_lock sync(pd_lock);
  giopStrand& s = (giopStrand&)((giopStream&)(*giop_c));
  if (s.connection == 0 && s.state() != giopStrand::DYING) {
    unsigned long deadline_secs,deadline_nanosecs;
    giop_c->getDeadline(deadline_secs,deadline_nanosecs);
    giopActiveConnection* c = s.address->Connect(deadline_secs,
						 deadline_nanosecs);
    if (c) s.connection = &(c->getConnection());
    if (!s.connection) {
      s.state(giopStrand::DYING);
    }
    else {
      // now make the connection manage by the giopServer.
      s.biDir = 1;
      s.gatekeeper_checked = 1;
      giopActiveCollection* watcher = c->registerMonitor();
      if (omniORB::trace(20)) {
	omniORB::logger log;
	log << "Client opened bidirectional connection to " 
	    << s.connection->peeraddress() << "\n";
      }
      if (!giopServer::singleton()->addBiDirStrand(&s,watcher)) {
	{
	  omni_tracedmutex_lock sync(*omniTransportLock);
	  s.connection->decrRefCount();
	}
	s.connection = 0;
	s.biDir = 0;
	giopRope::releaseClient(giop_c);
	OMNIORB_THROW(TRANSIENT,
		      TRANSIENT_BiDirConnUsedWithNoPOA,
		      CORBA::COMPLETED_NO);
      }
    }
  }
  return giop_c;
}

/////////////////////////////////////////////////////////////////////////////
//            Server side interceptor for code set service context         //
/////////////////////////////////////////////////////////////////////////////
static
CORBA::Boolean
getBiDirServiceContext(omniInterceptors::serverReceiveRequest_T::info_T& info)
{

  if (!orbParameters::acceptBiDirectionalGIOP) {
    // XXX If the ORB policy is "don't support bidirectional", don't bother 
    // doing any of the stuff below.
    return 1;
  }

  GIOP::Version ver = info.giop_s.version();

  giopStrand& strand = (giopStrand&)((giopStream&)info.giop_s);

  if (ver.minor != 2 || strand.isClient()) {
    // Only parse service context if the GIOP version is 1.2, on the
    // server side
    return 1;
  }

  IOP::ServiceContextList& svclist = info.giop_s.service_contexts();
  CORBA::ULong total = svclist.length();
  for (CORBA::ULong index = 0; index < total; index++) {
    if (svclist[index].context_id == IOP::BI_DIR_IIOP) {
      cdrEncapsulationStream e(svclist[index].context_data.get_buffer(),
			       svclist[index].context_data.length(),1);

      ListenPointList l;
      l <<= e;

      if (l.length() == 0) continue;

      if (omniORB::trace(25)) {
	omniORB::logger log;
	log << " receive bidir IIOP service context: ( ";

	for (CORBA::ULong i = 0; i < l.length(); i++) {
	  log << (const char*) l[i].host << ":" << l[i].port << " ";
	}

	log << ")\n";
      }

      giopAddressList addrList;

      for (CORBA::ULong i=0; i<l.length(); i++) {
	giopAddress* p = giopAddress::fromTcpAddress(l[i]);
	if (p) addrList.push_back(p);

	// XXX If this connection is SSL, we also add to the redirection
	// list a ssl version of the address. This will ensure that any 
	// callback objects from the other end will score a match.
	// This is necessary because the BI_DIR_IIOP service context does
	// not allow the other side to say an endpoint is SSL.
	// If omniORB 4 is the client, it sends both the TCP and the SSL
	// endpoint in the BI_DIR_IIOP listen point list. It may not 
	// have a TCP endpoint at all. In that case, the IORs from the
	// client will only contain a ssl endpoint. Our hack here will ensure
	// that the unmarshal object reference code will score a match.
	if (strncmp(strand.connection->myaddress(),"giop:ssl",8) == 0) {
	  p = giopAddress::fromSslAddress(l[i]);
	  if (p) addrList.push_back(p);
	}
      }

      if (addrList.empty()) continue;


      // Check serverTransportRule to see if we should allow bidir from
      // this client.
      {
	transportRules::sequenceString actions;
	CORBA::ULong matchedRule;
	CORBA::Boolean acceptbidir;
	CORBA::Boolean dumprule = 0;
	const char* why;

	if ( (acceptbidir = transportRules::serverRules().
	                    match(strand.connection->peeraddress(),
				  actions,matchedRule)) ) {
	  CORBA::ULong i;
	  for (i = 0; i < actions.length(); i++ ) {
	    if (strcmp(actions[i],"bidir") == 0) {
	      break;
	    }
	  }
	  if ( i == actions.length() ) {
	    acceptbidir = 0;
	    why = (const char*) "\"bidir\" is not found in the matched rule: ";
	    dumprule = 1;
	  }
	}
	else {
	  why = (const char*) "no matching rule is bound";
	}

	if ( !acceptbidir ) {
	  if ( omniORB::trace(1) ) {
	    omniORB::logger log;
	    log << "Request from " << strand.connection->peeraddress()
		<< " to switch to bidirectional is rejected because " 	
		<< why;
	    if (dumprule) {
	      CORBA::String_var rule;
	      rule = transportRules::serverRules().dumpRule(matchedRule);
	      log << "\"" << (const char*) rule << "\"";
	    }
	    log << "\n";
	  }
	  continue;
	}

	if ( omniORB::trace(5) ) {

	  CORBA::String_var rule;
	  rule = transportRules::serverRules().dumpRule(matchedRule);

	  omniORB::logger log;
	  log << "Accepted request from " 
	      << strand.connection->peeraddress()
	      << " to switch to bidirectional because of this rule: \""
	      << (const char*) rule << "\"\n";
	}
      }

      {
	ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,0);

	omni_tracedmutex_lock sync(*omniTransportLock);

	if (!strand.biDir) {
	  strand.biDir = 1;
	  strand.stopIdleCounter();
	  if (!strand.server->notifySwitchToBiDirectional(strand.connection))
	    return 1;
	}

	BiDirServerRope* r = BiDirServerRope::addRope(&strand,addrList);
      }
      
      giopAddressList::const_iterator addr, last;
      addr = addrList.begin();
      last = addrList.end();
      for (; addr != last; addr++) {
	delete (*addr);
      }
    }
  }
  return 1;
}

/////////////////////////////////////////////////////////////////////////////
//            Client side interceptor for code set service context         //
/////////////////////////////////////////////////////////////////////////////
static
CORBA::Boolean
setBiDirServiceContext(omniInterceptors::clientSendRequest_T::info_T& info) {

  if (!orbParameters::offerBiDirectionalGIOP) {
    // XXX If the ORB policy is "don't support bidirectional", don't bother 
    // doing any of the stuff below.
    return 1;
  }

  giopStrand& g = (giopStrand&)(info.giopstream);
  GIOP::Version ver = info.giopstream.version();

  if (ver.minor != 2 || !g.biDir || !g.isClient() || g.biDir_initiated) {
    // Only send service context if the GIOP version is 1.2, this is
    // a bidirectional connection, on the client side and it has not
    // been used yet.
    return 1;
  }

  const omnivector<const char*>& epts = omniObjAdapter::listMyEndpoints();
  omnivector<const char*>::const_iterator i = epts.begin();
  omnivector<const char*>::const_iterator last = epts.end();

  ListenPointList l(epts.size());
  CORBA::ULong j = 0;
  for ( ; i != last; i++) {
    if (strncmp((*i),"giop:tcp",8) == 0 || strncmp((*i),"giop:ssl",8) == 0) {
      const char* p = strchr((*i),':');
      OMNIORB_ASSERT(p);
      p = strchr(p+1,':');
      OMNIORB_ASSERT(p);
      l.length(j+1);
      if (!tcpTransportImpl::parseAddress(p+1,l[j])) OMNIORB_ASSERT(0);
    }
  }
  if (l.length()) {
    cdrEncapsulationStream e(CORBA::ULong(0),CORBA::Boolean(1));
    l >>= e;

    CORBA::Octet* data;
    CORBA::ULong max,datalen;
    e.getOctetStream(data,max,datalen);

    CORBA::ULong len = info.service_contexts.length() + 1;
    info.service_contexts.length(len);
    info.service_contexts[len-1].context_id = IOP::BI_DIR_IIOP;
    info.service_contexts[len-1].context_data.replace(max,datalen,data,1);

    if (omniORB::trace(25)) {
      omniORB::logger log;
      log << " send bidir IIOP service context: ( ";

      for (CORBA::ULong i = 0; i < l.length(); i++) {
	log << (const char*) l[i].host << ":" << l[i].port << " ";
      }

      log << ")\n";
    }
  }
  g.biDir_initiated = 1;
  return 1;
}

/////////////////////////////////////////////////////////////////////////////
//            Handlers for Configuration Options                           //
/////////////////////////////////////////////////////////////////////////////

/////////////////////////////////////////////////////////////////////////////
class acceptBiDirectionalGIOPHandler : public orbOptions::Handler {
public:

  acceptBiDirectionalGIOPHandler() : 
    orbOptions::Handler("acceptBiDirectionalGIOP",
			"acceptBiDirectionalGIOP = 0 or 1",
			1,
			"-ORBacceptBiDirectionalGIOP < 0 | 1 >") {}

  void visit(const char* value,orbOptions::Source) throw (orbOptions::BadParam) {

    CORBA::Boolean v;
    if (!orbOptions::getBoolean(value,v)) {
      throw orbOptions::BadParam(key(),value,
				 orbOptions::expect_boolean_msg);
    }
    orbParameters::acceptBiDirectionalGIOP = v;
  }

  void dump(orbOptions::sequenceString& result) {
    orbOptions::addKVBoolean(key(),orbParameters::acceptBiDirectionalGIOP,
			     result);
  }

};

static acceptBiDirectionalGIOPHandler acceptBiDirectionalGIOPHandler_;

/////////////////////////////////////////////////////////////////////////////
class offerBiDirectionalGIOPHandler : public orbOptions::Handler {
public:

  offerBiDirectionalGIOPHandler() : 
    orbOptions::Handler("offerBiDirectionalGIOP",
			"offerBiDirectionalGIOP = 0 or 1",
			1,
			"-ORBofferBiDirectionalGIOP < 0 | 1 >") {}

  void visit(const char* value,orbOptions::Source) throw (orbOptions::BadParam) {

    CORBA::Boolean v;
    if (!orbOptions::getBoolean(value,v)) {
      throw orbOptions::BadParam(key(),value,
				 orbOptions::expect_boolean_msg);
    }
    orbParameters::offerBiDirectionalGIOP = v;
  }

  void dump(orbOptions::sequenceString& result) {
    orbOptions::addKVBoolean(key(),orbParameters::offerBiDirectionalGIOP,
			     result);
  }

};

static offerBiDirectionalGIOPHandler offerBiDirectionalGIOPHandler_;

/////////////////////////////////////////////////////////////////////////////
//            Module initialiser                                           //
/////////////////////////////////////////////////////////////////////////////

class omni_giopbidir_initialiser : public omniInitialiser {
public:

  omni_giopbidir_initialiser() {
    orbOptions::singleton().registerHandler(offerBiDirectionalGIOPHandler_);
    orbOptions::singleton().registerHandler(acceptBiDirectionalGIOPHandler_);
  }

  void attach() {

    // install interceptors
    omniInterceptors* interceptors = omniORB::getInterceptors();
    interceptors->clientSendRequest.add(setBiDirServiceContext);
    interceptors->serverReceiveRequest.add(getBiDirServiceContext);
  }
  void detach() {
    // Get rid of any remaining ropes. By now they should all be strand-less.
    omni_tracedmutex_lock sync(*omniTransportLock);

    RopeLink* p = BiDirServerRope::ropes.next;
    giopRope* gr;
    int i=0;

    while (p != &BiDirServerRope::ropes) {
      gr = (giopRope*)p;
      OMNIORB_ASSERT(gr->pd_refcount == 0 &&
		     RopeLink::is_empty(gr->pd_strands) &&
		     !gr->pd_nwaiting);
      p = p->next;
      gr->RopeLink::remove();
      delete gr;
      ++i;
    }
    if (omniORB::trace(15)) {
      omniORB::logger l;
      l << i << " remaining bidir rope" << (i == 1 ? "" : "s")
	<< " deleted.\n";
    }
  }
};


static omni_giopbidir_initialiser initialiser;

omniInitialiser& omni_giopbidir_initialiser_ = initialiser;

OMNI_NAMESPACE_END(omni)