giop_s.cc

来自「编译工具」· CC 代码 · 共 827 行 · 第 1/2 页

    MARSHAL_SYSTEM_EXCEPTION(); \
  }

  OMNIORB_FOR_EACH_SYS_EXCEPTION(CATCH_AND_MARSHAL)

#    undef CATCH_AND_MARSHAL

  catch(omniORB::StubUserException& uex) {
    if (response_expected()) {
      CORBA::UserException& ex = *((CORBA::UserException*)uex.ex());
      MARSHAL_USER_EXCEPTION();
      delete uex.ex();  // ?? Possible memory leak?
    }
  }

#endif


  catch(CORBA::SystemException& ex) {
    MARSHAL_SYSTEM_EXCEPTION();
    // If the client does not expect a response, we quietly drop
    // the system exception.
  }
  catch(CORBA::UserException& ex) {
    if (response_expected()) {
      MARSHAL_USER_EXCEPTION();
    }
  }
#undef MARSHAL_USER_EXCEPTION
#undef MARSHAL_SYSTEM_EXCEPTION

  catch(const giopStream::CommFailure&) {
    throw;
  }

  catch(...) {
    if( omniORB::traceLevel > 1 ) {
      omniORB::logger l;
      l << "WARNING -- method \'" << operation() << "\' raised an unexpected\n"
	" exception (not a CORBA exception).\n";
    }
    CORBA::UNKNOWN ex(UNKNOWN_UserException,
		      (CORBA::CompletionStatus) completion());
    impl()->sendSystemException(this,ex);
  }
  pd_state = ReplyCompleted;
  return 1;
}

////////////////////////////////////////////////////////////////////////
CORBA::Boolean
GIOP_S::handleLocateRequest() {
  try {

    impl()->unmarshalLocateRequest(this);

    pd_state = RequestIsBeingProcessed;

    // Here we notify the giopServer that this thread has finished
    // reading the request. The server may want to keep a watch on
    // any more request coming in on the connection while this
    // thread does the upcall.

    CORBA::Boolean data_in_buffer = 0;
    if (pd_rdlocked) {
      // This is the thread that is reading from the connection. We
      // check if we have previously queued giopStream_Buffers on the
      // connection. In other words, we might have previously read
      // too much stuff out of the connection and these data belong to
      // other requests. If that is the case, we notify the giopServer
      // that there are already buffers waiting to be read.
      giopStrand& s = (giopStrand&) *this;
      data_in_buffer = ((s.head) ? 1 : 0);
    }
    pd_worker->server()->notifyWkPreUpCall(pd_worker,data_in_buffer);

    impl()->inputMessageEnd(this,0);
    
    pd_state = WaitingForReply;

    omniORB::logs(10, "Handling a GIOP LOCATE_REQUEST.");

    GIOP::LocateStatusType status = GIOP::UNKNOWN_OBJECT;

    if (keysize() > 0) {
      CORBA::ULong hash = omni::hash(key(), keysize());
      omni_tracedmutex_lock sync(*omni::internalLock);
      omniLocalIdentity* id;
      id = omniObjTable::locateActive(key(), keysize(), hash, 1);
      if( id )  status = GIOP::OBJECT_HERE;
     }
    if ( status == GIOP::UNKNOWN_OBJECT && keysize() > 0 ) {
      // We attempt to find the object adapter (activate it if necassary)
      // and ask it if the object exists, or if it has the *capability*
      // to activate such an object.  ie. is it able to do object loading
      // on demand?

      omniObjAdapter_var adapter(omniObjAdapter::getAdapter(key(),keysize()));
      if( adapter && adapter->objectExists(key(),keysize()) )
	status = GIOP::OBJECT_HERE;
    }
    if ( status == GIOP::UNKNOWN_OBJECT && 
	 keysize() == 4 && !memcmp(key(), "INIT", 4) &&
	 omniInitialReferences::is_bootstrap_agentImpl_initialised() ) {
      status = GIOP::OBJECT_HERE;
    }

    impl()->sendLocateReply(this,status,CORBA::Object::_nil(),0);
  }
  catch (omniORB::LOCATION_FORWARD& lf) {
    CORBA::Object_var release_it(lf.get_obj());
    impl()->sendLocateReply(this,
			    lf.is_permanent() ? GIOP::OBJECT_FORWARD_PERM :
			                        GIOP::OBJECT_FORWARD,
                            release_it,0);
  }

#define MARSHAL_SYSTEM_EXCEPTION() do { \
    if (pd_state == WaitForRequestHeader || \
        pd_state == RequestHeaderIsBeingProcessed) { \
      impl()->sendMsgErrorMessage(this); \
      return 0; \
    } else if (response_expected()) { \
      impl()->sendSystemException(this,ex); \
    } \
    if (pd_state == RequestIsBeingProcessed) { \
      SkipRequestBody(); \
    } \
} while (0) 

