Synch_Queued_Message.cpp

Go to the documentation of this file.

// "Synch_Queued_Message.cpp,v 1.20 2006/05/16 12:29:30 jwillemsen Exp"

#include "tao/Synch_Queued_Message.h"
#include "tao/debug.h"
#include "tao/ORB_Core.h"

#include "ace/Malloc_T.h"
#include "ace/Message_Block.h"

ACE_RCSID (tao,
           Synch_Queued_Message,
           "Synch_Queued_Message.cpp,v 1.20 2006/05/16 12:29:30 jwillemsen Exp")

TAO_BEGIN_VERSIONED_NAMESPACE_DECL

TAO_Synch_Queued_Message::TAO_Synch_Queued_Message (
  const ACE_Message_Block *contents,
  TAO_ORB_Core *oc,
  ACE_Allocator *alloc,
  bool is_heap_allocated)
  : TAO_Queued_Message (oc, alloc, is_heap_allocated)
  , contents_ (const_cast<ACE_Message_Block*> (contents))
  , current_block_ (contents_)
{
}

TAO_Synch_Queued_Message::~TAO_Synch_Queued_Message (void)
{

}

const ACE_Message_Block *
TAO_Synch_Queued_Message::current_block (void) const
{
  return this->current_block_;
}

size_t
TAO_Synch_Queued_Message::message_length (void) const
{
  if (this->current_block_ == 0)
    {
      return 0;
    }

  return this->current_block_->total_length ();
}

int
TAO_Synch_Queued_Message::all_data_sent (void) const
{
  return this->current_block_ == 0;
}

void
TAO_Synch_Queued_Message::fill_iov (int iovcnt_max,
                                    int &iovcnt,
                                    iovec iov[]) const
{
  ACE_ASSERT (iovcnt_max > iovcnt);

  for (const ACE_Message_Block *message_block = this->current_block_;
       message_block != 0 && iovcnt < iovcnt_max;
       message_block = message_block->cont ())
    {
      size_t const message_block_length = message_block->length ();

      // Check if this block has any data to be sent.
      if (message_block_length > 0)
        {
              // Collect the data in the iovec.
          iov[iovcnt].iov_base = message_block->rd_ptr ();
          iov[iovcnt].iov_len  = static_cast<u_long> (message_block_length);

          // Increment iovec counter.
          ++iovcnt;
        }
    }
}

void
TAO_Synch_Queued_Message::bytes_transferred (size_t &byte_count)
{
  this->state_changed_i (TAO_LF_Event::LFS_ACTIVE);

  while (this->current_block_ != 0 && byte_count > 0)
    {
      size_t const l = this->current_block_->length ();

      if (byte_count < l)
        {
          this->current_block_->rd_ptr (byte_count);
          byte_count = 0;
          return;
        }

      byte_count -= l;
      this->current_block_->rd_ptr (l);
      this->current_block_ = this->current_block_->cont ();

      while (this->current_block_ != 0
             && this->current_block_->length () == 0)
        {
          this->current_block_ = this->current_block_->cont ();
        }
    }

  if (this->current_block_ == 0)
    this->state_changed (TAO_LF_Event::LFS_SUCCESS,
                         this->orb_core_->leader_follower ());
}

TAO_Queued_Message *
TAO_Synch_Queued_Message::clone (ACE_Allocator *alloc)
{
  TAO_Synch_Queued_Message *qm = 0;

  // Clone the message block.
  // NOTE: We wantedly do the cloning from <current_block_> instead of
  // starting from <contents_> since we dont want to clone blocks that
  // have already been sent on the wire. Waste of memory and
  // associated copying.
  ACE_Message_Block *mb = this->current_block_->clone ();

  if (alloc)
    {
      ACE_NEW_MALLOC_RETURN (qm,
                             static_cast<TAO_Synch_Queued_Message *> (
                               alloc->malloc (sizeof (TAO_Synch_Queued_Message))),
                             TAO_Synch_Queued_Message (mb,
                                                       this->orb_core_,
                                                       alloc),
                             0);
    }
  else
    {
      // No allocator, so use the common heap!
      if (TAO_debug_level == 4)
        {
          // This debug is for testing purposes!
          ACE_DEBUG ((LM_DEBUG,
                      "TAO (%P|%t) - Synch_Queued_Message::clone\n"
                      "Using global pool for allocation \n"));
        }

      ACE_NEW_RETURN (qm,
                      TAO_Synch_Queued_Message (mb, this->orb_core_),
                      0);
    }

  // Set the flag to indicate that <qm> is created on the heap.
  if (qm)
    {
      qm->is_heap_created_ = true;
    }

  return qm;
}

void
TAO_Synch_Queued_Message::destroy (void)
{
  if (this->is_heap_created_)
    {
      ACE_Message_Block::release (this->contents_);
      this->current_block_ = 0;

      // If we have an allocator release the memory to the allocator
      // pool.
      if (this->allocator_)
        {
          ACE_DES_FREE (this,
                        this->allocator_->free,
                        TAO_Synch_Queued_Message);

        }
      else // global release..
        {
          delete this;
        }
    }
}

TAO_END_VERSIONED_NAMESPACE_DECL

---