/*
 * @(#)QueueThread.java      0.9.0 06/04/2000 - 13:31:52
 *
 * Copyright (C) 2000,,2003 2002 Matt Albrecht
 * groboclown@users.sourceforge.net
 * http://groboutils.sourceforge.net
 *
 *  Permission is hereby granted, free of charge, to any person obtaining a
 *  copy of this software and associated documentation files (the "Software"),
 *  to deal in the Software without restriction, including without limitation
 *  the rights to use, copy, modify, merge, publish, distribute, sublicense,
 *  and/or sell copies of the Software, and to permit persons to whom the 
 *  Software is furnished to do so, subject to the following conditions:
 *
 *  The above copyright notice and this permission notice shall be included in 
 *  all copies or substantial portions of the Software. 
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 
 *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 
 *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL 
 *  THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 
 *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 
 *  FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER 
 *  DEALINGS IN THE SOFTWARE.
 */

package net.sourceforge.groboutils.util.thread.v1;


import net.sourceforge.groboutils.util.datastruct.v1.SynchQueue;


/**
 * For threads which endlessly process events from a SynchQueue.  This
 * is a common technique for thread pooling.
 * <P>
 * Users must make a implementation of <tt>IObjectListener</tt>.
 * If the user does not give a <tt>SynchQueue</tt>, it is created for them.
 * Once the QueueThread is created, the queue and the listener cannot
 * be changed.
 * <P>
 * By default, the underlying {@link LoopThread} does not sleep between events,
 * it is a daemon thread, runs on the lowest thread priority, and has
 * not started yet.  Of course, this can all be changed.
 * <P>
 * It is advisable not to use the methods {@link
 * LoopThread#setSleepTime( int )} or
 * {@link LoopThread#setSleepTimeMillis( long )}, since these will cause the
 * <tt>QueueThread</tt> to not respond to incoming requests during this sleep
 * time.  However, there may be occations when this is necessary, so this is
 * left available to the user.
 * <P>
 * Since the {@link SynchQueue#dequeue()} method can wait indefinitely, you may
 * opt to set the dequeue's timeout to something other than 0.  Without setting
 * this, the thread may wait indefinitely for an incoming element to the queue
 * without ever checking for a {@link LoopThread#stop()} or
 * {@link LoopThread#suspend()} signal.  Thanks to
 * <a href="mailto:d.gallot@atoseuronext.be">Dominique Gallot</a> for pointing
 * this out.
 * <P>
 * After a {@link LoopThread#stop()} is invoked, you may allow the
 * object listener to finish processing the remaining elements in the
 * inner queue by calling {@link #processRemaining()}.
 *
 * @author   Matt Albrecht <a href="mailto:groboclown@users.sourceforge.net">groboclown@users.sourceforge.net</a>
 * @since    June 4, 2000
 * @version  $Date: 2003/02/10 22:52:48 $
 */
public class QueueThread extends LoopThread
{
    private IObjectListener m_objListener = null;
    
    private SynchQueue m_queue = null;
    
    private boolean m_isProcessingObject = false;
    
    private long m_timeout = 0;
    private int m_nanos = 0;
    

    /**
     * The runnable class - keep it private so no one
     * can directly access it in a thread but us.
     */
    private class QueueRunnable implements Runnable
    {
        public void run()
        {
            if (m_queue.isEmpty())
            {
                m_isProcessingObject = false;
            }

            try
            {
                Object o = m_queue.dequeue( m_timeout, m_nanos );
                m_isProcessingObject = true;
                m_objListener.processObject( o );
    
                if (m_queue.isEmpty())
                {
                    m_isProcessingObject = false;
                }
            }
            catch (InterruptedException ie)
            {
                // whoops - how do we handle this?
            }
        }
    }
    
    
    
    
    //--------------------------------------------------------------
    // Constructors
    
    
    /**
     *
     */
    public QueueThread( IObjectListener ol )
    {
        this( ol, new SynchQueue() );
    }
    
    /**
     *
     */
    public QueueThread( IObjectListener ol, SynchQueue sq )
    {
        super();
        
        initialize( ol, sq );
    }
    