# ifndef HAS_Cplusplus_catch_exception_by_base

  // We have to catch each type of system exception separately
  // here to support compilers which cannot catch more derived
  // types.
#   define CATCH_AND_MARSHAL(name)  \
  catch (CORBA::name& ex) {  \
    MARSHAL_SYSTEM_EXCEPTION(); \
  }

  OMNIORB_FOR_EACH_SYS_EXCEPTION(CATCH_AND_MARSHAL)

#    undef CATCH_AND_MARSHAL
#endif

  catch(CORBA::SystemException& ex) {
    MARSHAL_SYSTEM_EXCEPTION();
  }

  pd_state = ReplyCompleted;
  return 1;
}

////////////////////////////////////////////////////////////////////////
CORBA::Boolean
GIOP_S::handleCancelRequest() {
  // We do not have the means to asynchronously abort the execution of
  // an upcall by another thread. Therefore it is not possible to
  // cancel a request that has already been in progress. The best we
  // can do is prevent the reply from happening.
  omniORB::logs(5, "Received a CancelRequest message.");
  pd_state = WaitingForReply;
  response_expected(0);
  return 1;
}

////////////////////////////////////////////////////////////////////////
void
GIOP_S::ReceiveRequest(omniCallDescriptor& desc) {

  OMNIORB_ASSERT(pd_state == RequestIsBeingProcessed);

  calldescriptor(&desc);

  // When a user exception is throw by the stub code, the
  // call descriptor could have been deallocated before the
  // catch frame for the user exception is reached. Therefore
  // we store the user exception signatures inside our own
  // private states.
  pd_n_user_excns = desc.n_user_excns();
  pd_user_excns = desc.user_excns();

  cdrStream& s = *this;
  desc.unmarshalArguments(s);
  pd_state = WaitingForReply;

  // Here we notify the giopServer that this thread has finished
  // reading the request. The server may want to keep a watch on
  // any more request coming in on the connection while this
  // thread does the upcall.

  CORBA::Boolean data_in_buffer = 0;
  if (pd_rdlocked) {
    // This is the thread that is reading from the connection. We
    // check if we have previously queued giopStream_Buffers on the
    // connection. In other words, we might have previously read
    // too much stuff out of the connection and these data belong to
    // other requests. If that is the case, we notify the giopServer
    // that there are already buffers waiting to be read.
    giopStrand& s = (giopStrand&) *this;
    data_in_buffer = ((s.head) ? 1 : 0);
  }
  pd_worker->server()->notifyWkPreUpCall(pd_worker,data_in_buffer);

  impl()->inputMessageEnd(this,0);


  // Check if this call comes in from a bidirectional connection.
  // If so check if the servant's POA policy allows this.
  giopStrand& g = (giopStrand&)((giopStream&)(*this));
  if (g.biDir && g.isClient()) {
    if (!(pd_calldescriptor->poa() &&
	  pd_calldescriptor->poa()->acceptBiDirectional())) {
      OMNIORB_THROW(OBJ_ADAPTER,OBJ_ADAPTER_BiDirNotAllowed,
		    CORBA::COMPLETED_NO);
    }
  }
}

////////////////////////////////////////////////////////////////////////
void
GIOP_S::SkipRequestBody() {

  OMNIORB_ASSERT(pd_state == RequestIsBeingProcessed);

  pd_state = WaitingForReply;

  CORBA::Boolean data_in_buffer = 0;
  if (pd_rdlocked) {
    giopStrand& s = (giopStrand&) *this;
    data_in_buffer = ((s.head) ? 1 : 0);
  }
  pd_worker->server()->notifyWkPreUpCall(pd_worker,data_in_buffer);

  impl()->inputMessageEnd(this,1);
}

////////////////////////////////////////////////////////////////////////
void
GIOP_S::SendReply() {

  OMNIORB_ASSERT(pd_state == WaitingForReply);

  if (!response_expected()) {
    pd_state = ReplyCompleted;
    return;
  }

  pd_service_contexts.length(0);

  if (omniInterceptorP::serverSendReply) {
    omniInterceptors::serverSendReply_T::info_T info(*this);
    omniInterceptorP::visit(info);
  }
  pd_state = ReplyIsBeingComposed;
  impl()->outputMessageBegin(this,impl()->marshalReplyHeader);
  cdrStream& s = *this;
  calldescriptor()->marshalReturnedValues(s);
  impl()->outputMessageEnd(this);
  pd_state = ReplyCompleted;
}

