giopimpl12.cc

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

    // Same treatment as wrong header
    inputTerminalProtocolError(g, __FILE__, __LINE__);
    // Never reach here.
  }

  giop_c.replyStatus((GIOP::ReplyStatusType)v);

  if (omniInterceptorP::clientReceiveReply) {
    IOP::ServiceContextList sctxts;
    sctxts <<= s;
    omniInterceptors::clientReceiveReply_T::info_T info(giop_c, sctxts);
    omniInterceptorP::visit(info);
  }
  else {
    // Skip service context
    CORBA::ULong svcccount;
    CORBA::ULong svcctag;
    CORBA::ULong svcctxtsize;
    svcccount <<= s;
    while (svcccount-- > 0) {
      svcctag <<= s;
      svcctxtsize <<= s;
      s.skipInput(svcctxtsize);
    }
  }
  if (inputRemaining(g))
    s.alignInput(omni::ALIGN_8);
}

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

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

  if ((GIOP::MsgType)hdr[7] != GIOP::LocateReply) {
    // Unexpected reply. The other end is terribly confused. Drop the
    // connection and died.
    inputTerminalProtocolError(g, __FILE__, __LINE__);
    // Never reach here.
  }

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

  // We have already verified the request id in the header and the stream
  // have been setup to go pass it

  CORBA::ULong v;
  v <<= s;

  switch (v) {
  case GIOP::UNKNOWN_OBJECT:
  case GIOP::OBJECT_HERE:
  case GIOP::OBJECT_FORWARD:
  case GIOP::OBJECT_FORWARD_PERM:
  case GIOP::LOC_SYSTEM_EXCEPTION:
  case GIOP::LOC_NEEDS_ADDRESSING_MODE:
    break;
  default:
    // Should never receive anything other that the above
    // Same treatment as wrong header
    inputTerminalProtocolError(g, __FILE__, __LINE__);
    // never reach here.
  }
  giop_c.locateStatus((GIOP::LocateStatusType)v);
}

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

  g->inputMatchedId(1);

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

  ((GIOP_S*)g)->requestType((GIOP::MsgType)hdr[7]);
  switch (((GIOP_S*)g)->requestType()) {
  case GIOP::Request:
  case GIOP::LocateRequest:
  case GIOP::CancelRequest:
    CORBA::ULong reqid;
    reqid <<= (cdrStream&)(*g);
    ((GIOP_S*)g)->requestId(reqid);
    break;
  case GIOP::CloseConnection:
    if (g->pd_strand->biDir && g->pd_strand->isClient()) {
      // proper shutdown of a connection.
      if (omniORB::trace(30)) {
	omniORB::logger l;
	l << "Server has closed a bi-directional connection on strand "
	  << (void*)g->pd_strand << ". Will scavenge it.\n";
      }
      g->pd_strand->startIdleCounter();
    }
    inputRaiseCommFailure(g);
    break;
  default:
    inputTerminalProtocolError(g, __FILE__, __LINE__);
    // Never reach here.
    break;
  }
}

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

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

  // We have already verified the request id in the header and the stream
  // have been setup to go pass it

  CORBA::ULong   vl;
  CORBA::Octet   vb;

  // response flag
  vb = s.unmarshalOctet();
  giop_s.response_expected((vb & 0x1));
  giop_s.result_expected((vb & 0x2) ? 1 : 0);

  // padding
  vb = s.unmarshalOctet();
  vb = s.unmarshalOctet();
  vb = s.unmarshalOctet();

  // object key
  giop_s.unmarshalIORAddressingInfo();

  // operation
  vl <<= s;
  if (!vl || !s.checkInputOverrun(1,vl)) {
    OMNIORB_THROW(MARSHAL,MARSHAL_InvalidVariableLenComponentSize,
		  CORBA::COMPLETED_NO);
  }
  giop_s.set_operation_size(vl);
  char* op = giop_s.operation();
  s.get_octet_array((CORBA::Octet*)op,vl);
  op[vl-1] = '\0';

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

  if (inputRemaining(g))
    s.alignInput(omni::ALIGN_8);
}

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

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


  // We have already verified the request id in the header and the stream
  // have been setup to go pass it

  giop_s.unmarshalIORAddressingInfo();
}

