/*
 *  ZThreads, a platform-independant, multithreading and 
 *  synchroniation library
 *
 *  Copyright (C) 2000-2002 Eric Crahen, See LGPL.TXT for details
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This library is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
 *
 *  SUNY @ Buffalo, hereby disclaims all copyright interest in the
 *  ZThreads library written by Eric Crahen
 */

#ifndef __ZTMONITOREDQUEUE_H__
#define __ZTMONITOREDQUEUE_H__

#include "zthread/ZThread.h"
#include "zthread/Condition.h"
#include "zthread/Guard.h"
#include "zthread/Mutex.h"
#include "zthread/Queue.h"
#include "zthread/Thread.h"

#include <deque>


namespace ZThread {

/**
 * @class MonitoredQueue
 *
 * @author Eric Crahen <crahen@cse.buffalo.edu>
 * @date <2002-02-24T09:26:45-0500>
 * @version 1.5.3
 *
 */

template <class T, class LOCK=Mutex>
class ZTHREAD_API MonitoredQueue : public Queue<T> {
private:

  MonitoredQueue(MonitoredQueue*);
  MonitoredQueue(const MonitoredQueue&);
  MonitoredQueue& operator=(const MonitoredQueue&);

protected:

  //! Serialize access
  LOCK _lock;
  //! Signaled on not empty
  Condition _notEmpty;
  //! Signaled on empty
  Condition _isEmpty;
  //! Backing deque
  std::deque<T> _queue;

  //! Cancellation flag
  bool _canceled;

public:

  /**
   * Create a new Queue
   */
  MonitoredQueue() throw() : _canceled(false) {
  }

  /**
   * Destroy this Queue
   */
  virtual ~MonitoredQueue() throw() {}

  /**
   * Insert a new item into the queue. One thread blocked by a next()
   * request will be woken up.
   * 
   * This implementation signals a condition variable used to block
   * threads waiting for items to be added to the queue
   *
   * @param T o - item to add
   *
   * @exception InterruptedException
   * @exception CancelationException - the queue has 
   * been canceled.
   *
   * @pre Objects add()ed should not be 'null'.
   */
  virtual void add(T item) 
    throw(Synchronization_Exception) {

    Guard<LOCK> g(_lock);
    
    // Allow no further additions in the canceled state
    if(_canceled)
      throw Cancelation_Exception();

    _queue.push_back( item );

    _notEmpty.signal(); // Wake one waiter


  }


  /**
   * Take an item from the queue, block calling thread if empty.
   *
   * Any threads blocked by an empty call are awoken if this 
   * call empties the queue
   *
   * @return T - next available object
   *
   * @exception InterruptedException
   * @exception CancelationException - the queue has 
   * been canceled.
   * @exception Timeout_Exception - thrown if the given amount
   * of time has expired and the queue could not return an item
   */
  virtual T next()
    throw(Synchronization_Exception) {

    Guard<LOCK> g(_lock);
      
    while ( _queue.size() == 0 && !_canceled) 
      _notEmpty.wait(_lock);
    
    if( _queue.size() == 0) // Queue canceled
      throw Cancelation_Exception();  
      
    T item = _queue.front();
    _queue.pop_front();

    if(_queue.size() == 0) // Wake empty waiters
      _isEmpty.broadcast();

    return item;

    
  }

  /**
   * Take an item from the queue, block if empty. 
   *
   * Any threads blocked by an empty call are awoken if this 
   * call empties the queue. The queue will try to acquire the 
   * lock it owns for up to timeout milliseconds, it will throw
   * a timeout only when that time has expired w/o obtaining a lock 
   *
   * @param unsigned long - max time
   *
   * @return T - next available object
   *
   * @exception InterruptedException
   * @exception CancelationException - the queue has 
   * been canceled.
   * @exception Timeout_Exception - thrown if the given amount
   * of time has expired and the queue could not return an item
   */
  virtual T next(unsigned long timeout)
    throw(Synchronization_Exception) {

    Guard<LOCK> g(_lock, timeout);
      
    while ( _queue.size() == 0 && !_canceled) {
      if(!_notEmpty.wait(_lock, timeout))
        throw Timeout_Exception();
    }

    if( _queue.size() == 0) // Queue canceled
      throw Cancelation_Exception();  


    T item = _queue.front();
    _queue.pop_front();

    if(_queue.size() == 0) // Wake empty waiters
      _isEmpty.broadcast();


    return item;

  }

  /**
   * Block calling until the queue becomes empty. This implementation
   * should always return true.
   * 
   * This implementation waits on a condition variable that
   * is broadcast to each time the queue removed its last
   * element. 
   * This can be a periodic occurrence; you may have a faster consumer
   * than producer, for example. In this case, the queue might be 
   * emptied several times before it is canceled.
   *
   * @return boolean true if empty, false if not.
   * @exception InterruptedException
   */
  bool empty() 
    throw(Synchronization_Exception) {


    Guard<LOCK> g(_lock);

    while(_queue.size() > 0) // Wait for an empty signal
      _isEmpty.wait(_lock);
    
    return true;


  }

  /**
   * Cancel this queue, no more additions will be accepted.
   *
   * @post - next() will no longer block when he queue is empty 
   *
   * @exception InterruptedException
   */
  virtual void cancel() 
    throw(Synchronization_Exception) {

    Guard<LOCK> g(_lock);

    _canceled = true;
    _notEmpty.broadcast(); // Wake next() waiters

  }


  /**
   * Check the cancelation status of this Queue
   * 
   * @return bool - Queue cancled
   */
  inline virtual bool isCanceled()
    throw(Synchronization_Exception) {

    // Faster check since the queue will not become un-canceled
    if(_canceled)
      return true;
    
    Guard<LOCK> g(_lock);

    return _canceled;

  }

};


};


#endif

