giopimpl12.cc

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

  GIOP_S& giop_s = *(GIOP_S*)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);
    operator>>= ((CORBA::ULong)0,cs);
    operator>>= ((CORBA::ULong)0,cs);
    CORBA::Object::_marshalObjRef(obj,cs);
    outputSetFragmentSize(g,cs.total()-12);
    *((CORBA::ULong*)(hdr + 8)) = cs.total() - 12;
  }

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

  // reply status
  CORBA::ULong rc = (permanent ? GIOP::LOCATION_FORWARD_PERM : 
                                 GIOP::LOCATION_FORWARD);
  rc >>= s;

  // Service context
  operator>>= ((CORBA::ULong)0,s);


  // object reference
  CORBA::Object::_marshalObjRef(obj,s);
  outputMessageEnd(g);
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::sendLocateReply(giopStream* g,GIOP::LocateStatusType rc,
			    CORBA::Object_ptr obj,CORBA::SystemException* p) {


  outputNewMessage(g);

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

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

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

  CORBA::Object_ptr extra = CORBA::Object::_nil();

  // reply status
  switch (rc) {
  case GIOP::OBJECT_FORWARD:
  case GIOP::OBJECT_FORWARD_PERM:
    extra = obj;
    break;
  case GIOP::LOC_SYSTEM_EXCEPTION:
  case GIOP::UNKNOWN_OBJECT:
  case GIOP::OBJECT_HERE:
    break;
  case GIOP::LOC_NEEDS_ADDRESSING_MODE:
    // We never use this!
  default:
    // Never reach here.
    OMNIORB_ASSERT(0);
    break;
  }

  int repoid_size;
  const char* repoid;
  
  // 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);
    if (!CORBA::is_nil(extra)) {
      CORBA::Object::_marshalObjRef(extra,cs);
    }
    else if (rc == GIOP::LOC_SYSTEM_EXCEPTION) {
      repoid = p->_NP_repoId(&repoid_size);
      operator>>= ((CORBA::ULong)0,cs);
      cs.put_octet_array((const CORBA::Octet*) repoid, repoid_size);
      operator>>= ((CORBA::ULong)0,cs);
      operator>>= ((CORBA::ULong)0,cs);
    }

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


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

  operator>>= ((CORBA::ULong)rc,s);

  if (!CORBA::is_nil(extra)) {
    CORBA::Object::_marshalObjRef(extra,s);
  }
  else if (rc == GIOP::LOC_SYSTEM_EXCEPTION) {
    CORBA::ULong(repoid_size) >>= s;
    s.put_octet_array((const CORBA::Octet*) repoid, repoid_size);
    p->minor() >>= s;
    operator>>= ((CORBA::ULong)p->completed(),s);
  }

  outputMessageEnd(g);
}

