package com.sibilantsolutions.iptools.net;

import java.lang.Thread.UncaughtExceptionHandler;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sibilantsolutions.iptools.event.LostConnectionEvt;
import com.sibilantsolutions.iptools.event.ReceiveEvt;
import com.sibilantsolutions.iptools.event.SocketListenerI;

public class ReceiveQueue implements SocketListenerI
{
    final static private Logger log = LoggerFactory.getLogger( ReceiveQueue.class );

    private SocketListenerI dest;

    final private ThreadPoolExecutor executorService = createExecutorService();

    public ReceiveQueue( SocketListenerI dest )
    {
        this.dest = dest;
    }

    private ThreadPoolExecutor createExecutorService()
    {
        ThreadFactory threadFactory = new ThreadFactory() {

            @Override
            public Thread newThread( Runnable r )
            {
                Thread t = new Thread( r, "queueTaker " + dest );

                t.setUncaughtExceptionHandler( new UncaughtExceptionHandler() {

                    @Override
                    public void uncaughtException( Thread t, Throwable e )
                    {
                        log.error( "Uncaught exception in thread=" + t + ":", e );
                    }
                } );

                return t;
            }
        };

        return new ThreadPoolExecutor( 1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(),
                threadFactory );
    }

    @Override
    public void onLostConnection( LostConnectionEvt evt )
    {
        log.info( "Lost connection: Shutting down executor service." );

        shutdownAndAwaitTermination( executorService, executorService.getQueue().size() );

        log.info( "Lost connection: Firing event to destination." );

        dest.onLostConnection( evt );
    }

    @Override
    public void onReceive( final ReceiveEvt evt )
    {
        Runnable r = new Runnable() {

            @Override
            public void run()
            {
                dest.onReceive( evt );
            }
        };

        executorService.execute( r );
    }

        //Adapted from example in javadoc of ExecutorService.
    private void shutdownAndAwaitTermination( ExecutorService pool, int queueSize )
    {
        log.info( "Shutting down executor service; queue size={}.", queueSize );

        pool.shutdown(); // Disable new tasks from being submitted

        log.info( "Waiting for executor service to finish all queued tasks." );

        try
        {
            // Wait a while for existing tasks to terminate
            if ( ! pool.awaitTermination( 60, TimeUnit.SECONDS ) )
            {
                log.info( "Executor service did not terminate before timeout; forcing shutdown." );

                List<Runnable> queuedTasks = pool.shutdownNow(); // Cancel currently executing tasks

                log.info( "Discarded {} tasks that were still queued; still waiting for executor service to terminate."
                        , queuedTasks.size() );

                // Wait a while for tasks to respond to being cancelled
                if ( ! pool.awaitTermination( 60, TimeUnit.SECONDS ) )
                {
                    log.error( "Pool did not terminate." );
                }
                else
                {
                    log.info( "Executor service terminated after forced shutdown." );
                }
            }
            else
            {
                log.info( "Executor service finished all queued tasks and terminated." );
            }
        }
        catch ( InterruptedException ie )
        {
            log.error( "Interrupted while waiting for pool to terminate; forcing shutdown.", ie );

            // (Re-)Cancel if current thread also interrupted
            List<Runnable> queuedTasks = pool.shutdownNow();

            log.info( "Discarded {} tasks that were still queued; no longer waiting for executor service to terminate."
                    , queuedTasks.size() );

            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }

}

@@ -33,24 +33,259 @@
 tools.net;%0A%0A
+import com.sibilantsolutions.iptools.event.LostConnectionEvt;%0Aimport com.sibilantsolutions.iptools.event.ReceiveEvt;%0Aimport com.sibilantsolutions.iptools.event.SocketListenerI;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0A
 import java.
@@ -573,243 +573,1070 @@
 t;%0A%0A
-import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport com.sibilantsolutions.iptools.event.LostConnectionEvt;%0Aimport com.sibilantsolutions.iptools.event.ReceiveEvt;%0Aimport com.sibilantsolutions.iptools.event.SocketListenerI;%0A
+/**%0A * ReceiveQueue is a %3Ca href=%22https://en.wikipedia.org/wiki/Decorator_pattern%22%3Edecorator%3C/a%3E for a SocketListenerI and%0A * is intended to be used in a %22receiver chain%22 with another instance of SocketListenerI.  This class maintains an%0A * executor service containing a single thread, which is used as a queue.  When data is received from the socket and%0A * onReceive is called, the data will be put into the queue and the calling thread can immediately go back to read more%0A * data.  Meanwhile, the queued message will be fired to the decorated SocketListenerI for handling.%0A * %3Cp%3E%0A * This is a low-level queue which contains only the raw byte stream received from the operating system.  It is the%0A * responsibility of the application to assemble the byte stream into more meaningful data structures.%0A * %3Cp%3E%0A * Benefits:%0A * - gets data out of the operating system receive queue and into the control of the application%0A * - allows queued data to be drained into the application in case of a lost connection before notifying the application of lost connection%0A */
 %0Apub

Add javadoc for ReceiveQueue.