public class MessagingConsole {

    private static final Logger logger = LoggerFactory.getLogger(MessagingConsole.class);
    private static final MessagingConsole me = new MessagingConsole();
    private final Map<PID,MsgQueue> mQueues = new ConcurrentHashMap<PID,MsgQueue>();

    private final static int POLL_EVERY_MS = 10;

    public static MessagingConsole getConsole() {
        return me;
    }

    /**
     * Remove mappings from previous execution in the same JVM / ClassLoader.
     * Useful e.g. for Unit Testing
     * 
     */

    public void reset() {
        logger.warn( "=========== SlicedBread console reset ================");
        mQueues.clear();
    }



    /**
     * Queues the message for delivery.
     *
     * @param message
     * @return The nummber of messages on the queue it's been delivered to.
     */
    public int send( Msg message ) {
        if ( message == null ) {
            return 0;
        }

        logger.trace("S: {}", message);

        MsgQueue q = getQueue( message.getToPid() );

        // if the destination address was not found...
        if ( q == null ) { 
            returnMsgUndeliverable(message);
            return 0;           
        }
        
        return q.push(message);
    }

    /**
     * Gets you a message from a queue.
     * If the queue is empty, then this call will block for the given timeout (in ms).
     * If you don't want it blocking, pass 0 as the timeout.
     *
     * @param myPid which mailbox to check
     * @param timeout in milliseconds
     * @return a messagem, or null if none is found.
     * @throws InterruptedException
     */
    
    public Msg receive( PID myPid, int timeout ) throws InterruptedException {        
        int timeRemaining = timeout;

        MsgQueue q = getQueue( myPid );
        if ( q == null ) {
            return null;
        }

        do {
            Msg message = q.pull();
            if ( message == null ) {

                if ( timeRemaining > 0 ) {
                    Thread.sleep( POLL_EVERY_MS );
                    timeRemaining = timeRemaining - POLL_EVERY_MS;
                }

            } else {
                // I found a message at last                
                logger.trace("R: {}", message);
                return message;
            }

        } while ( timeRemaining > 0 );

        // no message was found        
        return null;
    }



    /**
     * Returns all messages.
     *
     * @param myPID
     * @return a list of all messages on the mailbox.
     */

    public List<Msg> receiveAll( PID myPID ) {
        MsgQueue q = getQueue( myPID );
        return q.fetchAllMessages();
    }



    /**
     * Crea un nuovo PID per un thread esistente.
     * Il PID e' sempre diverso.
     * 
     * @param description
     * @return a a new, unique PID.
     */

    public PID registerExistingThread( String description ) {
        final PID processPid = PID.getNextPid( description );
        addQueue(processPid);
        return processPid;
    }


    /**
     * Given an unique name, creates a new mailbox.
     * 
     * @param uniqueName
     * @return a new PID, unles there is already one under that name.
     */
    public PID registerExistingThreadByName( String uniqueName ) {
        final PID px = whois( uniqueName );
        if ( px != null ) {
            return px;
        } else {
            return registerExistingThread(uniqueName);
        }
    }


    /**
     * Start a thread
     *
     * @param callerPid
     * @param description
     * @param process
     * @return the PID for the new thread.
     */

    public PID register( final PID callerPid, String description, final TaskProcess process ) {

        final PID processPid = PID.getNextPid( description );
        addQueue(processPid);

        process.setOwnPid(processPid);
        process.setParentPid(callerPid);
        process.setConsole(me);

        Runnable task = new Runnable() {

            public void run() {
                try {

                    send( MsgProcessStarted.build( processPid, callerPid));
                    process.run();
                    send( MsgProcessEnded.build( processPid, callerPid ) );


                } catch ( Throwable t)  {

                    logger.error("Process " + processPid + " died ", t);
                    send( MsgErrProcessDied.build( processPid, callerPid, t ) );
                    
                } finally {

                    // Marking all existing messages as "undeliverable"
                    MsgQueue myMailbox = getQueue(processPid);
                    List<Msg> undeliveredMsgs = myMailbox.fetchAllMessages();
                    for ( Msg m: undeliveredMsgs ) {
                        returnMsgUndeliverable(m);
                    }
                    removeQueue(processPid);
                }
            }
        };

        Thread tx = new Thread(task, description);
        tx.setDaemon(true);
        tx.start();

        logger.info( "Thread {} started", description );

        return processPid;

    }
    