////////////////////////////////////////////////////////////////////////
size_t
giopImpl12::outputRemaining(const giopStream* g) {

  CORBA::ULong total = g->outputFragmentSize();
  if (!total) {
    CORBA::ULong avail = (CORBA::Long)orbParameters::giopMaxMsgSize -
                         (CORBA::Long)currentOutputPtr(g);
    
    // Adjust avail to exactly the same value as calculated in outputFlush().
    // See the comment in outputFlush() for the reason why.
    avail = ((avail + 4) >> 3) << 3;
    avail = (avail ? avail - 4 : 0);
    return avail;
  }
  else {
    return total - currentOutputPtr(g);
  }
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::outputFlush(giopStream* g,CORBA::Boolean knownFragmentSize) {

  // Note: g->outputFragmentSize() == 0 implies that the full message
  //       size has been pre-calculated and no GIOP Fragment should be
  //       sent! This also means that the message size limit has been
  //       checked and there is no need to check against
  //       orbParameters::giopMaxMsgSize.
  //       

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

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

  if (!g->outputFragmentSize()) {
    ((char*)outbuf_begin)[6] |= 0x2;
    CORBA::ULong sz;
    if (!knownFragmentSize) {
      sz = fsz - 12;
      *((CORBA::ULong*)((omni::ptr_arith_t)outbuf_begin + 8)) = sz;
    }
    else {
      sz = *((CORBA::ULong*)((omni::ptr_arith_t)outbuf_begin + 8));
    }
    g->outputMessageSize(g->outputMessageSize()+sz);
    if (g->outputMessageSize() > orbParameters::giopMaxMsgSize) {
      OMNIORB_THROW(MARSHAL,MARSHAL_MessageSizeExceedLimitOnClient,
		    (CORBA::CompletionStatus)g->completion());
    }
  }
  else {
    CORBA::Long msz = g->outputMessageSize();
    if (msz) {
      g->outputMessageSize(msz+fsz);
    }
    else {
      g->outputMessageSize(fsz-12);
    }
  }

  g->pd_currentOutputBuffer->last = g->pd_currentOutputBuffer->start + fsz;
  g->sendChunk(g->pd_currentOutputBuffer);

  if (outbuf_begin & 0x7) {
    // start has previously been changed to non 8-bytes aligned
    g->pd_currentOutputBuffer->alignStart(omni::ALIGN_8);
    outbuf_begin = ((omni::ptr_arith_t) g->pd_currentOutputBuffer + 
		                        g->pd_currentOutputBuffer->start);
  }

  g->pd_outb_mkr = (void*) outbuf_begin;
  g->pd_outb_end = (void*)((omni::ptr_arith_t)g->pd_currentOutputBuffer + 
			   g->pd_currentOutputBuffer->end);
  g->pd_currentOutputBuffer->last = g->pd_currentOutputBuffer->start;

  if (!g->outputFragmentSize()) {
    char* hdr = (char*) outbuf_begin;

    hdr[0] = 'G'; hdr[1] = 'I'; hdr[2] = 'O'; hdr[3] = 'P';
    hdr[4] = 1;   hdr[5] = 2;   hdr[6] = _OMNIORB_HOST_BYTE_ORDER_;
    hdr[7] = (char)GIOP::Fragment;
    *((CORBA::ULong*)(hdr + 12)) = g->requestId();
    g->pd_outb_mkr = (void*)(outbuf_begin + 16);

    // Now determine how much space we have left.
    // If the message size has already reach orbParameters::giopMaxMsgSize,
    // outputHasReachedLimit() will return TRUE.
    CORBA::ULong avail = orbParameters::giopMaxMsgSize - g->outputMessageSize();

    // Adjust avail to make sure that it a multiple of 8.
    // This preserves our invariant: g->pd_outb_end always align on 8 bytes
    // boundary.
    avail = ((avail + 7) >> 3) << 3;

    omni::ptr_arith_t newmkr = (omni::ptr_arith_t) g->pd_outb_mkr + avail;

    // If the new position is inside the buffer, set the end pointer.
    // Note that if avail is very large, newmkr may wrap around and be
    // < pd_outb_mkr.

    if ((newmkr >= (omni::ptr_arith_t)g->pd_outb_mkr &&
	 newmkr <  (omni::ptr_arith_t)g->pd_outb_end)) {

      g->pd_outb_end = (void*) newmkr;
    }
  }
}

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

  // The caller has already checked that align == sz, or sz == 0.

  if (outputHasReachedLimit(g)) {
    // Already reached the message size limit
    goto overflow;
  }

  g->pd_outb_mkr = (void*) omni::align_to((omni::ptr_arith_t)g->pd_outb_mkr,
					  align);

  if (sz == 0)
    return;

  if ((omni::ptr_arith_t)g->pd_outb_mkr < (omni::ptr_arith_t)g->pd_outb_end) {
    omni::ptr_arith_t newmkr = ((omni::ptr_arith_t)g->pd_outb_mkr + sz); 
    if (newmkr <= (omni::ptr_arith_t)g->pd_outb_end)
      return;
    // Should never happen!!
    OMNIORB_ASSERT(0);
  }

  // Reach here only if the buffer has been filled up completely.

  outputFlush(g);
  if (!outputHasReachedLimit(g))
    return;

 overflow:
    OMNIORB_THROW(MARSHAL,MARSHAL_MessageSizeExceedLimitOnClient,
		  (CORBA::CompletionStatus)g->completion());
}

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

  if (outputHasReachedLimit(g)) {
    // Already reached the message size limit
    OMNIORB_THROW(MARSHAL,MARSHAL_MessageSizeExceedLimitOnClient,
		  (CORBA::CompletionStatus)g->completion());
  }

  omni::ptr_arith_t newmkr = omni::align_to((omni::ptr_arith_t)g->pd_outb_mkr,
					    align);
  OMNIORB_ASSERT(newmkr <= (omni::ptr_arith_t)g->pd_outb_end);

  g->pd_outb_mkr = (void*)newmkr;

  if (sz >= giopStream::directSendCutOff) {


    // The fragment including this vector of bytes must end on a 8 byte
    // boundary. Therefore we may have to leave behind 0-7 bytes in the
    // next fragment.

    size_t leftover = (newmkr + sz) & 0x7;

    if (!g->outputFragmentSize()) {

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

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

      *((CORBA::ULong*)((omni::ptr_arith_t)outbuf_begin + 8)) = fsz;
    }

    outputFlush(g,1);

    g->sendCopyChunk(b,sz - leftover);

    if (leftover) {
      if (outputHasReachedLimit(g)) {
	// Already reached the message size limit
	OMNIORB_THROW(MARSHAL,MARSHAL_MessageSizeExceedLimitOnClient,
		      (CORBA::CompletionStatus)g->completion());
      }
      b = (void*) ((omni::ptr_arith_t)b + sz - leftover);
    }
    sz = leftover;
  }

  while (sz) {
    size_t avail = (omni::ptr_arith_t) g->pd_outb_end	- 
      (omni::ptr_arith_t) g->pd_outb_mkr;
    if (avail > sz) avail = sz;
    memcpy(g->pd_outb_mkr,b,avail);
    sz -= avail;
    g->pd_outb_mkr = (void*)((omni::ptr_arith_t) g->pd_outb_mkr + avail);
    b = (void*)((omni::ptr_arith_t) b + avail);
    if (g->pd_outb_mkr == g->pd_outb_end) {
      outputFlush(g);
      if (sz && outputHasReachedLimit(g)) {
	// Already reached the message size limit
	OMNIORB_THROW(MARSHAL,MARSHAL_MessageSizeExceedLimitOnClient,
		      (CORBA::CompletionStatus)g->completion());
      }
    }
  }
}


