/* Network-Integrated Multimedia Middleware (NMM)
 * http://www.motama.com/nmm.html
 *  
 * Copyright (C) 2006-2007
 * Motama GmbH, Saarbruecken, Germany
 * http://www.motama.com
 *  
 * This file is part of NMM. 
 * 
 * NMM is free software; you can redistribute it and/or modify it    
 * under the terms of the GNU General Public License as published by 
 * the Free Software Foundation; only version 2 of the License.      
 * 
 * NMM 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 General Public 
 * License for more details.  
 * 
 * You should have received a copy of the GNU General Public License 
 * along with NMM; if not, write to the Free Software Foundation,    
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 */

#ifndef NMM_STREAM_QUEUE_HPP
#define NMM_STREAM_QUEUE_HPP NMM_STREAM_QUEUE_HPP

#include "nmm/nmm_prefix.hpp"

#include "Signal.hpp"
#include "Timestamp.hpp"

#include "nmm/Result.hpp"
#include "nmm/utils/NamedObject.hpp"
#include "nmm/utils/thread/ThreadMutex.hpp"
#include "nmm/utils/thread/ThreadCondition.hpp"

#include "MessageDuration.hpp"

#include <queue>

#include "visibility.hpp"

namespace NMM {
  
  class Message;
  class ISignalReceiver;
  
  /** This class represents an input queue.
   *
   * A QueuedThreadedNode has a Queue at the input jack.  The queue
   * may be configured to behave in a specific way if a buffer arrives
   * when the queue is full: The node may chose to block its
   * predecessor's thread, discard the newly incoming buffer in favour
   * of the older ones in the queue, or discard the oldest waiting
   * buffer in the queue in favour of the newly entered one.
   */
  class DLLEXPORT StreamQueue : public NamedObject {
  public:
    /** Specifies what to do when the queue is full and a new buffer
	arrives.                                                            */
    enum Mode { DEFAULT_MODE = 1,     /*!< Default mode, uses  */
		MODE_SUSPEND = 1,     /*!< Wait if the queue is full.   */
		MODE_KEEP_OLDEST, /*!< Keep oldest buffer if the
				    queue is full.                       */
		MODE_KEEP_NEWEST  /*!< Keep newest buffer if the
				    queue is full.                       */
    };

    /** The main constructor.
     * 
     * @param mode QueueMode of the new queue.
     * @param size Size of the new queue. Default is 16.
     * @param name NamedObject name
     * @param free_queue_slot_sig enumeration value for free queue slot signal
     * @param queue_not_empty_sig enumeration value for queue not empty
     */
    StreamQueue(const Mode mode = DEFAULT_MODE,
		const unsigned int size = DEFAULT_SIZE,
		const char* name = 0,
		const Signal free_queue_slot_sig = FREE_QUEUE_SLOT,
		const Signal queue_not_empty_sig = QUEUE_NOT_EMPTY);

    /** Virtual destructor.
     */
    virtual ~StreamQueue();

    /** Receives message.
     *
     * This method is called by the node's receiveMessage method.
     *
     * @param buffer message to be received
     * @return BLOCKED if the queue is full now. SUCCESS if the buffer is handled according to the QueueMode.
     */
    Result receiveMessage(Message* buffer);

    /** Returns next buffer.
     *
     * This function returns the next buffer or null if queue is empty. 
     *
     * @return pointer to next buffer
     */
    Message* dequeueMessage();

    /** Get the current size of this queue.
     * This method returns the number of buffers currently queued.
     *
     * @return Current size of the queue (in number of buffers)
     */
    unsigned int getSize() const;

    /** Get the maximum size of this queue.
     * This method returns the maximum number of buffers that may be queued.
     *
     * @return Maximum size of the queue (in number of buffers)
     */
    unsigned int getMaxSize() const;

    /** Set the maximum size of this queue.
     * This method sets the maximum number of buffers that may be queued.
     *
     * @param size Maximum size of the queue (in number of buffers)
     */
    Result setMaxSize(unsigned int size);

    /** Set the maximum duration of the queue.
     * 
     * @param duration the maximum duration of the queue.
     * @return SUCCESS
     */
    Result setMaxDuration(Interval duration);

    /** Sets the prefetch queue size.
     *
     * Sets the prefetch queue size and checks that is not longer than maximum queue size.
     *
     * @param size prefetch queue size 
     */
    void setPrefetchQueueSize(unsigned int size);

    /** Sets the prefetch queue duration.
     * 
     * Sets the prefetch queue duration and checks that is not longer than maximum queue duration.
     *
     * @param duration prefetch queue duration
     */
    void setPrefetchQueueDuration(const Interval& duration);

    /** Wait until prefetch is finished.
     *
     * Waits until prefetch flag is false.
     */
    void waitPrefetchFinished();

    /** Sets the internal max. queue length to 'infinite'.
     */
    void unlimitSize();

    /** Re-sets the internal max. queue length to the value specified by setMaxSize.
     */
    void limitSize();

    /** Get the mode of this queue.
     * This method returns the queue mode.
     *
     * @return Queue mode
     */
    Mode getMode() const;

