/*
    * Jour - java profiler and monitoring library
    *
    * Copyright (C) 2004 Jour team
    *
    * This library is free software; you can redistribute it and/or
    * modify it under the terms of the GNU Library General Public
    * License as published by the Free Software Foundation; either
    * version 2 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
   * Library General Public License for more details.
   *
   * You should have received a copy of the GNU Library 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.
   */
  package net.sf.jour.rt.agent;
  
  import java.util.Enumeration;
  import java.util.Hashtable;
  import java.util.LinkedList;
  
  import org.apache.log4j.Logger;
  
  import net.sf.jour.rt.RtProperties;
  import net.sf.jour.util.ShutdownListener;
  import net.sf.jour.util.ShutdownHook;
  import net.sf.jour.util.queue.Queue;
  
  /***
   * TODO Add docs
   * Created on 02.12.2004
   *
   * Contributing Author(s):
   *
   *   Misha Lifschitz <mishalifschitz at users.sourceforge.net> (Inital implementation)
   *   Vlad Skarzhevskyy <vlads at users.sourceforge.net> (Inital implementation)
   *
   * @author vlads
   * @version $Revision: 1.8 $ ($Author: vlads $) $Date: 2004/12/12 02:00:01 $
   */
  public class AsyncEventWriter implements BulkEventLogger {
  
      protected static final Logger log = Logger.getLogger(AsyncEventWriter.class);
  
  	public static final int DEFAULT_DISPATCHER_THREADS = 100;
  	public static final int DEFAULT_DISPATCHER_BUFFER = 100;
  
      /***
       * There could be more than one thread in each Dispatcher
       */
      Hashtable dispatchers = new Hashtable(50);
      int dispatcherCnt;
  
      Dispatcher lastUsed = null;
  	int maxThreads;
  
  	public AsyncEventWriter() {
  		log.info("new AsyncEventWriter()");
  		maxThreads = RtProperties.getInstance().getProperty("AsyncEventLogger.threads", -1);
  	}
  
  	public boolean isThreadable() {
  		return (maxThreads != 0);
  	}
  
  	public void recive(Thread key, Queue eventQueue) {
  		if (isThreadable()) {
  			dispatch(key, eventQueue);
  		} else {
  			if (lastUsed == null) {
  				lastUsed = new Dispatcher("JourD");
  				dispatchers.put(key, lastUsed);
  			}
  			lastUsed.write(key, eventQueue);
  		}
  	}
  
      /* (non-Javadoc)
       * @see net.sf.jour.rt.agent.BulkEventLogger#recive(java.lang.Thread, net.sf.jour.util.queue.Queue)
       */
      public synchronized void dispatch(Thread key, Queue eventQueue) {
  		log.debug("dispatch");
          Dispatcher d = (Dispatcher)dispatchers.get(key);
          if (d == null) {
  			int per_thread = RtProperties.getInstance().getProperty("AsyncEventLogger.dispatcher.per_thread", DEFAULT_DISPATCHER_THREADS);
  			log.debug("per_thread " + per_thread);
              if ((lastUsed == null) || (lastUsed.size() >= per_thread)) {
                  dispatcherCnt ++;
                  lastUsed = new Dispatcher("JourD-" + dispatcherCnt);
                  lastUsed.start();
              }
              d = lastUsed;
  			dispatchers.put(key, d);
  			d.attach(key);
         }
         d.recive(key, eventQueue);
     }
 
 	/***
 	 * Close the Queue for this thread.
 	 */
     public void close(Thread key) {
 		log.debug("close Thread");
         Dispatcher d;
 
         if (isThreadable()) {
             d = (Dispatcher) dispatchers.get(key);
         } else {
             d = lastUsed;
         }
 
         if (d != null) {
             d.close(key);
         }
 
         if (d.size() == 0) {
             if (isThreadable()) {
                 dispatchers.remove(key);
             } else {
                 lastUsed = null;
             }
 			d.close();
             if (isThreadable()) {
                 try {
                     d.join();
                 } catch (InterruptedException ex) {
                     log.error("Got an InterruptedException while waiting for the dispatcher to finish.", ex);
                 }
             }
         }
     }
 
 	/***
 	 * Close All and be ready to exit.
 	 */
     public void close() {
 		log.debug("close, dispatchers " + dispatchers.size());
         synchronized (dispatchers) {
             for (Enumeration e = dispatchers.keys(); e.hasMoreElements();) {
                 Thread key = (Thread) e.nextElement();
                 Dispatcher d = (Dispatcher)dispatchers.get(key);
                 d.close();
                 if (isThreadable()) {
                     try {
                         d.join();
                     } catch (InterruptedException ex) {
                         log.error("Got an InterruptedException while waiting for the dispatcher to finish.", ex);
                     }
                 } else {
                 	d.closeWriters();
                 }
             }
             dispatchers.clear();
             lastUsed = null;
         }
 		log.debug("closed");
     }
 
 	public boolean isClose() {
 		return (dispatchers.size() == 0);
 	}
 
     class QueueItem {
 
         Thread key;
 
         Queue eventQueue;
 
         QueueItem(Thread key, Queue eventQueue) {
             this.key = key;
             this.eventQueue = eventQueue;
         }
     }
 
     class Dispatcher extends Thread implements ShutdownListener {
 
         boolean interrupted = false;
 
         Hashtable writers = new Hashtable(10);
 		Hashtable threadQueueCount = new Hashtable(10);
 		Hashtable toClose = new Hashtable(10);
 
         private LinkedList buffer = new LinkedList();
         private int buffer_max;
 
         Dispatcher(String name) {
         	super(name);
             log.info("new Dispatcher()" + name);
             //It is the user's responsibility to close appenders before exiting.
             this.setDaemon(true);
 			this.buffer_max = RtProperties.getInstance().getProperty("AsyncEventLogger.dispatcher.buffer_max", DEFAULT_DISPATCHER_BUFFER);
 			ShutdownHook.addListener(this);
         }
 
 		void attach(Thread key) {
 			getReciver(key);
 		}
 
         int size() {
             return writers.size();
         }
 
         void recive(Thread key, Queue eventQueue) {
             log.debug("recive");
             QueueItem qi = new QueueItem(key, eventQueue);
 			int sz;
 			int thisQueueCount;
             synchronized (buffer) {
                 sz = buffer.size();
                 buffer.addLast(qi);
 				thisQueueCount = queueCount(key, +1);
                 if (sz == 0) {
                 	// wakeup
                     buffer.notify();
                 }
             }
 
             // TODO review this.
             // Only Thread that actual have items Queue would sleep.
 			if ((sz > this.buffer_max) && (thisQueueCount > 10)) {
 				int sleeps = 0;
 				do {
 					try {
                         Thread.sleep(100);
                     } catch (InterruptedException e) {
 						break;
                     }
 					sleeps ++;
 					if (sleeps > 10) {
 						log.warn("Bufer too small Queue:" + thisQueueCount);
 						break;
 					}
 				} while (buffer.size() > this.buffer_max);
 			}
         }
 
         void close() {
             log.debug("close, buffer " + buffer.size());
             synchronized (buffer) {
                 interrupted = true;
                 // We have a waiting dispacther.
                 // In that case, we need to give it a death kiss.
 				if (buffer.size() == 0) {
 					buffer.notify();
 				}
             }
 			log.debug("closed");
         }
 
         /***
           * Close the Queue for this thread.
           */
         void close(Thread key) {
 			log.debug("close Thread");
             toClose.put(key, key);
             int cnt = queueCount(key, 0);
 			BulkEventLogger w = (BulkEventLogger)writers.get(key);
             while ((cnt > 0) || (!w.isClose())) {
             	synchronized (buffer) {
                 	buffer.notify();
             	}
 				cnt = queueCount(key, 0);
             }
         }
 
 		synchronized void closeWriters() {
 			log.debug("closeWriters " + writers.size());
 			for (Enumeration e = writers.keys(); e.hasMoreElements();) {
 				Object key = e.nextElement();
 				BulkEventLogger w = (BulkEventLogger) writers.get(key);
 				if (w != null) {
 					w.close();
 				}
 			}
 		}
 
         void write(Thread key, Queue eventQueue) {
 			BulkEventLogger reciver = getReciver(key);
 			reciver.recive(key, eventQueue);
         }
 
 		BulkEventLogger getReciver(Thread key) {
             BulkEventLogger reciver = (BulkEventLogger) writers.get(key);
             if (reciver == null) {
                 String dest = RtProperties.getInstance().getProperty("AsyncEventLogger.dispatcher.destination", "BulkEventFileWriter");
                 if (dest.equals("BulkEventLog4jWriter")) {
                     reciver = new BulkEventLog4jWriter();
                 } else if (dest.equals("BulkEventJMSWriter")) {
                     reciver = new BulkEventJMSWriter();
                 } else {
                     reciver = new BulkEventFileWriter();
                 }
                 writers.put(key, reciver);
                 log.info("dispatcher.destination selected " + dest);
             }
             return reciver;
         }
 
         public int queueCount(Thread key, int inc) {
             Integer i = (Integer) threadQueueCount.get(key);
             if (i == null) {
                 i = new Integer(inc);
                 if (inc != 0) {
                     threadQueueCount.put(key, i);
                 }
             } else {
 				if (inc != 0) {
 					i = new Integer( i.intValue() + inc);
 					threadQueueCount.put(key, i);
 				}
             }
             return i.intValue();
         }
 
         public void shutdown() {
             log.debug("Dispatcher " + this.getName() + " shutdown");
             this.interrupted = true;
         }
 
         public void run() {
             try {
                 log.info("run");
                 while (true) {
                     QueueItem qi = null;
                     BulkEventLogger reciver = null;
 					int cnt = -1;
                     synchronized (buffer) {
                         if (buffer.size() == 0) {
                             // Exit loop if interrupted but only if the the buffer is empty.
                       	  	if (interrupted) {
                       	  	    log.info("Exiting.");
                       	  	    break;
                       	  	}
 
                             try {
 								log.debug("wait");
                                 buffer.wait();
                             } catch (InterruptedException e) {
                                 log.error("The dispathcer should not be interrupted.");
                                 break;
                             }
                         }
 
 						if (buffer.size() != 0) {
                         	qi = (QueueItem)buffer.removeFirst();
 							cnt = queueCount(qi.key, -1);
 							reciver = getReciver(qi.key);
 						}
                     }
                     log.debug("continue");
                     if ((reciver != null) && (qi != null)) {
                         reciver.recive(qi.key, qi.eventQueue);
                         if (cnt == 0) {
 							if (toClose.remove(qi.key) != null) {
 								writers.remove(qi.key);
 								reciver.close();
 							}
                         }
                     }
                 }
                 closeWriters();
             } catch (RuntimeException e) {
 				log.error("The dispathcer terminated", e);
                 throw e;
             } finally {
                 ShutdownHook.removeListener(this);
             }
         }
     }
 
 }