    /**
     *  
     */
    public QueueThread( IObjectListener ol, ThreadGroup tg )
    {
        this( ol, new SynchQueue(), tg );
    }
    
    
    /**
     *  
     */
    public QueueThread( IObjectListener ol, SynchQueue sq, ThreadGroup tg )
    {
        super( null, tg );
        
        initialize( ol, sq );
    }
    
    
    /**
     *  
     */
    public QueueThread( IObjectListener ol, String threadName )
    {
        this( ol, new SynchQueue(), threadName );
    }
    
    
    /**
     *  
     */
    public QueueThread( IObjectListener ol, SynchQueue sq, String threadName )
    {
        super( null, threadName );
        
        initialize( ol, sq );
    }
    
    
    /**
     *  
     */
    public QueueThread( IObjectListener ol, ThreadGroup tg, String threadName )
    {
        this( ol, new SynchQueue(), tg, threadName );
    }
    
    
    /**
     *  
     */
    public QueueThread( IObjectListener ol, SynchQueue sq, ThreadGroup tg,
            String threadName )
    {
        super( null, tg, threadName );
        
        initialize( ol, sq );
    }
    
    
    
    
    //--------------------------------------------------------------
    // Public Methods
    
    
    /**
     * Retrieves the internal listened queue.
     */
    public SynchQueue getQueue()
    {
        return this.m_queue;
    }
    
    
    /**
     * @return <tt>false</tt> if the thread is waiting for an object
     *      to be placed in the queue and be processed, otherwise
     *      <tt>true</tt>.
     */
    public boolean isProcessingObjects()
    {
        return this.m_isProcessingObject;
    }
    
    
    /**
     * Set the maximum time (in milliseconds and nanoseconds) to wait for
     * an incoming element on the inner queue before checking for
     * {@link LoopThread#stop()} and {@link LoopThread#suspend()}
     * signals.
     *
     * @param timeout the maximum time to wait in milliseconds.
     * @param nanos additional time, in nanoseconds range 0-999999.
     * @see SynchQueue#dequeue( long, int )
     */
    public void setTimeout( long timeout, int nanos )
    {
        this.m_timeout = timeout;
        this.m_nanos = nanos;
    }
    
    
    /**
     * Set the maximum time (in milliseconds) to wait for
     * an incoming element on the inner queue before checking for
     * {@link LoopThread#stop()} and {@link LoopThread#suspend()}
     * signals.
     *
     * @param timeout the maximum time to wait in milliseconds.
     * @param nanos additional time, in nanoseconds range 0-999999.
     * @see SynchQueue#dequeue( long, int )
     * @see #setTimeout( long, int )
     */
    public void setTimeout( long timeout )
    {
        setTimeout( timeout, 0 );
    }
    
    
    /**
     * Retrieve the millisecond part of the maximum timeout to wait for an
     * incoming element on the inner queue before checking for thread
     * event signals.
     *
     * @see #setTimeout( long, int )
     */
    public long getTimeoutMilliseconds()
    {
        return this.m_timeout;
    }
    
    
    /**
     * Retrieve the nanosecond part of the maximum timeout to wait for an
     * incoming element on the inner queue before checking for thread
     * event signals.
     *
     * @see #setTimeout( long, int )
     */
    public int getTimeoutNanoseconds()
    {
        return this.m_nanos;
    }
    
    
    /**
     * Process all elements in the queue until the queue is empty.
     * This may only be called while the thread is not running.
     * <P>
     * This should be invoked with care, as it can cause an infinite
     * loop if another thread is pushing in data after this processing
     * thread has finished (but, that could also lead to an out-of-memory
     * error if this method is never invoked).
     */
    public void processRemaining()
            throws InterruptedException
    {
        if (isRunning())
        {
            throw new IllegalStateException(
                "cannot call processRemaining() while the underlying thread "+
                "is still running." );
        }
        if (!this.m_queue.isEmpty())
        {
            this.m_isProcessingObject = true;
            do
            {
                Object o = this.m_queue.dequeue();
                this.m_objListener.processObject( o );
            }
            while (!this.m_queue.isEmpty());
            this.m_isProcessingObject = false;
        }
    }
    
    
    //--------------------------------------------------------------
    // Protected Methods
    
    
    protected void initialize( IObjectListener ol, SynchQueue sq )
    {
        setRunnable( new QueueRunnable() );
        
        this.m_objListener = ol;
        this.m_queue = sq;
        
        initializeDefaults();
    }
    
    
    protected void initializeDefaults()
    {
        setSleepTime( 0 );
        setDaemon( true );
        setPriority( Thread.MIN_PRIORITY );
    }
    
    
}