    /** Set the mode of this queue.
     * This method sets the queue mode.
     *
     * @param mode Queue mode
     */
    Result setMode(Mode mode);
    
    /** Set parent signal receiver.
     *
     * Set @a parent as the signal receiver which is notified when the queue is not empty.
     *
     * @param parent parent signal receiver
     * @return SUCCESS
     */
    Result setParentSignalReceiver(ISignalReceiver* parent);
    
    /** Set partner signal receiver.
     *
     * Set @a partner as the signal receiver which is notified when there is a free slot in the queue.
     *
     * @param partner partner signal receiver
     * @return BLOCKED if queue is over full, SUCCESS otherwise
     */
    Result setPartnerSignalReceiver(ISignalReceiver* partner);
    
    /** Set parent signal receiver.
     *
     * Set @a parent as the signal receiver which is notified when the queue is not empty.
     * Also set @a queue_not_empty_sig as the signal which is sent.
     *
     * @param parent parent signal receiver
     * @param queue_not_empty_sig queue not empty Signal enumeration value
     * @return SUCCESS
     */
    Result setParentSignalReceiver(ISignalReceiver* parent, const Signal queue_not_empty_sig);
    
    /** Set parent signal receiver.
     *
     * Set @a parent as the signal receiver which is notified when the queue is not empty.
     * Also set @a free_queue_slot_sig as the signal which is sent.
     *
     * @param partner partner signal receiver
     * @param free_queue_slot_sig free queue slot Signal enumeration value
     * @return BLOCKED if queue is full or overfull, SUCCESS otherwise.
     */
    Result setPartnerSignalReceiver(ISignalReceiver* partner, const Signal free_queue_slot_sig);
    
    /** Lock the internal mutex of the queue.
     */
    void lock() const;

    /** Unlock the internal mutex of the queue.
     */
    void unlock() const;
    
    /** Removes all non-mandatory messages not newer than given timestamp. 
     *   
     * @param flush_ts Timestamp to compare Message timestamps with
     */
    Result flush(const Timestamp& flush_ts = Timestamp());
    
    static const unsigned int DEFAULT_SIZE = 16; ///< The default size of a queue.   

  private:
    /** Check if queue is full (using size or duration).
     *
     * @return true if queue contains maximum queue size or more elements, false otherwise
     */
    bool isQueueFull() const;

    /** Check if queue is more than full (using size or duration).
     * 
     * @return true if queue contains more than maximum queue size elements, false otherwise
     */
    bool isQueueMoreThanFull() const;

    /** Check if prefetching is finished (using size or duration).
     *
     * @return false if prefetching is running currently, true otherwise 
     */
    bool isPrefetchingFinished() const;

    /** Removes all non-mandatory messages not newer than given timestamp. 
     * 
     * @param flush_ts Timestamp to compare Message timestamps with
     */
    Result internalFlush(const Timestamp& flush_ts = Timestamp());

    std::queue<Message*> message_queue;   ///< stores incoming messages

    Mode queue_mode; ///< Behaviour on full queue (see enum QueueMode).  

    bool use_duration;               ///< If true, use max_queue_duration, otherwise use max_queue_size
    unsigned int max_queue_size;     ///< Length of the queue array.
    Interval max_queue_duration;     ///< Duration of the queue array.

    ThreadMutex queue_mutex;         ///< Mutex for queue access.

    /**
     * This counter specifies how often we sent out an
     * AbortReceiveBufferException on a blocking enqueue. Only send this
     * many free slot events to avoid the premature end of a virtual stop.
     */
    int abort_receive_exception_count;


    ISignalReceiver* parent_of_jack; ///< Here, a ISignalReceiver-pointer to the parent of this stream queue's jack will be stored

    ISignalReceiver* parent_of_partner_jack; ///< Here, a ISignalReceiver-pointer to the parent of this stream queue's partner jack will be stored

    Signal sig_free_queue_slot; ///< Signal to be sent for FREE_QUEUE_SLOT

    Signal sig_queue_not_empty;  ///< Signal to be sent for QUEUE_NOT_EMPTY

    unsigned int limited_size; ///< Specifies whether queue size is limited or not, true per default
 
    unsigned int pre_fetch_queue_size; ///< Specifies the size of messages that must queued if pre-fetch modus is enabled, 0 per default

    Interval pre_fetch_queue_duration; ///< Specifies the duration  of messages that must queued if pre-fetch modus is enabled, 0 per default

    bool pre_fetch_flag; ///< Specifies whether the pre-fetch modus is currently enabled or not, false per default

    bool initial_pre_fetch_flag; ///< Specifies whether the pre-fetch modus is initially enabled or not, false per default

    ThreadMutex pre_fetch_mutex; ///< Mutex for pre_fetch_condition attribute

    ThreadCondition pre_fetch_condition; ///< Condition variable used to wait until pre-fetching is finished 

    MessageDuration in_message_duration;  ///< helps compute duration of incoming messages
    MessageDuration out_message_duration; ///< helps compute duration of outgoing messages
    Interval message_queue_duration;      ///< duration of message queue
  };
  
}

#include "nmm/nmm_suffix.hpp"

#endif