////////////////////////////////////////////////////////////////////////
size_t
giopImpl12::inputRemaining(giopStream* g) {

  if (g->inputExpectAnotherFragment()) {
    return orbParameters::giopMaxMsgSize - currentInputPtr(g);
  }
  else {
    return (g->inputFragmentToCome() + ((omni::ptr_arith_t)g->pd_inb_end -
					(omni::ptr_arith_t)g->pd_inb_mkr));
  }
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::getInputData(giopStream* g,omni::alignment_t align,size_t sz) {

 again:
  omni::ptr_arith_t last = omni::align_to((omni::ptr_arith_t)g->pd_inb_mkr,
					  align);

  omni::ptr_arith_t end = last + sz;

  if ( end <= (omni::ptr_arith_t) g->pd_inb_end) {
    return;
  }
  else {
    // Invariant check 
    // Either the full message is already in the buffer or the part that is
    // in the buffer always ends at an 8 byte aligned boundary.
    // Also remember that sz is <= 8 always!
    size_t extra = end - (omni::ptr_arith_t) g->pd_inb_end;
    if (extra != sz) {
      if ( !(g->inputFragmentToCome() || g->inputExpectAnotherFragment()) ) {
	// The full message is already in the buffer. The unmarshalling
	// code is asking for more. This is an error causes by the received
	// data. We'll let the code below to raise a MARSHAL exception
	sz = extra;            // in case sz == 0
      }
      else {

	if (g->inputExpectAnotherFragment()) {
	  // The incoming message is fragmented at the wrong boundary!!!
	  inputTerminalProtocolError(g, __FILE__, __LINE__);
	  // never reach here
	}
	// Very bad. Should never happen given our invariant.
	{
	  if( omniORB::trace(1) ) {
	    omniORB::logger l;
	    l << "Fatal error in unmarshalling message from "
	      << g->pd_strand->connection->peeraddress()
	      << ", invariant was violated at "
	      << __FILE__ 	
	      << ":" << __LINE__ << '\n';
	  }
	}
	OMNIORB_ASSERT(0);
	// never reach here.
      }
    }
  }

  if (g->inputFragmentToCome() < sz && !g->inputExpectAnotherFragment()) {
    OMNIORB_THROW(MARSHAL,MARSHAL_PassEndOfMessage,
		  (CORBA::CompletionStatus)g->completion());
  }

  if (!g->inputFragmentToCome()) {
    inputNewFragment(g);
    if (g->inputMessageSize() > orbParameters::giopMaxMsgSize) {
      OMNIORB_THROW(MARSHAL,MARSHAL_MessageSizeExceedLimit,
		    (CORBA::CompletionStatus)g->completion());
    }
    goto again;
  }

  // Reach here if we have some bytes to fetch for the current fragment
  if (g->pd_currentInputBuffer) {
    g->releaseInputBuffer(g->pd_currentInputBuffer);
    g->pd_currentInputBuffer = 0;
  }

  if (!g->pd_input) {
    g->pd_currentInputBuffer = g->inputChunk(g->inputFragmentToCome());
  }
  else {
    g->pd_currentInputBuffer = g->pd_input;
    g->pd_input = g->pd_currentInputBuffer->next;
    g->pd_currentInputBuffer->next = 0;
  }

  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));
}


