giopimpl10.cc

来自「编译工具」· CC 代码 · 共 1,684 行 · 第 1/4 页

	// We keep the buffer around until the id of the reply is established.

	if (g->pd_currentInputBuffer) {
	  giopStream_Buffer** pp = &g->pd_input;
	  while (*pp) {
	    pp = &((*pp)->next);
	  }
	  *pp = g->pd_currentInputBuffer;
	  g->pd_currentInputBuffer = 0;
	  g->pd_currentInputBuffer = g->inputChunk(g->inputFragmentToCome());
	}
      }

      g->pd_inb_mkr = (void*)((omni::ptr_arith_t)g->pd_currentInputBuffer + 
			      g->pd_currentInputBuffer->start);
      g->pd_inb_end = (void*)((omni::ptr_arith_t)g->pd_currentInputBuffer + 
			      g->pd_currentInputBuffer->last);
      g->inputFragmentToCome(g->inputFragmentToCome() - 
			     (g->pd_currentInputBuffer->last -
			      g->pd_currentInputBuffer->start));
    }
  }	
}

////////////////////////////////////////////////////////////////////////
CORBA::ULong
giopImpl10::currentInputPtr(const giopStream* g) {

  return  g->inputMessageSize() - 
          g->inputFragmentToCome() -
         ((omni::ptr_arith_t) g->pd_inb_end - 
	  (omni::ptr_arith_t) g->pd_inb_mkr);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl10::inputTerminalProtocolError(giopStream* g) {

  sendMsgErrorMessage(g);

  inputRaiseCommFailure(g);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl10::inputRaiseCommFailure(giopStream* g) {

  CORBA::ULong minor;
  CORBA::Boolean retry;
  g->notifyCommFailure(0,minor,retry);
  g->pd_strand->state(giopStrand::DYING);
  giopStream::CommFailure::_raise(minor,
				  (CORBA::CompletionStatus)g->completion(),
				  0,__FILE__,__LINE__);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl10::outputNewMessage(giopStream* g) {

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

  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] = 0;   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);
}

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

  outputNewMessage(g);
  marshalHeader(g);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl10::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->outputMessageSize()) {

	CORBA::ULong sz = (omni::ptr_arith_t)g->pd_outb_mkr - outbuf_begin -12;
	*((CORBA::ULong*)(outbuf_begin + 8)) = sz;
	outputSetMessageSize(g,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
giopImpl10::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.0 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] = 0;   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
giopImpl10::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);

  {
    // calculate the request header size
    cdrCountingStream cs(g->TCS_C(),g->TCS_W(),12);
    info.service_contexts >>= cs;
    giop_c.requestId() >>= cs;
    cs.marshalBoolean(response_expected);
    giop_c.keysize() >>= cs;
    cs.put_octet_array(giop_c.key(),giop_c.keysize());
    operator>>= ((CORBA::ULong)calldesc.op_len(),cs);
    cs.put_octet_array((CORBA::Octet*) calldesc.op(), calldesc.op_len());
    omni::myPrincipalID >>= cs;
    *((CORBA::ULong*)(hdr+8)) = cs.total();

#if defined(PRE_CALCULATE_MESSAGE_SIZE)
    giop_c.calldescriptor()->marshalArguments(cs);
    CORBA::ULong msgsz = cs.total() - 12;
    *((CORBA::ULong*)(hdr + 8)) = msgsz;
    outputSetMessageSize(g,msgsz);
#endif
  }

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

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

  // response expected flag
  s.marshalBoolean(response_expected);

  // object key
  giop_c.keysize() >>= s;
  s.put_octet_array(giop_c.key(),giop_c.keysize());

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

  // principal
  omni::myPrincipalID >>= s;
}

////////////////////////////////////////////////////////////////////////
void
giopImpl10::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
  {
    CORBA::ULong totalsz = (omni::ptr_arith_t)g->pd_outb_mkr - 
                           (omni::ptr_arith_t)hdr;

    cdrCountingStream cs(g->TCS_C(),g->TCS_W(),totalsz);
    giop_c.requestId() >>= cs;
    giop_c.keysize() >>= cs;
    cs.put_octet_array(giop_c.key(),giop_c.keysize());
    outputSetMessageSize(g,cs.total()-12);
    *((CORBA::ULong*)(hdr + 8)) = cs.total() - 12;
  }

  // request id
  giop_c.requestId() >>= s;
  
  // object key
  giop_c.keysize() >>= s;
  s.put_octet_array(giop_c.key(),giop_c.keysize());

  outputMessageEnd(g);
}



////////////////////////////////////////////////////////////////////////
void
giopImpl10::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 request header size
    cdrCountingStream cs(g->TCS_C(),g->TCS_W(),12);
    giop_s.service_contexts() >>= cs;
    giop_s.requestId() >>= cs;
    rc >>= cs;
    *((CORBA::ULong*)(hdr+8)) = cs.total();

#if defined(PRE_CALCULATE_MESSAGE_SIZE)
    giop_s.calldescriptor()->marshalReturnedValues(cs);
    CORBA::ULong msgsz = cs.total() - 12;
    *((CORBA::ULong*)(hdr + 8)) = msgsz;
    outputSetMessageSize(g,msgsz);
#endif
  }

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

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

  // reply status
  rc >>= s;
}

////////////////////////////////////////////////////////////////////////
void
giopImpl10::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.
    // If we have already send part of the message out, it is too late
    // to marshal the exception. Can only indicate that something
    // fatal about this request.

#if defined(PRE_CALCULATE_MESSAGE_SIZE)
    if (0)
#else
    if (g->outputMessageSize())
#endif
      {
	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);
      giop_s.service_contexts() >>= cs;
      operator>>= ((CORBA::ULong)0,cs);
      operator>>= ((CORBA::ULong)0,cs);
      CORBA::ULong(repoid_size) >>= cs;
      cs.put_octet_array((const CORBA::Octet*) repoid, repoid_size);
      ex.minor() >>= cs;
      operator>>= ((CORBA::ULong)0,cs);

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

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

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

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

  // system exception value
  CORBA::ULong(repoid_size) >>= s;
  s.put_octet_array((const CORBA::Octet*) repoid, repoid_size);
  ex.minor() >>= s;
  CORBA::ULong(ex.completed()) >>= s;

  outputMessageEnd(g);

}

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

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

  int i, 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);