    /**
     * Registers a thread only if it is not already present.
     * If it is present, returns the current PID.
     * If not, registers it and returns the new PID.
     * 
     * @param callerPid The parent's PID. Can be null.
     * @param description A description of this thread.
     * @param process The TaskProcess object that will be run.
     * @return a PID for the running thread.
     */
    
    
    public PID registerIfNotRunning( final PID callerPid, String description, final TaskProcess process ){
        
        // First I try without locking - in most cases it's either there or not.
        PID pid = findByDescription(description);
        if ( pid != null ) {
            return pid;
        }
        
        synchronized (this) {
            // If it is not there, I try again locking to mke sure that this canot be run 
            // in parallel.
            pid = findByDescription(description);
            if ( pid != null ) {
                return pid;
            }
            
            return register(callerPid, description, process);
        }
        
    }
    

    /**
     * Looks up for a PID given its description.
     * If not found, returns null.
     * This is jus an alias for findByDescription.
     * 
     * @param description
     * @return the PID found, or null.
     */

    public PID whois( String description ) {
        return findByDescription(description);
    }

    /**
     * Get a list of objects that describe the current open mailboxes.
     *
     * @return The list of MailBoxes info objects
     */

    public List<QueueInfo> list() {
        List<QueueInfo> lQueues = new ArrayList<QueueInfo>();

        for ( PID pid: mQueues.keySet() ) {

            MsgQueue mq = getQueue(pid);

            if ( mq != null) {
                QueueInfo qi = new QueueInfo();
                qi.queuePid = pid;
                qi.queueSize = mq.size();
                qi.queueStats = mq.getStats();
                lQueues.add(qi);
            }
        }

        return lQueues;
    }




    /**
     * Adds a queue.
     * If a queue for the same PID alredy exists, the old queue
     * is discarded.
     *
     * This method requires no synchronization as the map is synchonized.
     *
     * @param pid
     */

    public void addQueue(PID pid) {
        if (mQueues.containsKey(pid)) {
            mQueues.remove(pid);
        }

        mQueues.put(pid, new MsgQueue());
    }

    /**
     * Gets you a  the list of messages from a queue.
     * 
     * @param pid
     * @return the inner MsgQueue object.
     */

    private MsgQueue getQueue( PID pid ) {
        
        if ( pid != null) {
            if ( mQueues.containsKey(pid)) {
                return mQueues.get(pid);
            }
        }
        return null;
    }

    /**
     * Drops a queue if it exists.
     * 
     * @param pid
     */

    public void removeQueue(PID pid) {
        if (mQueues.containsKey(pid)) {
            mQueues.remove(pid);
        }
    }

    /**
     * Loads a PID by description.
     * If no PID found, returns null.
     * 
     * @param description
     * @return the PID (if found) or null.
     */
    
    
    public PID findByDescription( String description ) {
        
        PID res = null;

        // TOLGO synchronized ( mQueues ) {
        // non serve la sincronizzazione per gli oggetti "concurrent"
        //
        //    The view's returned iterator is a "weakly consistent" iterator
        //    that will never throw ConcurrentModificationException, and
        //    guarantees to traverse elements as they existed upon construction
        //    of the iterator, and may (but is not guaranteed to) reflect any
        //    modifications subsequent to construction.

        Iterator<PID> i = mQueues.keySet().iterator();
        while ( i.hasNext() ) {
            PID px = i.next();

            if ( description.equals( px.getDescr() ) ) {
                res = px;
                break;
            }
        }

        return res;
    }

    /**
     * Returns a message to the sender.
     * We make sure that there is a valid received.
     *
     * @param message
     */

    private void returnMsgUndeliverable(Msg message) {

        // Is there a valid receiver?
        PID sourcePid = message.getFromPid();

        if (sourcePid != null) {
            // we check that the source address is a valid queue
            MsgQueue q2 = getQueue(sourcePid);

            if (q2 != null) {
                Msg error = MsgErrUndeliverable.build(message.getFromPid(), message);
                q2.push(error);

            } else {
                logger.error("Cannot return message {} to non-existent mailbox {}", message, sourcePid);
            }
        } else {
            logger.debug("Cannot return message with empty PID {}", message);
        }

    }