////////////////////////////////////////////////////////////////////////
void
giopImpl12::skipInputData(giopStream* g,size_t sz) {
  copyInputData(g,0,sz,omni::ALIGN_1);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::copyInputData(giopStream* g,void* b, size_t sz,
			  omni::alignment_t align) {

  // If b == 0, we don't actually copy the data but just skip <sz> bytes.

  omni::ptr_arith_t last = omni::align_to((omni::ptr_arith_t)g->pd_inb_mkr,
					  align);

  if ( last > (omni::ptr_arith_t) g->pd_inb_end ) {
    // Invariant check 
    // Either the full message is already in the buffer or the part that is
    // in the buffer always ends at an 8 byte aligned boundary.
    if ( !(g->inputFragmentToCome() || g->inputExpectAnotherFragment()) ) {
      // The full message is already in the buffer. The unmarshalling
      // code is asking for more. This is an error causes by the received
      // data.
      OMNIORB_THROW(MARSHAL,MARSHAL_PassEndOfMessage,
		    (CORBA::CompletionStatus)g->completion());
    }
    else {

      if (g->inputExpectAnotherFragment()) {
	// The incoming message is fragmented at the wrong boundary!!!
	inputTerminalProtocolError(g, __FILE__, __LINE__);
	// never reach here
      }
      // Very bad. Should never happen given our invariant.
      {
	if( omniORB::trace(1) ) {
	  omniORB::logger l;
	  l << "Fatal error in unmarshalling message from "
	    << g->pd_strand->connection->peeraddress()
	    << ", invariant was violated at "
	    << __FILE__ 	
	    << ":" << __LINE__ << '\n';
	}
      }
      OMNIORB_ASSERT(0);
      // never reach here.
    }
  }

  g->pd_inb_mkr = (void*) last;
  while (sz) {

    size_t avail = (omni::ptr_arith_t) g->pd_inb_end - 
                   (omni::ptr_arith_t) g->pd_inb_mkr;
    if (avail >= sz) {
      avail = sz;
    }
    if (avail) {
      if (b) memcpy(b,g->pd_inb_mkr,avail);
      g->pd_inb_mkr = (void*)((omni::ptr_arith_t)g->pd_inb_mkr + avail);
    }
    sz -= avail;
    if (b) b = (void*)((omni::ptr_arith_t)b + avail);

    if (!sz) break;

    if (g->pd_inb_mkr == g->pd_inb_end) {

      if (g->inputFragmentToCome() < sz && !g->inputExpectAnotherFragment()) {
	OMNIORB_THROW(MARSHAL,MARSHAL_PassEndOfMessage,
		      (CORBA::CompletionStatus)g->completion());
	// never reach here
      }

      if (!g->inputFragmentToCome()) {
	inputNewFragment(g);
	if (g->inputMessageSize() > orbParameters::giopMaxMsgSize) {
	  OMNIORB_THROW(MARSHAL,MARSHAL_MessageSizeExceedLimit,
			(CORBA::CompletionStatus)g->completion());
	}
	continue;
      }

      if (g->pd_currentInputBuffer) {
	g->releaseInputBuffer(g->pd_currentInputBuffer);
	g->pd_currentInputBuffer = 0;
      }
      if (g->pd_input) {
	g->pd_currentInputBuffer = g->pd_input;
	g->pd_input = g->pd_currentInputBuffer->next;
	g->pd_currentInputBuffer->next = 0;
      }
      else {

	if ( b && sz >= giopStream::directReceiveCutOff ) {
	  
	  CORBA::ULong transz = g->inputFragmentToCome();
	  if (transz > sz) transz = sz;
	  transz = (transz >> 3) << 3;
	  g->inputCopyChunk(b,transz);
	  sz -= transz;
	  b = (void*)((omni::ptr_arith_t)b + transz);
	  g->inputFragmentToCome(g->inputFragmentToCome() - transz);
	  continue;
	}
	else {
	  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
giopImpl12::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
giopImpl12::inputTerminalProtocolError(giopStream* g,
				       const char* file, int line)
{
  // XXX We may choose to send a message error to the other end.
  if (omniORB::trace(1)) {
    omniORB::logger l;
    l << "From endpoint: " << g->pd_strand->connection->peeraddress()
      << ". Detected GIOP 1.2 protocol error in input message. "
      << file << ":" << line
      << ". Connection is closed.\n";
  }

  inputRaiseCommFailure(g);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::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
giopImpl12::outputNewMessage(giopStream* g) {

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