////////////////////////////////////////////////////////////////////////
void
GIOP_S::SendException(CORBA::Exception* ex) {

  OMNIORB_ASSERT(pd_state == WaitingForReply);

  if (!response_expected()) throw terminateProcessing();

  int idsize;
  const char* repoid = ex->_NP_repoId(&idsize);

# define TEST_AND_MARSHAL_SYSEXCEPTION(name) \
  if ( strcmp("IDL:omg.org/CORBA/" #name ":1.0",repoid) == 0 ) { \
    impl()->sendSystemException(this,*((CORBA::SystemException*)ex)); \
    pd_state = ReplyCompleted; \
    return; \
  }

  OMNIORB_FOR_EACH_SYS_EXCEPTION(TEST_AND_MARSHAL_SYSEXCEPTION)
# undef TEST_AND_MARSHAL_SYSEXCEPTION

  // this is not a system exception, treat it as a user exception
  // we do not check if this is a valid user exception for this operation.
  // caller should have checked this or else the user exception should have
  // been thrown as a C++ exception and got handled by the catch clause in
  // handleRequest.
  impl()->sendUserException(this,*((CORBA::UserException*)ex));
  pd_state = ReplyCompleted;
}

////////////////////////////////////////////////////////////////////////
void
GIOP_S::notifyCommFailure(CORBA::Boolean,
			  CORBA::ULong& minor,
			  CORBA::Boolean& retry) {
  retry = 0;

  if (pd_state == WaitForRequestHeader ||
      pd_state == RequestIsBeingProcessed) {
    minor = COMM_FAILURE_UnMarshalArguments;
  }
  else if (pd_state == WaitingForReply) {
    minor = COMM_FAILURE_WaitingForReply;
  }
  else if (pd_state == ReplyIsBeingComposed) {
    minor = COMM_FAILURE_MarshalResults;
  }
  else {
    minor = TRANSIENT_ConnectionClosed;
  }
}

////////////////////////////////////////////////////////////////////////
CORBA::ULong
GIOP_S::completion() {
  if (pd_state == WaitingForReply) {
    return (CORBA::ULong)CORBA::COMPLETED_MAYBE;
  }
  else if (pd_state == ReplyIsBeingComposed) {
    return (CORBA::ULong)CORBA::COMPLETED_YES;
  }
  else {
    return (CORBA::ULong)CORBA::COMPLETED_NO;
  }
}

////////////////////////////////////////////////////////////////////////
const char*
GIOP_S::operation_name() const {
  return operation();
}

////////////////////////////////////////////////////////////////////////
void
GIOP_S::unmarshalIORAddressingInfo() {
  GIOP::AddressingDisposition vp;
  CORBA::ULong   vl;

  resetKey();

  cdrStream& s = *this;

  vp <<= s;
  if (vp == GIOP::KeyAddr) {
    vl <<= s;
    if (!s.checkInputOverrun(1,vl)) {
      OMNIORB_THROW(MARSHAL,MARSHAL_SequenceIsTooLong,
		    (CORBA::CompletionStatus)completion());
    }
    keysize((int)vl);
    s.get_octet_array(key(),vl);
  }
  else {
    
    GIOP::IORAddressingInfo& ta = pd_target_address;
    if (vp == GIOP::ProfileAddr) {
      ta.ior.profiles.length(1);
      ta.ior.profiles[0] <<= s;
      ta.selected_profile_index = 0;
    }
    else {
      // GIOP::ReferenceAddr
      ta.selected_profile_index <<= s;
      ta.ior <<= s;
    }
    if (ta.selected_profile_index >= ta.ior.profiles.length() ||
	ta.ior.profiles[ta.selected_profile_index].tag != 
	                                              IOP::TAG_INTERNET_IOP) {
      if ( omniORB::trace(25) ) {
	omniORB::logger l;
	l << "unmarshal corrupted targetAddress at "
	  << __FILE__ << " line no. " << __LINE__ << "\n";
      }
      OMNIORB_THROW(BAD_PARAM,BAD_PARAM_IndexOutOfRange,
		    (CORBA::CompletionStatus)completion());
    }

    IIOP::ProfileBody decodedBody;
    IIOP::unmarshalProfile(ta.ior.profiles[ta.selected_profile_index],
			decodedBody);

#if 0 // XXX Not finalise yet
    _OMNI_NS(giopAddressList) addresses;
    IIOP::extractAddresses(decodedBody,addresses);
    if ( isLocal(addresses) ) {
      keysize((int)decodedBody.object_key.length());
      memcpy((void*)key(),decodedBody.object_key.get_buffer(),keysize());
    }
    // XXX delete all the addresses.
#else
    OMNIORB_ASSERT(0);
#endif

    // Reach here either we have got the key of the target object
    // or we have the target address info in targetAddress().
    //
    if (keysize() < 0) {
      // We couldn't decode the target address to a local object key. Unless
      // an interceptor can decode it further, this request will be rejected.
      if ( omniORB::trace(25) ) {
	omniORB::logger l;
	l << "ProfileAddr or ReferenceAddr addresses unknown target at "
	  << __FILE__ << " line no. " << __LINE__ << "\n";
      }
    }
  }

}

OMNI_NAMESPACE_END(omni)