// -*- C++ -*-
// Sync_Strategies.cpp,v 1.11 2000/01/12 03:39:41 irfan Exp

#include "tao/Sync_Strategies.h"

#include "tao/Buffering_Constraint_Policy.h"
#include "tao/Stub.h"
#include "tao/ORB_Core.h"

#if !defined (__ACE_INLINE__)
# include "tao/Sync_Strategies.i"
#endif /* ! __ACE_INLINE__ */

TAO_Sync_Strategy::~TAO_Sync_Strategy (void)
{
}

ssize_t
TAO_Transport_Sync_Strategy::send (TAO_Transport &transport,
                                   TAO_Stub &,
                                   const ACE_Message_Block *message_block,
                                   const ACE_Time_Value *max_wait_time)
{
  // Immediate delegation to the transport.
  return transport.send (message_block,
                         max_wait_time);
}

#if (TAO_HAS_CORBA_MESSAGING == 1)

ssize_t
TAO_Delayed_Buffering_Sync_Strategy::send (TAO_Transport &transport,
                                           TAO_Stub &stub,
                                           const ACE_Message_Block *mb,
                                           const ACE_Time_Value *max_wait_time)
{
  ACE_Message_Block *message_block =
    ACE_const_cast (ACE_Message_Block *, mb);

  ssize_t result = 0;

  // Get the message queue from the transport.
  TAO_Transport_Buffering_Queue &buffering_queue =
    transport.buffering_queue ();

  // Check if there are messages already in the queue.
  if (!buffering_queue.is_empty ())
    return TAO_Eager_Buffering_Sync_Strategy::send (transport,
                                                  stub,
                                                  message_block,
                                                  max_wait_time);

  //
  // Otherwise there were no queued messages.  We first try to send
  // the message right away.
  //

  // Actual network send.
  result = transport.send (message_block,
                           max_wait_time);

  // Cannot send.
  if (result == -1 ||
      result == 0)
    {
      // Timeout.
      if (errno == ETIME)
        {
          // Queue message.
          return TAO_Eager_Buffering_Sync_Strategy::send (transport,
                                                        stub,
                                                        message_block,
                                                        max_wait_time);
        }

      // Non-timeout error.
      return -1;
    }

  // If successful in sending some or all of the data, reset the
  // message block appropriately.
  transport.reset_sent_message (message_block,
                                result);

  // If there is still data left over, i.e., incomplete send, queue
  // the rest.
  if (message_block->total_length () != 0)
    {
      return result +
        TAO_Eager_Buffering_Sync_Strategy::send (transport,
                                               stub,
                                               message_block,
                                               max_wait_time);
    }

  // Everything was successfully delivered.
  return result;
}

ssize_t
TAO_Eager_Buffering_Sync_Strategy::send (TAO_Transport &transport,
                                       TAO_Stub &stub,
                                       const ACE_Message_Block *message_block,
                                       const ACE_Time_Value *max_wait_time)
{
  ssize_t result = 0;

  // Get the message queue from the transport.
  TAO_Transport_Buffering_Queue &buffering_queue =
    transport.buffering_queue ();

  // Copy the message.
  ACE_Message_Block *copy = message_block->clone ();

  // Enqueue current message.
  result = buffering_queue.enqueue_tail (copy);

  // EnBuffering error.
  if (result == -1)
    {
      // Eliminate the copy.
      copy->release ();

      // Return error.
      return -1;
    }

  // Check if upper bound has been reached.
  if (this->buffering_constraints_reached (transport,
                                           stub,
                                           buffering_queue))
    {
      return transport.send_buffered_messages (max_wait_time);
    }

  // Hoping that this return value is meaningful or at least
  // acceptable.
  return message_block->total_length ();
}

int
TAO_Eager_Buffering_Sync_Strategy::buffering_constraints_reached (TAO_Transport &transport,
                                                                TAO_Stub &stub,
                                                                TAO_Transport_Buffering_Queue &buffering_queue)
{
  TAO_Buffering_Constraint_Policy *buffering_constraint_policy =
    stub.buffering_constraint ();

  if (buffering_constraint_policy == 0)
    return 1;

  TAO::BufferingConstraint buffering_constraint =
    buffering_constraint_policy->buffering_constraint ();

  this->timer_check (transport,
                     buffering_constraint);

  if (buffering_constraint.mode == TAO::BUFFER_FLUSH)
    return 1;

  if (ACE_BIT_ENABLED (buffering_constraint.mode,
                       TAO::BUFFER_MESSAGE_COUNT) &&
      buffering_queue.message_count () >= buffering_constraint.message_count)
    return 1;

  if (ACE_BIT_ENABLED (buffering_constraint.mode,
                       TAO::BUFFER_MESSAGE_BYTES) &&
      buffering_queue.message_length () >= buffering_constraint.message_bytes)
    return 1;

  return 0;
}

void
TAO_Eager_Buffering_Sync_Strategy::timer_check (TAO_Transport &transport,
                                              const TAO::BufferingConstraint &buffering_constraint)
{
  if (transport.buffering_timer_id () != 0)
    {
      //
      // There is a timeout set by us, though we are not sure if we
      // still need the timeout or if the timeout value is correct or
      // not.
      //

      // Get our reactor.
      ACE_Reactor *reactor = transport.orb_core ()->reactor ();

      if (!ACE_BIT_ENABLED (buffering_constraint.mode,
                            TAO::BUFFER_TIMEOUT))
        {
          // Timeouts are no longer needed.  Cancel existing one.
          reactor->cancel_timer (transport.buffering_timer_id ());
          transport.buffering_timer_id (0);
        }
      else
        {
          ACE_Time_Value timeout =
            this->time_conversion (buffering_constraint.timeout);

          if (transport.buffering_timeout_value () == timeout)
            {
              // Timeout value is the same, nothing to be done.
            }
          else
            {
              // Timeout value has changed, reset the old timer.
              reactor->reset_timer_interval (transport.buffering_timer_id (),
                                             timeout);
            }
        }
    }
  else if (ACE_BIT_ENABLED (buffering_constraint.mode,
                            TAO::BUFFER_TIMEOUT))
    {
      // We didn't have timeouts before, but we want them now.
      ACE_Time_Value timeout =
        this->time_conversion (buffering_constraint.timeout);

      // Get our reactor.
      ACE_Reactor *reactor = transport.orb_core ()->reactor ();

      long timer_id = reactor->schedule_timer (transport.event_handler (),
                                               0,
                                               timeout,
                                               timeout);

      transport.buffering_timer_id (timer_id);
      transport.buffering_timeout_value (timeout);
    }
}

ACE_Time_Value
TAO_Eager_Buffering_Sync_Strategy::time_conversion (const TimeBase::TimeT &time)
{
  TimeBase::TimeT seconds = time / 10000000u;
  TimeBase::TimeT microseconds = (time % 10000000u) / 10;
  return ACE_Time_Value (ACE_U64_TO_U32 (seconds),
                         ACE_U64_TO_U32 (microseconds));
}

#endif /* TAO_HAS_CORBA_MESSAGING == 1 */