    /**
     * A blocking version of findByDescription.
     * 
     * @param description to be found
     * @param maxTimeout in milliseconds
     * @return the PID; or null if not found and timeout expired.
     */
    
    public PID blockingFindByDescription( final String description, long maxTimeout ) {
        
        boolean ok = new SbTools.BlockUntil() {

            @Override
            public boolean stopIfTrue() {
                PID lastPid = findByDescription(description);
                return (lastPid != null);
            }
        }.sync(maxTimeout);
        
        if ( ok ) {
            return findByDescription(description);
        } else {
            return null;
        }
        
    } 
    
    /**
     * Blocks until a mailbox is deleted.
     * 
     * @param description
     * @param maxTimeout
     * @return null if all went well, or the PID if a mailbox is still there 
     *         after timeout expired.
     */
    
    public PID blockUntilPidDisappears( final String description, long maxTimeout ) {
        
        boolean ok = new SbTools.BlockUntil() {

            @Override
            public boolean stopIfTrue() {
                PID lastPid = findByDescription(description);
                return (lastPid == null);
            }
        }.sync(maxTimeout);
        
        if ( ok ) {
            return null;
        } else {
            return findByDescription(description);
        }
        
    }
    
    
    
    
} public class SbTools {

    private static final Logger logger = LoggerFactory.getLogger(SbTools.class);
    
    /**
     * Turns an exception - actually a Throwable - into a String.
     * It stores the complete stack trace and inner exceptions as well.
     * Very handy for logging and storing it.
     *
     * @param ex The exception
     * @return A string representation
     */


    public static String stringifyException(Throwable ex) {
        StringBuilder stThrowable = new StringBuilder();
        stThrowable.append( "-- Inner Exception --\n" );

        if (ex != null) {
                stThrowable.append("Exception: ").append(ex.getClass().getName()).append("\n");
                String stMessage = ex.getMessage();

                if ( stMessage != null) {
                        stThrowable.append( "Error:\n" );
                        stThrowable.append( stMessage );
                }

                StringWriter out = new StringWriter();
                PrintWriter printOut = new PrintWriter(out);
                ex.printStackTrace(printOut);

                stThrowable.append( "\nStack trace:\n" );
                stThrowable.append(out.toString());
        }

        stThrowable.append( "-- End Inner Exception --\n");

        return stThrowable.toString();
    }


    /**
     * Funzione statica per fare la sleep di un thread.
     * Visto che molte classi che usano il framework la utilizzeranno,
     * tanto vale averla in un posto solo.
     * Mi evita di gestire l'eccezione - se la voglio gestire basta che guardo
     * lo stato risultante.
     *
     * @param ms Durata del sonno.
     * @return Se interrotta, ritorna false. la durata della pausa in questo caso
     *         non e' definita.
     *
     * \todo TRANSLATE
     */

    public static boolean sleep( int ms ) {
        try {
            Thread.sleep(ms);
            return true;
        } catch ( InterruptedException e ) {
            return false;
        }
    }

    /**
     * A simple way to block until a condition is met.
     */
    
    public static abstract class BlockUntil {

        /**
         * Waits until the specified condition is met.
         * 
         * @param maxWait in ms
         * @return true: ok - false: timeout
         */
        public boolean sync(long maxWait) {
            long beginning = System.currentTimeMillis();
            long took = 0;
            do {

                if (stopIfTrue()) {
                    return true;
                }

                SbTools.sleep(100);
                took = System.currentTimeMillis() - beginning;

            } while (took < maxWait);
            return false;
        }

        
        /**
         * Stop procesing if this function returns true.
         * 
         * @return true to stop (condition met); else false. 
         */
        public abstract boolean stopIfTrue();
    }

    /**
     * Waits for a mailbox to be existent or non-existent.
     * 
     * @param mailbox the name
     * @param maxTimeout max timeout
     * @param isUp true: up, false: downs
     * @return 
     */
    
    
    private static boolean awaitMailbox( final String mailbox, int maxTimeout, final boolean isUp ) {
        
        boolean result = new BlockUntil() {

            MessagingConsole mc = MessagingConsole.getConsole();
            
            @Override
            public boolean stopIfTrue() {
                return (( mc.findByDescription( mailbox ) != null) == isUp);
            }
            
        }.sync(maxTimeout);
        
        if ( !result  ) {
            logger.error( "Timed out waiting for mailbox '" + mailbox + "' "
                        + "to go" + (isUp ? "up": "down"));
        }
        
        return result;
    }
        
        
    public static boolean awaitMailboxUp( String mailbox, int maxTimeout ) {
        return awaitMailbox( mailbox, maxTimeout, true);
    }
    
    
    public static boolean awaitMailboxDown( String mailbox, int maxTimeout ) {
        return awaitMailbox( mailbox, maxTimeout, false);
    }
    
    /**
     * Shuts down a PID and waits for confirmation.
     * Uses a temporary mailbox that is deleted after some answer is received.
     * 
     * @param to the PID to shut down.
     * @return 
     */
    
    public static Msg shutdown( PID to ) {
        
        PID temp = null; 
        Msg msg = null;
        MessagingConsole mc = MessagingConsole.getConsole();
            
        try {
            
            temp = mc.registerExistingThread( "Shutdown/" + Math.random() );
            mc.send( MsgPleaseStop.build(temp, to));
            // waits for the answer.
            msg = mc.receive(temp, 5000);
        
            if ( msg == null ) {
                logger.error( "Thread " + to + " did not terminate within the timeout.");
            }
            
        } catch ( Throwable t ) {
            logger.error("While shutting down " + to, t);
            
        } finally {
            if ( temp != null ) {
                mc.removeQueue(temp);
            }
        }
        
        return msg;
    }
    

} public class PID {

    private static int nextPid = 0;

    private PID() {
        // no public constructor
        // use the static build method
    }

    private  int pid = 0;
    private  String descr = "";


    @Override
    public String toString() {

        StringBuilder sb = new StringBuilder();
        sb.append( descr )
           .append( '#' )
           .append( pid );

        return  sb.toString();
    }

    // gettr e setter
    public int getPid() {
        return pid;
    }

    public void setPid(int pid) {
        this.pid = pid;
    }

    public void setDescr(String descr) {
        this.descr = descr;
    }

    public String getDescr() {
        return descr;
    }




    /**
     * Metodo builder.
     * 
     * @param myPid
     * @return Oggetto PID
     */

    public static PID build( int myPid, String description ) {
        PID p = new PID();
        p.setPid(myPid);
        p.setDescr(description);
        return p;
    }

    /**
     * Ottiene un nuovo PID unico.
     * 
     * @return the new PID.
     */

    public static synchronized PID getNextPid( String description) {

        nextPid += 1;
        return build(nextPid, description);

    }

    public static PID systemPid() {
        return PID.build(0, "SYSTEM");
    }

    /**
     * \todo vedi Effective Java
     * @param obj
     * @return equality
     */

    @Override
    public boolean equals(Object obj) {

        if ( obj instanceof PID ) {
            PID pObj = (PID) obj;
            return ( pObj.getPid() == this.getPid() );
        } else {
            return false;
        }

    }

    /**
     * Per l'hashCode, uso il PID corrente
     * \todo vedi Effective Java
     *
     * @return hashcode
     */

    @Override
    public int hashCode() {
        return pid;
    }

    // \todo Metodo Compare e Hash - vedi Effective Java
    // \todo rendere final la classe?
} public class QueueInfo {

    public PID queuePid = null;
    public int queueSize = 0;
    public List<LogBin> queueStats = null;

    @Override
    public String toString() {
        return queuePid + " Size: " + queueSize + "\n" 
                + LogBin.printAsText(queueStats) + "\n";
    }

} public class Msg {

    private PID  fromPid   = null;
    private PID  toPid     = null;
    private long sentOn    = 0L;
    private int  replyTo   = 0;

    // ======================================================
    //        m e t o d i     d i     a c c e s s o
    // ======================================================


    /**
     * @return the sender Pid
     */
    public PID getFromPid() {
        return fromPid;
    }

    /**
     * @param fromPid the fromPid to set
     */
    public void setFromPid(PID fromPid) {
        this.fromPid = fromPid;
    }

    /**
     * @return the toPid
     */
    public PID getToPid() {
        return toPid;
    }

    /**
     * @param toPid the toPid to set
     */
    public void setToPid(PID toPid) {
        this.toPid = toPid;
    }

    /**
     * @return the sentOn
     */
    public long getSentOn() {
        return sentOn;
    }

    /**
     * @param sentOn the sentOn to set
     */
    public void setSentOn(long sentOn) {
        this.sentOn = sentOn;
    }

    /**
     * @return the replyTo
     */
    public int getReplyTo() {
        return replyTo;
    }

    /**
     * @param replyTo the replyTo to set
     */
    public void setReplyTo(int replyTo) {
        this.replyTo = replyTo;
    }



    // metodi comuni

    /**
     * Use only for debug - the builder is pretty unefficient.
     * 
     * @return a serialized view.
     */

    @Override
    public String toString() {
        return "F:" + fromPid
                + " T:" + toPid
                + ( replyTo > 0 ? " R: " + replyTo : "");
    }


    /**
     * Imposta i recipients.
     * 
     * @param from
     * @param to
     */
    public void setRecipients( PID from, PID to) {
        setFromPid(from);
        setToPid(to);
    }

} public class MsgActionIds extends MsgCommon {

    int code1 = 0;
    int code2 = 0;
    String text = "";

    public int getCode1() {
        return code1;
    }

    public int getCode2() {
        return code2;
    }

    public String getText() {
        return text;
    }

    public void setCode1(int code1) {
        this.code1 = code1;
    }

    public void setCode2(int code2) {
        this.code2 = code2;
    }

    public void setText(String text) {
        this.text = text;
    }




    @Override
    public String toString() {
        return super.toString() + " - ActionIds "
                + code1 + " "
                + code2 + " "
                + text;
    }

    public static MsgActionIds build( PID from, PID to, int c1, int c2, String t) {
        final MsgActionIds m =  new MsgActionIds();
        m.setFromPid(from);
        m.setToPid(to);
        m.setCode1(c1);
        m.setCode2(c2);
        m.setText(t);
        return m;
    }


} public class MsgPleaseStop extends MsgCommon {

    @Override
    public String toString() {
        return super.toString() + " - PleaseStop ";
    }

    public static MsgPleaseStop build( PID from, PID to) {
        final MsgPleaseStop m =  new MsgPleaseStop();
        m.setFromPid(from);
        m.setToPid(to);
        return m;
    }


} public class MsgPoolRunnable extends Msg {

    RTask task = null;
    
    /**
     * This interface is to be implemented by the pool tasks.
     */
    public static interface RTask {
        /**
         * This method shall clean up its own mess.
         * Do NOT leave open database connections, resources or anything.
         * 
         * @return a message to be enqueued, or null if not needed.
         */
        public Msg process();
    }
    
    /**
     * Builds an instance of the object.
     * 
     * @param from
     * @param to
     * @param task
     * @return a runnable message wrapping the RTask object.
     */
    
    public static MsgPoolRunnable build( PID from, PID to, RTask task ) {
        MsgPoolRunnable mpr = new MsgPoolRunnable();
        mpr.setRecipients(from, to);
        mpr.task = task;
        return mpr;
    } 
    
    /**
     * Builds a runnable getRTask out of an expired deferred getRTask.
     * 
     * @param m
     * @return 
     */
    
    public static MsgPoolRunnable build( MsgPoolDeferred m ) {
        return build(m.getFromPid(), m.getPool(), m.getRTask() );
    }
    
    /**
     * Obtains the RTask.
     * \todo 
     * 
     * @return the current getRTask. 
     */
    
    public RTask getRTask() {
        return task;
    }
    
    public void setRTask( RTask t) {
        task = t;
    }
    
} public class MsgThreadPool extends Msg {
    
    String name = "";
    int minChannels = 0;
    int maxChannels = 0;
    
    /**
     * Creates a thread pool.
     * 
     * @param from
     * @param to
     * @param name
     * @param min
     * @param max
     * @return 
     */
    
    public static MsgThreadPool build( PID from, PID to, String name, int min, int max ) {
        
        MsgThreadPool m = new MsgThreadPool();
        m.name = name;
        m.minChannels = min;
        m.maxChannels = max;
        m.setRecipients(from, to);
        return m;
        
    }
    
    /**
     * Shut down a thread pool.
     * 
     * @param from
     * @param to
     * @param name
     * @return 
     */
    
    public static MsgThreadPool buildShutdown( PID from, PID to, String name ) {
        return build( from, to , name, 0, 0 );
    }
    
    
    
    public boolean isShutdown() {
        return (( minChannels == 0 ) && (maxChannels == 0));
    }
    
    public String getName() {
        return name;
    }

    public int getMinChannels() {
        return minChannels;
    }
    
    public int getMaxChannels() {
        return maxChannels;
    }

} public class ThreadPool {

    private final static Logger logger = LoggerFactory.getLogger(ThreadPool.class);
    public final static String POOL_PREFIX = "Pool/";
    public final static String POOL_CTRL_SUFFIX = "/CTRL";
    
    
    MessagingConsole console = MessagingConsole.getConsole();
    String name = "";
    int    minChannels = 0;
    int    maxChannels = 0;
    PID    poolPID     = null;
    PID    fromPID     = null;
    Map<PID,ThreadState> threads   = new HashMap<PID,ThreadState>();
    int    currThreadNo = 0;
    
    public void create( String name, int min, int max ) {
        
        this.name = name;
        this.minChannels = min;
        this.maxChannels = max;
        
        this.poolPID = PID.getNextPid( getPoolName(name) );
        
        this.fromPID = PID.getNextPid( getPoolCtrlName(name) );
        console.addQueue(fromPID);

        for ( int i=0; i < minChannels; i++ ) {
            PID newThread = createNewThread();
            threads.put( newThread, ThreadState.STARTING );
        }
        
        console.addQueue(poolPID);

    }

    public void processMessages() {
        
        // Maybe the thread is already closed.
        if ( fromPID == null ) {
            return;
        }
        
        
        List<Msg> msgs = console.receiveAll(fromPID);
        
        for ( Msg m: msgs ) {
            if ( m instanceof MsgProcessStarted ) {
                updateThreadState( m.getFromPid(), ThreadState.RUNNING );
            } else
            if ( m instanceof MsgProcessEnded ) {
                // Segnalo se non e' in stato stopping ma in ogni
                // caso elimino il thread.
                
                if ( threadState( m.getFromPid()) == ThreadState.STOPPING ) {  
                    logger.error("Received MsgPleaseStop " + m + " with state " + threads );
                }
                
                threads.remove( m.getFromPid() );

                if ( threads.isEmpty() ) {

                    console.removeQueue(fromPID);
                    console.removeQueue(poolPID);
                    fromPID = null;
                    poolPID = null;

                    logger.warn("Thread pool " + name + " removed");

                }
                
            } else {
                logger.error("Unknown message " + m.toString() );
            }
        }
    }
    
    /**
     *  Close down everything.
     */
    
    public void shutdown() {
        
        if ( fromPID == null ) {
            logger.error("Thread pool " + name + " is already being shut down.");
            return;
        }
        
        
        Set<PID> t = threads.keySet();
        for (PID p : t) {
            ThreadState st = threadState(p);
            if (st != ThreadState.STOPPING) {
                console.send(MsgPleaseStop.build(fromPID, p));
                updateThreadState(p, ThreadState.STOPPING);
            }
        }
        
    }

    /***
     * Creates a new worker thread in this pool.
     * 
     * @return the PID for the new thread.
     */
    private PID createNewThread() {
        String threadname = getNewThreadName();
        
        PID newThread = console.register(fromPID, threadname, PoolMemberTask.build(poolPID) );
        return newThread;
    }
    
    /**
     * Gets a new unique thread name.
     * 
     * @return the name, like Pool/name/123
     */
    
    private String getNewThreadName() {
        currThreadNo += 1;
        String n = POOL_PREFIX + getPoolName(name) + "/" + currThreadNo;
        return n;
    }
    
    /**
     * Updated the thread state.
     * If the thread is unknowns it is ignored.
     * 
     * @param pid our thread
     * @param newState 
     */
    
    private void updateThreadState( PID pid, ThreadState newState ) {
        if (threads.containsKey(pid) ) {
            threads.put(pid, newState);
            logger.error(" - PID " + pid + " -> " + newState);
        }
    }
    
    /**
     * Gets the thread state.
     * 
     * @param pid
     * @return the current state, or UNKNOWN
     */
    
    private ThreadState threadState( PID pid ) {
        if (threads.containsKey(pid) ) {
            return threads.get(pid);
        } else {
            return ThreadState.UNKNOWN;
        }
    }
    
    /**
     * Finds (by name) the mailbox for a pool.
     * 
     * @param name
     * @return the name, like Pool/abcd
     */
    
    public static String getPoolName( String name ) {
        return name; 
    }
    
    /**
     * Gets the controller name.
     * 
     * @param name
     * @return the name, like Pool/abcd/CTRL
     */
    
    
    public static String getPoolCtrlName( String name ) {
        return getPoolName(name) + POOL_CTRL_SUFFIX; 
    }
    
    /**
     * Return the PID for a pool.
     * 
     * @param name
     * @return the PID if found, or null.
     */
    
    public static PID getPoolPid( String name ) {
        return MessagingConsole.getConsole().findByDescription( getPoolName(name) );
    }
    
    /**
     * Blocks until there is a PID for a pool and returns it.
     * 
     * @param name
     * @param maxWait
     * @return the PID if found, or null if not found until it times out.
     */
    
    public static PID getPoolPid_blocking( String name, long maxWait ) {
        return MessagingConsole.getConsole().blockingFindByDescription(getPoolName(name), maxWait  );
    }
    
    
    /**
     * Possible states for a thread
     */
    public static enum ThreadState {
        UNKNOWN,   /** No known thread */
        STARTING,  /** Thread started, not confirmed yet */
        RUNNING,  /** Thread is running */
        STOPPING; /** Thread is shutting down */
    }
} public class WatchdogTask extends TaskProcess {

    public static final String WATCHDOG = "Sb_Watchdog_Task";
    
    private final Logger logger = LoggerFactory.getLogger(WatchdogTask.class);
    PID watchdogPid = null;
    Map<String,ThreadPool> pools = new HashMap<String, ThreadPool>();
    List<MsgPoolDeferred> deferredTasks = new ArrayList();
    
    /**
     * This is our main loop.
     */
    
    @Override
    public void run() {

        watchdogPid = getOwnPid();
        log (
            "I am Watchdog "  + getProcessDescription()
            + " (created by " + getParentDescription() + ")"
        );
        MessagingConsole console = getConsole();

        try {
            for ( ;; ) {
                Msg m = console.receive(watchdogPid, 100);

                if ( m instanceof MsgPleaseStop ) {
                    logger.error("Watchdog gracefully stopping");
                    cleanup();
                    getConsole().send(MsgProcessEnded.build(watchdogPid, m.getFromPid()));
                    return;
                } else
                if ( m instanceof MsgPoolDeferred ) {
                    addToRunnablePool((MsgPoolDeferred) m);
                } else 
                if ( m instanceof MsgThreadPool ) {
                    updateThreadPool( (MsgThreadPool) m );
                }
                
                // Processes our thread pool
                processThreadPoolMailboxes();
                runDeferred();
               
                
                
            }
        } catch ( InterruptedException e ) {
            logger.error( "Watchdog interrupted: ", e);
            return;
        }

    };



    private void log( String s ) {        
        logger.warn(s);
    }

    /***
     * Creates / updates / deleted a thread pool.
     * 
     * @param m 
     */
    
    private void updateThreadPool( MsgThreadPool m ) {
        
        String name = m.getName();
        
        if ( pools.containsKey( name ) ) {
            
            ThreadPool tp = pools.get(name);
            if ( m.isShutdown() ) {
                // close all threads
                tp.shutdown();
                
            } else {
                // update number of threads
            }
            
        } else {
            if ( !m.isShutdown() ) {
                ThreadPool newPool = new ThreadPool();
                newPool.create( name, m.getMinChannels(), m.getMaxChannels() );
                pools.put( name, newPool );
            }
        } 
    }
    
    /**
     * Process messages for thread pools.
     * 
     */
    
    private void processThreadPoolMailboxes() {
        for ( ThreadPool pool: pools.values() ) {
            pool.processMessages();
        }
    }
    
    /**
     * Check whether all thread pools are down.
     * The idea is taht the mailbox for the Pool will be removed only when 
     * all threads confirm termination; so when it's not found, this means all 
     * threads are dead.
     * 
     * @return true if no thread pool is found; false if some are found.
     */
    
    private boolean areAllThreadPoolsDown() {
        for ( ThreadPool pool: pools.values() ) {
            PID thisPid = ThreadPool.getPoolPid( pool.name );
            if ( thisPid != null ) {
                return false;
            }
        }
        return true;
    }
    
    /***
     * Cleans up any resources that are still there.
     * 
     */
    public void cleanup() {
        cleanupThreadPools();
    }
    
    /**
     * Cleans up the thread pool.
     * Blocks until all ogf them are gone.
     * 
     */
    
    public void cleanupThreadPools() {
        
        long beginning = System.currentTimeMillis();
        
        for ( ThreadPool pool: pools.values() ) {
            pool.shutdown();
        }
        
        // It is very important that I process events
        // because thread pools release resorces based on closure 
        // events.
        boolean ok = new SbTools.BlockUntil() {

            @Override
            public boolean stopIfTrue() {
                WatchdogTask.this.processThreadPoolMailboxes();
                return WatchdogTask.this.areAllThreadPoolsDown();
            }
            
        }.sync(10000);
            
        long took = System.currentTimeMillis() - beginning;
        logger.error( "All thread pools cleaned in " + took + " ms - OK: " + ok );
        
    }
    
    /**
     * This is a static method that start the watchdog.
     * 
     * @return 
     */
    
    public static PID up() {
        MessagingConsole mc = MessagingConsole.getConsole();
        WatchdogTask wd = new WatchdogTask();
        PID watchdog = mc.registerIfNotRunning(null, WatchdogTask.WATCHDOG, wd);
        return watchdog;
        
    }
    
    /**
     * Shuts down the watchdog.
     * 
     */
    
    public static void down() {
        MessagingConsole mc = MessagingConsole.getConsole();
        PID watchdog = mc.findByDescription(WatchdogTask.WATCHDOG);
        SbTools.shutdown(watchdog);
    }
    
    /**
     * 
     */
    public void addToRunnablePool( MsgPoolDeferred m ) {
        deferredTasks.add(m);
    }
    
    public void runDeferred() {
        long now = System.currentTimeMillis();
        List<MsgPoolDeferred> lDeletions = new ArrayList();
        
        for ( MsgPoolDeferred m: deferredTasks ) {
            if ( m.getScheduleAfter() < now ) {
                lDeletions.add(m);
            }
        }
        
        deferredTasks.removeAll(lDeletions);
        for ( MsgPoolDeferred r: lDeletions ) {
            getConsole().send( MsgPoolRunnable.build(r));
        }
        
    }
    
    
}
import ch.loway.oss.slicedbread.MessagingConsole;
import ch.loway.oss.slicedbread.SbTools;
import ch.loway.oss.slicedbread.containers.PID;
import ch.loway.oss.slicedbread.containers.QueueInfo;
import ch.loway.oss.slicedbread.messages.Msg;
import ch.loway.oss.slicedbread.messages.common.MsgActionIds;
import ch.loway.oss.slicedbread.messages.common.MsgPleaseStop;
import ch.loway.oss.slicedbread.messages.wd.MsgPoolRunnable;
import ch.loway.oss.slicedbread.messages.wd.MsgThreadPool;
import ch.loway.oss.slicedbread.tasks.ThreadPool;
import ch.loway.oss.slicedbread.tasks.WatchdogTask;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

package ch.loway.oss.sbDemos.helloWorld;


/**
 * This class shows how to run a thread pool in SlicedBread.
 *
 * 
 * 
 * @author l3nz
 */
public class TestPool {
    
    private final static Logger logger = LoggerFactory.getLogger(TestPool.class);

    public static void main( String args[] ) {
        logger.warn("Starting up....");
        
        MessagingConsole console = MessagingConsole.getConsole();
        final PID myself = console.registerExistingThread("Main");

        PID watchdog = console.register(myself, WatchdogTask.WATCHDOG, new WatchdogTask() );
        
        console.send( MsgThreadPool.build(myself, watchdog, "P1", 10, 10) );