// Batch_Buffering_Strategy.cpp,v 1.11 2005/01/27 19:44:47 jtc Exp

#include "Batch_Buffering_Strategy.h"
#include "../Method_Request_Event.h"
#include "ace/Null_Condition.h"

ACE_RCSID (Notify, TAO_Notify_Batch_Buffering_Strategy, "Batch_Buffering_Strategy.cpp,v 1.11 2005/01/27 19:44:47 jtc Exp")

TAO_Notify_Batch_Buffering_Strategy::TAO_Notify_Batch_Buffering_Strategy (TAO_Notify_Message_Queue& msg_queue, TAO_Notify_AdminProperties_var& admin_properties, CORBA::Long batch_size)
  :TAO_Notify_Buffering_Strategy (msg_queue, admin_properties, batch_size)
{
}

TAO_Notify_Batch_Buffering_Strategy::~TAO_Notify_Batch_Buffering_Strategy ()
{
}

int
TAO_Notify_Batch_Buffering_Strategy::dequeue_batch (CosNotification::EventBatch& event_batch)
{
  ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->global_queue_lock_, -1);

  // if batch_size is infinite, simply dequeue everything available.

  int pending = 0; // not used.

  if (this->batch_size_ == 0)
    {
      return this->dequeue_available (event_batch, pending);
    }
  else
    {
      // block till batch size of events are available.
      while (this->msg_queue_.message_count () < this->batch_size_)
        {
          if (this->shutdown_ == 1) // if we're shutdown, don't play this silly game.
            return -1;

          this->batch_size_reached_condition_.wait ();
        }

      return this->dequeue_i (this->batch_size_, event_batch);
    }
}

int
TAO_Notify_Batch_Buffering_Strategy::dequeue_available (CosNotification::EventBatch& event_batch, int &pending)
{
  ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->global_queue_lock_, -1);

  int deq_count = this->msg_queue_.message_count ();

  if (this->batch_size_ != 0 && deq_count > this->batch_size_) // Restrict upto batch size.
    deq_count = this->batch_size_;

  pending = this->msg_queue_.message_count () - deq_count;

  return this->dequeue_i (deq_count, event_batch);
}

int
TAO_Notify_Batch_Buffering_Strategy::dequeue_i (int max_deq_count, CosNotification::EventBatch& event_batch)
{
  ACE_Message_Block *mb;

  int deq_count = 0;

  event_batch.length (max_deq_count);

  for (; deq_count < max_deq_count; ++deq_count)
    {
      if (this->msg_queue_.dequeue (mb) == -1)
        break; // error, simply return what we could extract so far.

      --this->global_queue_length_;

      TAO_Notify_Method_Request_Event_Queueable* mre = dynamic_cast<TAO_Notify_Method_Request_Event_Queueable*> (mb);

      mre->event ()->convert (event_batch[deq_count]);

      ACE_Message_Block::release (mb);
    }

  event_batch.length (deq_count);

  this->global_queue_not_full_condition_.signal ();
  this->local_queue_not_full_condition_.signal ();

  return deq_count;
}