////////////////////////////////////////////////////////////////////////
CORBA::ULong
giopImpl12::currentOutputPtr(const giopStream* g) {

  CORBA::ULong fsz = (omni::ptr_arith_t) g->pd_outb_mkr - 
                     ((omni::ptr_arith_t) g->pd_currentOutputBuffer + 
		      g->pd_currentOutputBuffer->start);

  if (g->outputMessageSize()) {
    return fsz + g->outputMessageSize();
  }
  else {
    return fsz - 12;
  }
}

////////////////////////////////////////////////////////////////////////
void
giopImpl12::outputSetFragmentSize(giopStream* g,CORBA::ULong msz) {

  if (msz > orbParameters::giopMaxMsgSize) {
    OMNIORB_THROW(MARSHAL,MARSHAL_MessageSizeExceedLimitOnClient,
		  (CORBA::CompletionStatus)g->completion());
  }
  g->outputFragmentSize(msz);
}

////////////////////////////////////////////////////////////////////////
CORBA::Boolean
giopImpl12::outputHasReachedLimit(giopStream* g) {

  return ( (omni::ptr_arith_t) g->pd_outb_end == 
          ((omni::ptr_arith_t) g->pd_currentOutputBuffer + 
		  	       g->pd_currentOutputBuffer->start + 12) );
}

////////////////////////////////////////////////////////////////////////
static giopStreamImpl* giop_1_2_singleton = 0;

class omni_giopImpl12_initialiser : public omniInitialiser {
public:

  void attach() {
    if (!giop_1_2_singleton) {
      GIOP::Version ver = { 1, 2};
      giopStreamImpl* p;
      giop_1_2_singleton = p = new giopStreamImpl(ver);
      
      // Shared by the client and server side
      // Process message header
      p->outputMessageBegin             = giopImpl12::outputMessageBegin;
      p->outputMessageEnd               = giopImpl12::outputMessageEnd;
      p->inputMessageBegin              = giopImpl12::inputMessageBegin;
      p->inputMessageEnd                = giopImpl12::inputMessageEnd;
      p->sendMsgErrorMessage            = giopImpl12::sendMsgErrorMessage;

      // Client side
      // Process message header
      p->marshalRequestHeader           = giopImpl12::marshalRequestHeader;
      p->sendLocateRequest              = giopImpl12::sendLocateRequest;
      p->unmarshalReplyHeader           = giopImpl12::unmarshalReplyHeader;
      p->unmarshalLocateReply           = giopImpl12::unmarshalLocateReply;

      // Server side
      // Process message header
      p->unmarshalWildCardRequestHeader = giopImpl12::unmarshalWildCardRequestHeader;
      p->unmarshalRequestHeader         = giopImpl12::unmarshalRequestHeader;
      p->unmarshalLocateRequest         = giopImpl12::unmarshalLocateRequest;
      p->marshalReplyHeader             = giopImpl12::marshalReplyHeader;
      p->sendSystemException            = giopImpl12::sendSystemException;
      p->sendUserException              = giopImpl12::sendUserException;
      p->sendLocationForwardReply       = giopImpl12::sendLocationForwardReply;
      p->sendLocateReply                = giopImpl12::sendLocateReply;


      // Shared by the client and the server side
      // Process message body
      p->inputRemaining                 = giopImpl12::inputRemaining;
      p->getInputData                   = giopImpl12::getInputData;
      p->skipInputData                  = giopImpl12::skipInputData;
      p->copyInputData                  = giopImpl12::copyInputData;
      p->outputRemaining                = giopImpl12::outputRemaining;
      p->getReserveSpace                = giopImpl12::getReserveSpace;
      p->copyOutputData                 = giopImpl12::copyOutputData;
      p->currentInputPtr                = giopImpl12::currentInputPtr;
      p->currentOutputPtr               = giopImpl12::currentOutputPtr;

      giopStreamImpl::registerImpl(giop_1_2_singleton);
    }
  }

  void detach() { 
    if (giop_1_2_singleton) {
      delete giop_1_2_singleton;
      giop_1_2_singleton = 0;
    }
  }
};

static omni_giopImpl12_initialiser initialiser;

omniInitialiser& omni_giopImpl12_initialiser_ = initialiser;

OMNI_NAMESPACE_END(omni)