/*
 *  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 __ZTBUFFEREDQUEUE_H__
#define __ZTBUFFEREDQUEUE_H__

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

#include <deque>

namespace ZThread {

/**
 * @class BufferedQueue
 *
 * @author Eric Crahen <crahen@cse.buffalo.edu>
 * @date <2002-02-24T09:22:25-0500>
 * @version 1.3.4
 *
 * BufferedQueue's are the most heavyweight queue included in this
 * package. This queue will block threads adding too many items, 
 * withdrawing items from it in th empty state, and threads waiting
 * for the queue to become empty.
 */

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

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

  unsigned int _capacity;


  protected:

  //! Serialize access
  LOCK _lock;
  //! Signaled if not full
  Condition _notFull;
  //! Signaled if not empty
  Condition _notEmpty;
  //! Signaled if empty
  Condition _isEmpty;
  //! Backing deque
  std::deque<T> _queue;
  //! Cancellation flag
  bool _canceled;


  public:

  /**
   * Create a new queue with the given capacity. This 
   * capacity can not be altered once the queue is created
   *
   * @param int capacity
   */
  BufferedQueue(unsigned int capacity) { 
    _capacity = capacity;
    _canceled = false;
  }

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

  /**
   * Get the capacity of this queue.
   *
   * @return unsigned int capacity
   */
  unsigned int capacity() { 
    return _capacity; 
  }


  /**
   * Insert a new item into the queue. 
   * 
   * The calling thread will be blocked if the queue has reached 
   * its capacity. Execution of that thread will continue after 
   * items are removed and the queue is not filled to capacity any longer.
   * 
   * 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);
      
    // Wait for the capacity of the queue to drop 
    while ((_queue.size() == _capacity) && !_canceled)
      _notFull.wait(_lock);
      
    if(_canceled)
      throw Cancelation_Exception();

    _queue.push_back(item);
    _notEmpty.signal(); // Wake any waiters
      
    
  }
    

  /**
   * Take an item from the queue, block if empty. If the
   * given time expires
   *
   * @return T - next available object
   *
   * @exception InterruptedException
   * @exception CancelationException - the queue has 
   * been canceled.
   */ 
  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();
      
    _notFull.signal(); // Wake any thread trying to add

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

    return item;

  }

  /**
   * Take an item from the queue, block if empty. If the
   * given time expires
   *
   * @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);
    
    // Wait for items to be added
    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();

    _notFull.signal(); // Wake add() waiters

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


  
  /**
   * Cancel this queue, no more additions will be accepted.
   *
   * @post - next() will no longer block when he queue is empty
   * it will return 'null'. Thread waiting to add will wake and 
   * throw CancelationException
   *
   * @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;

  }

  /**
   * Test the queue to determine if it has been emptied. 
   * The calling thread is blocked until the queue empties
   *
   * A Queue can reach empty many times during its lifetime
   * it is not a one time occurrence. However, after a queue has
   * been canceled it will reach empty only once more
   *
   * @exception InterruptedException
   */
  virtual bool empty() 
    throw(Synchronization_Exception) {

    Guard<LOCK> g(_lock);    

    while(!_canceled && !_queue.size() == 0)
      _isEmpty.wait(_lock);
    
    return !_canceled;

      
  }

};


};

#endif
