giopimpl12.cc

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

  if (!g->pd_currentOutputBuffer) {
    g->pd_currentOutputBuffer = giopStream_Buffer::newBuffer();
  }
  g->pd_currentOutputBuffer->alignStart(omni::ALIGN_8);

  char* hdr = (char*)g->pd_currentOutputBuffer + 
                     g->pd_currentOutputBuffer->start;

  hdr[0] = 'G'; hdr[1] = 'I'; hdr[2] = 'O'; hdr[3] = 'P';
  hdr[4] = 1;   hdr[5] = 2;   hdr[6] = _OMNIORB_HOST_BYTE_ORDER_;

  g->pd_outb_mkr = (void*)(hdr + 12);
  g->pd_outb_end = (void*)((omni::ptr_arith_t)g->pd_currentOutputBuffer + 
			   g->pd_currentOutputBuffer->end);
  g->outputFragmentSize(0);
  g->outputMessageSize(0);
  *((CORBA::ULong*)(hdr + 8)) = 0xffffffff;
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::outputMessageBegin(giopStream* g,
			       void (*marshalHeader)(giopStream*)) {

  outputNewMessage(g);
  marshalHeader(g);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::outputMessageEnd(giopStream* g) {

  if (g->pd_currentOutputBuffer) {

    omni::ptr_arith_t outbuf_begin = ((omni::ptr_arith_t) 
				    g->pd_currentOutputBuffer + 
				    g->pd_currentOutputBuffer->start);

    if ( outbuf_begin != (omni::ptr_arith_t)g->pd_outb_mkr ) {

      if (!g->outputFragmentSize()) {


	CORBA::ULong sz = (omni::ptr_arith_t)g->pd_outb_mkr - outbuf_begin -12;

	if (!g->outputMessageSize()) {
	  // The whole message is still in the buffer, we have to check
	  // if any arguments have been marshalled after the header.
	  // If there is none, we have to remove the paddings at the end
	  // of the header.
	  // The header's size is stored in the buffer
	  CORBA::ULong hdrsz = *((CORBA::ULong*)(outbuf_begin + 8));
	  OMNIORB_ASSERT(hdrsz != 0xffffffff);
	  if ( omni::align_to(hdrsz,omni::ALIGN_8) == (omni::ptr_arith_t)(sz+12)) {
	    // No argument has been marshalled, remove the padding
	    sz = hdrsz - 12;
	    g->pd_outb_mkr = (void*)(outbuf_begin + hdrsz);
	  }
	}
	*((CORBA::ULong*)(outbuf_begin + 8)) = sz;
	// g->outputMessageSize(g->outputMessageSize()+sz);
      }

      g->pd_currentOutputBuffer->last = (omni::ptr_arith_t) g->pd_outb_mkr - 
					(omni::ptr_arith_t) 
                                        g->pd_currentOutputBuffer;

      g->sendChunk(g->pd_currentOutputBuffer);
    
    }
    // Notice that we do not release the buffer. Next time this giopStream
    // is re-used, the buffer will be reused as well.
  }

  {
    omni_tracedmutex_lock sync(*omniTransportLock);
    g->wrUnLock();
  }

}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::sendMsgErrorMessage(giopStream* g) {

  if (!g->pd_wrlocked) {
    omni_tracedmutex_lock sync(*omniTransportLock);
    g->wrLock();
  }


  if (omniORB::trace(1)) {
    omniORB::logger l;
    l << "To endpoint: " << g->pd_strand->connection->peeraddress()
      <<". Send GIOP 1.2 MessageError because a protocol error has been detected. "
      << "Connection is closed.\n";
  }

  if (!g->pd_currentOutputBuffer) {
    g->pd_currentOutputBuffer = giopStream_Buffer::newBuffer();
  }
  g->pd_currentOutputBuffer->alignStart(omni::ALIGN_8);

  char* hdr = (char*)g->pd_currentOutputBuffer + 
                     g->pd_currentOutputBuffer->start;

  hdr[0] = 'G'; hdr[1] = 'I'; hdr[2] = 'O'; hdr[3] = 'P';
  hdr[4] = 1;   hdr[5] = 2;   hdr[6] = _OMNIORB_HOST_BYTE_ORDER_;

  g->pd_outb_mkr = (void*)(hdr + 12);
  g->pd_outb_end = (void*)((omni::ptr_arith_t)g->pd_currentOutputBuffer + 
			   g->pd_currentOutputBuffer->end);
  g->outputFragmentSize(0);
  g->outputMessageSize(0);

  hdr[7] = (char)GIOP::MessageError;
  hdr[8] = hdr[9] = hdr[10] = hdr[11] = 0;

  (void)  g->pd_strand->connection->Send(hdr,12);

  g->pd_strand->state(giopStrand::DYING);

  {
    omni_tracedmutex_lock sync(*omniTransportLock);
    g->wrUnLock();
  }
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::marshalRequestHeader(giopStream* g) {

  char* hdr = (char*) g->pd_currentOutputBuffer + 
                      g->pd_currentOutputBuffer->start;

  hdr[7] = (char) GIOP::Request;

  GIOP_C& giop_c = *(GIOP_C*)g;
  cdrStream& s = (cdrStream&) *g;
  omniCallDescriptor& calldesc = *giop_c.calldescriptor();
  CORBA::Boolean response_expected = !calldesc.is_oneway();
  

  omniInterceptors::clientSendRequest_T::info_T info(*g,
				                     *(giop_c.ior()),
						     calldesc.op(),
						     !response_expected,
						     response_expected);
  omniInterceptorP::visit(info);

  CORBA::Octet v = 0;
  {
    // calculate the request header size
    cdrCountingStream cs(g->TCS_C(),g->TCS_W(),12);
    operator>>= ((CORBA::ULong)0,cs);
    cs.marshalOctet(v);
    cs.marshalOctet(v);
    cs.marshalOctet(v);
    cs.marshalOctet(v);

    if (giop_c.ior()->addr_mode() == GIOP::KeyAddr) {
      giop_c.ior()->addr_mode() >>= cs;
      giop_c.keysize() >>= cs;
      cs.put_octet_array(giop_c.key(),giop_c.keysize());
    }
    else {
      giop_c.ior()->marshalIORAddressingInfo(cs);
    }

    operator>>= ((CORBA::ULong) calldesc.op_len(),cs);
    cs.put_octet_array((CORBA::Octet*) calldesc.op(), calldesc.op_len());
    info.service_contexts >>= cs;
    *((CORBA::ULong*)(hdr+8)) = cs.total();
  }

  // request id
  giop_c.requestId() >>= s;

  // response flags
  v = ((response_expected << 1) + response_expected);
  s.marshalOctet(v);

  // reserved[3]
  v = 0;
  s.marshalOctet(v);
  s.marshalOctet(v);
  s.marshalOctet(v);

  // Target address
  if (giop_c.ior()->addr_mode() == GIOP::KeyAddr) {
    giop_c.ior()->addr_mode() >>= s;
    giop_c.keysize() >>= s;
    s.put_octet_array(giop_c.key(),giop_c.keysize());
  }
  else {
    giop_c.ior()->marshalIORAddressingInfo(s);
  }

  // operation
  operator>>= ((CORBA::ULong) calldesc.op_len(),s);
  s.put_octet_array((CORBA::Octet*) calldesc.op(), calldesc.op_len());

  // service context
  info.service_contexts >>= s;

  s.alignOutput(omni::ALIGN_8);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::sendLocateRequest(giopStream* g) {

  outputNewMessage(g);

  char* hdr = (char*) g->pd_currentOutputBuffer + 
                      g->pd_currentOutputBuffer->start;

  hdr[7] = (char)GIOP::LocateRequest;

  GIOP_C& giop_c = *(GIOP_C*)g;
  cdrStream& s = (cdrStream&) *g;

  // Compute and initialise the message size field
  {
    cdrCountingStream cs(g->TCS_C(),g->TCS_W(),12);
    operator>>= ((CORBA::ULong)0,cs);
    giop_c.ior()->marshalIORAddressingInfo(cs);
    outputSetFragmentSize(g,cs.total()-12);
    *((CORBA::ULong*)(hdr + 8)) = cs.total() - 12;
  }

  // request id
  giop_c.requestId() >>= s;
  
  // object key
  giop_c.ior()->marshalIORAddressingInfo(s);

  outputMessageEnd(g);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::marshalReplyHeader(giopStream* g) {

  char* hdr = (char*) g->pd_currentOutputBuffer + 
                      g->pd_currentOutputBuffer->start;

  hdr[7] = (char) GIOP::Reply;

  GIOP_S& giop_s = *(GIOP_S*)g;
  cdrStream& s = (cdrStream&) *g;

  CORBA::ULong rc = GIOP::NO_EXCEPTION;
  {
    // calculate the reply header size
    cdrCountingStream cs(g->TCS_C(),g->TCS_W(),12);
    operator>>= ((CORBA::ULong)0,cs);
    rc >>= cs;
    giop_s.service_contexts() >>= cs;
    *((CORBA::ULong*)(hdr+8)) = cs.total();
  }

  // request id
  giop_s.requestId() >>= s;

  // reply status
  rc >>= s;

  // Service context
  giop_s.service_contexts() >>= s;
  
  s.alignOutput(omni::ALIGN_8);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::sendSystemException(giopStream* g,const CORBA::SystemException& ex) {
  GIOP_S& giop_s = *(GIOP_S*)g;
  cdrStream& s = (cdrStream&) *g;

  if (giop_s.state() == GIOP_S::ReplyIsBeingComposed) {
    // This system exception is raised during the marshalling of the reply.
    // We cannot marshal the exception. Can only indicate that something
    // fatal about this request.
    sendMsgErrorMessage(g);

    CORBA::ULong minor;
    CORBA::Boolean retry;
    giop_s.notifyCommFailure(0,minor,retry);
    giopStream::CommFailure::_raise(minor,(CORBA::CompletionStatus)
				    giop_s.completion(),
				    retry,__FILE__,__LINE__);
  }

  int repoid_size;
  const char* repoid = ex._NP_repoId(&repoid_size);

  outputNewMessage(g);

  char* hdr = (char*) g->pd_currentOutputBuffer + 
                      g->pd_currentOutputBuffer->start;

  hdr[7] = (char) GIOP::Reply;

  giop_s.service_contexts().length(0);

  if (omniInterceptorP::serverSendException) {
    omniInterceptors::serverSendException_T::info_T info(giop_s, &ex);
    omniInterceptorP::visit(info);

    if (giop_s.service_contexts().length() > 0) {

      // Compute and initialise the message size field. Only necessary
      // if there are service contexts, since we know a message without
      // service contexts will fit in a single buffer.

      cdrCountingStream cs(g->TCS_C(),g->TCS_W(),12);
      operator>>= ((CORBA::ULong)0,cs);
      operator>>= ((CORBA::ULong)0,cs);
      giop_s.service_contexts() >>= cs;
      cs.alignOutput(omni::ALIGN_8);

      CORBA::ULong(repoid_size) >>= cs;
      cs.put_octet_array((const CORBA::Octet*) repoid, repoid_size);
      ex.minor() >>= cs;
      operator>>= ((CORBA::ULong)0,cs);

      outputSetFragmentSize(g,cs.total()-12);
      *((CORBA::ULong*)(hdr + 8)) = cs.total() - 12;
    }
  }

  // request id
  giop_s.requestId() >>= s;

  // reply status
  CORBA::ULong rc = GIOP::SYSTEM_EXCEPTION;
  rc >>= s;

  // service contexts
  giop_s.service_contexts() >>= s;

  // End of reply header

  if (giop_s.service_contexts().length() == 0){
    // Set header size in buffer
    CORBA::ULong sz = ((omni::ptr_arith_t)g->pd_outb_mkr - 
		       (omni::ptr_arith_t)hdr - 12);
    *((CORBA::ULong*)(hdr + 8)) = sz;
  }

  s.alignOutput(omni::ALIGN_8);

  // RepoId
  CORBA::ULong(repoid_size) >>= s;
  s.put_octet_array((const CORBA::Octet*) repoid, repoid_size);

  // system exception value
  ex.minor() >>= s;

  CORBA::ULong(ex.completed()) >>= s;

  outputMessageEnd(g);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::sendUserException(giopStream* g,const CORBA::UserException& ex) {

  GIOP_S& giop_s = *(GIOP_S*)g;
  cdrStream& s = (cdrStream&) *g;

  int repoid_size;
  const char* repoid = ex._NP_repoId(&repoid_size);

  outputNewMessage(g);

  char* hdr = (char*)g->pd_currentOutputBuffer + 
                     g->pd_currentOutputBuffer->start;

  hdr[7] = (char)GIOP::Reply;

  giop_s.service_contexts().length(0);

  if (omniInterceptorP::serverSendException) {
    omniInterceptors::serverSendException_T::info_T info(giop_s, &ex);
    omniInterceptorP::visit(info);
  }

  // user exception value

  // Compute and initialise the message size field
  {
    cdrCountingStream cs(g->TCS_C(),g->TCS_W(),12);
    operator>>= ((CORBA::ULong)0,cs);
    operator>>= ((CORBA::ULong)0,cs);
    giop_s.service_contexts() >>= cs;
    CORBA::ULong(repoid_size) >>= cs;
    cs.put_octet_array((const CORBA::Octet*) repoid, repoid_size);
    ex._NP_marshal(cs);

    outputSetFragmentSize(g,cs.total()-12);
    *((CORBA::ULong*)(hdr + 8)) = cs.total() - 12;
  }

  // request id
  giop_s.requestId() >>= s;

  // reply status
  CORBA::ULong rc = GIOP::USER_EXCEPTION;
  rc >>= s;

  // Service context
  giop_s.service_contexts() >>= s;

  // RepoId
  CORBA::ULong(repoid_size) >>= s;
  s.put_octet_array((const CORBA::Octet*) repoid, repoid_size);

  // Exception value
  ex._NP_marshal(s);

  outputMessageEnd(g);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::sendLocationForwardReply(giopStream* g,CORBA::Object_ptr obj,
				     CORBA::Boolean permanent) {

  outputNewMessage(g);
  
  char* hdr = (char*)g->pd_currentOutputBuffer + 
                     g->pd_currentOutputBuffer->start;

  hdr[7] = (char)GIOP::Reply;