/*
    * @(#) ThreadedEventProducer.java Jul 12, 2004
    * 
    * Copyright (c) 2003-2005 Delft University of Technology, Jaffalaan 5, 2628 BX
    * Delft, the Netherlands. All rights reserved.
    * 
    * See for project information <a href="http://www.simulation.tudelft.nl/">
    * www.simulation.tudelft.nl </a>.
    * 
   * The source code and binary code of this software is proprietary information
   * of Delft University of Technology.
   */
  
  package org.gscg.common.interactionlayer;
  
  import java.io.IOException;
  import java.io.ObjectOutputStream;
  import java.io.Serializable;
  import java.lang.reflect.Field;
  import java.rmi.ConnectException;
  import java.rmi.RemoteException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  
  import nl.tudelft.simulation.event.Event;
  import nl.tudelft.simulation.event.EventInterface;
  import nl.tudelft.simulation.event.EventListenerInterface;
  import nl.tudelft.simulation.event.EventProducer;
  import nl.tudelft.simulation.language.reflection.ClassUtil;
  import nl.tudelft.simulation.logger.Logger;
  
  import org.gscg.common.DijkstraSemaphore;
  
  /***
   * The ThreadedEventProducer extends the EventProducer. A different thread than
   * the one used in the simulation is used to fire events to a client-side
   * EventListener. Whenever the processing of an event client-side takes a long
   * time, the simulation will continue and not wait. Meanwhile, other events are
   * stored until the 'slow' event has been processed client-side. After this, the
   * ThreadedEventProducer will try to catch up.
   * <p>
   * 
   * Copyright (c) 2003-2005 Delft University of Technology, Jaffalaan 5, 2628 BX
   * Delft, the Netherlands. All rights reserved.
   * 
   * See for project information <a href="http://www.simulation.tudelft.nl/">
   * www.simulation.tudelft.nl </a>.
   * 
   * The source code and binary code of this software is proprietary information
   * of Delft University of Technology.
   * 
   * @see nl.tudelft.simulation.event.EventProducer
   * 
   * @author <a
   *         href="http://www.tbm.tudelft.nl/webstaf/stijnh/index.htm">Stijn-Pieter
   *         van Houten </a>
   * @author <a
   *         href="http://www.tbm.tudelft.nl/webstaf/alexandv/index.htm">Alexander
   *         Verbraeck </a>
   * 
   * @version $Revision: 1.1 $ $Date: 2005/06/16 12:33:58 $
   * @since 1.0.0
   */
  public class ThreadedEventProducer extends EventProducer
  {
  	/*** the dijkstra semaphore, value is adjusted in the model class */
  	public transient static DijkstraSemaphore DIJKSTRASEMAPHORE = new DijkstraSemaphore(
  			1);
  
  	/*** the serial version uid */
  	private static final long serialVersionUID = 13L;
  
  	/*** the source field of an event */
  	private static Field sourceField = null;
  
  	static
  	{
  		try
  		{
  			ThreadedEventProducer.sourceField = ClassUtil.resolveField(
  					Event.class, "source");
  			sourceField.setAccessible(true);
  		} catch (Exception exception)
  		{
  			Logger.severe(ThreadedEventProducer.class, "<clinit>", exception);
  		}
  	}
  
  	/*** this thread executes events */
  	private transient NotifyThread notifyThread = null;
  
  	/*** the list with events to fire */
  	private transient List events = Collections
  			.synchronizedList(new ArrayList());
  
  	/***
 	 * constructs a new ThreadedEventProducer
 	 */
 	public ThreadedEventProducer()
 	{
 		super();
 		this.notifyThread = new NotifyThread(this);
 		// since we use -Xss2m as an argument for the VM
 		// in order to support serialization.
 		// here we create a thread with stack size 1;
 		// as such forcing the VM to use the default value (128k)
 		// for the stack size for these threads instead of 2m
 		Thread thread = new Thread(null, this.notifyThread, "", 1);
 		thread.start();
 	}
 
 	/***
 	 * @see nl.tudelft.simulation.event.EventProducer#fireEvent(nl.tudelft.simulation.event.EventListenerInterface,
 	 *      nl.tudelft.simulation.event.EventInterface)
 	 */
 	protected synchronized EventInterface fireEvent(
 			final EventListenerInterface listener, final EventInterface event)
 			throws RemoteException
 	{
 		this.notifyThread.add(listener, event);
 		return event;
 	}
 
 	//
 	// private methods
 	//
 	/***
 	 * writes a serializable method to stream
 	 * 
 	 * @param out the outputstream
 	 * @throws IOException on IOException
 	 */
 	private synchronized void writeObject(final ObjectOutputStream out)
 			throws IOException
 	{
 		out.defaultWriteObject();
 		// we write the maximum count
 		out.writeObject(new Integer(ThreadedEventProducer.DIJKSTRASEMAPHORE
 				.getMaximumCount()));
 	}
 
 	/***
 	 * reads a serializable method from stream
 	 * 
 	 * @param in the inputstream
 	 */
 	private synchronized void readObject(final java.io.ObjectInputStream in)
 	{
 		try
 		{
 			in.defaultReadObject();
 			Integer maximumCount = (Integer) in.readObject();
 
 			// custom deserialize methods
 			this.events = Collections.synchronizedList(new ArrayList());
 
 			// set the maximum count of the dijkstra semaphore
 			ThreadedEventProducer.DIJKSTRASEMAPHORE
 					.setMaximumCount(maximumCount.intValue());
 
 			this.notifyThread = new NotifyThread(this);
 			Thread thread = new Thread(null, this.notifyThread, "", 1);
 			thread.start();
 		} catch (Exception exception)
 		{
 			Logger.severe(this, "readObject", exception);
 		}
 	}
 
 	/***
 	 * A combination of a reference and an event.
 	 * <p>
 	 * (c) copyright 2005 <a href="http://www.simulation.tudelft.nl">Delft
 	 * University of Technology </a>, the Netherlands. <br>
 	 * See for project information <a
 	 * href="http://www.simulation.tudelft.nl">www.simulation.tudelft.nl </a>
 	 * <br>
 	 * 
 	 * Copyright (c) 2003-2005 Delft University of Technology, Jaffalaan 5, 2628
 	 * BX Delft, the Netherlands. All rights reserved.
 	 * 
 	 * See for project information <a href="http://www.simulation.tudelft.nl/">
 	 * www.simulation.tudelft.nl </a>.
 	 * 
 	 * The source code and binary code of this software are proprietary
 	 * information of Delft University of Technology.
 	 * 
 	 * @author <a
 	 *         href="http://www.tbm.tudelft.nl/webstaf/stijnh/index.htm">Stijn-Pieter
 	 *         van Houten </a>
 	 * @version $Revision: 1.1 $ $Date: 2005/06/16 12:33:58 $
 	 * @since 1.1.3
 	 */
 	private class EventReference implements Serializable
 	{
 		/*** the serial version uid */
 		private static final long serialVersionUID = 11L;
 
 		/*** the reference */
 		private EventListenerInterface reference;
 
 		/*** the event */
 		private EventInterface event;
 
 		/***
 		 * constructs a new EventReference
 		 * 
 		 * @param event the event
 		 * @param reference the reference
 		 */
 		public EventReference(final EventInterface event,
 				final EventListenerInterface reference)
 		{
 			this.event = event;
 			this.reference = reference;
 		}
 
 		/***
 		 * @return Returns the event.
 		 * 
 		 */
 		public EventInterface getEvent()
 		{
 			return this.event;
 		}
 
 		/***
 		 * @return Returns the reference.
 		 * 
 		 */
 		public EventListenerInterface getReference()
 		{
 			return this.reference;
 		}
 	}
 
 	/***
 	 * The NotifyThread fires the event and notifies a client-side remote
 	 * object.
 	 * <p>
 	 * Copyright (c) 2003-2005 Delft University of Technology, Jaffalaan 5, 2628
 	 * BX Delft, the Netherlands. All rights reserved.
 	 * 
 	 * See for project information <a href="http://www.simulation.tudelft.nl/">
 	 * www.simulation.tudelft.nl </a>.
 	 * 
 	 * The source code and binary code of this software is proprietary
 	 * information of Delft University of Technology. *
 	 * 
 	 * @author <a
 	 *         href="http://www.tbm.tudelft.nl/webstaf/stijnh/index.htm">Stijn-Pieter
 	 *         van Houten </a>
 	 * @author <a
 	 *         href="http://www.tbm.tudelft.nl/webstaf/alexandv/index.htm">Alexander
 	 *         Verbraeck </a>
 	 * @version $Revision: 1.1 $ $Date: 2005/06/16 12:33:58 $
 	 * @since 1.0.0
 	 */
 	private class NotifyThread implements Runnable
 	{
 		/*** the number of events, used for performance measuring */
 		private int number = 0;
 
 		/*** the counter for the total number of events */
 		private int counter = 0;
 
 		/*** the size of the content of events in kb sent to remote clients */
 		private int size = 0;
 
 		/***
 		 * indicates whether to measure performance (for output stream size;
 		 * slows down the simulation)
 		 */
 		private boolean performance = false;
 
 		/*** the owner of the thread */
 		private ThreadedEventProducer owner = null;
 
 		/*** the set with dead references */
 		private Set deadReferences = new HashSet();
 
 		/***
 		 * constructs a new NotifyThread
 		 * 
 		 * @param owner the owner
 		 */
 		public NotifyThread(final ThreadedEventProducer owner)
 		{
 			super();
 			this.owner = owner;
 		}
 
 		/***
 		 * Method add.
 		 * 
 		 * @param reference the reference
 		 * @param event the event
 		 */
 		public void add(final EventListenerInterface reference,
 				final EventInterface event)
 		{
 			if (!this.deadReferences.contains(reference))
 			{
 				this.owner.events.add(new EventReference(event, reference));
 				if (this.performance)
 				{
 					this.number++;
 					this.counter++;
 					// we measure the output stream
 					this.size += ObjectMeter.getObjectSize(event.getContent());
 					if (this.counter % 10 == 0)
 					{
 						System.out.println("ThreadedEventProducer: Number: "
 								+ this.number);
 						System.out.println("ThreadedEventProducer: Size: "
 								+ this.size / 1024 + " kb");
 					}
 				}
 			} else
 			{
 				// the reference died.....
 				this.owner.removeListener(reference, event.getType());
 			}
 		}
 
 		/***
 		 * @see java.lang.Runnable#run()
 		 */
 		public void run()
 		{
 			while (true)
 			{
 				try
 				{
 					if (this.owner.events.size() > 0)
 					{
 						// we get our handle from the dijkstra semaphore
 						// during serialization no handles are available,
 						// leading to a pause of this thread
 						ThreadedEventProducer.DIJKSTRASEMAPHORE.acquire();
 						{
 							while (this.owner.events.size() > 0)
 							{
 								try
 								{
 									EventReference er = (EventReference) this.owner.events
 											.remove(0);
 									if (this.performance)
 									{
 										this.number--;
 										this.counter++;
 										if (this.counter % 10 == 0)
 										{
 											System.out
 													.println("ThreadedEventProducer: Number: "
 															+ this.number);
 											System.out
 													.println("ThreadedEventProducer: Size: "
 															+ this.size
 															/ 1024
 															+ " kb");
 										}
 									}
 									if (!this.deadReferences.contains(er
 											.getReference()))
 									{
 										try
 										{
 											// replace the source of the
 											// event
 											ThreadedEventProducer.sourceField
 													.set(er.getEvent(),
 															this.owner);
 											// invoke notify
 											er.getReference().notify(
 													er.getEvent());
 										} catch (ConnectException connectException)
 										{
 											if (this.performance)
 											{
 												System.out
 														.println("ThreadedEventProducer: Reference died: "
 																+ er
 																		.getReference()
 																+ " event: "
 																+ er.getEvent());
 											}
 											this.deadReferences.add(er
 													.getReference());
 											// we ignore this exception
 											// probably the client
 											// application is dead
 											this.owner.removeListener(er
 													.getReference(), er
 													.getEvent().getType());
 										}
 									}
 								} catch (Exception exception)
 								{
 									nl.tudelft.simulation.logger.Logger.severe(
 											this, "run", exception);
 								}
 							}
 						}
 						ThreadedEventProducer.DIJKSTRASEMAPHORE.release();
 					}
 					try
 					{
 						// we sleep one nanosecond and see whether we need
 						// to fire an event
 						Thread.sleep(1, 0);
 					} catch (InterruptedException interruptedException)
 					{
 						nl.tudelft.simulation.logger.Logger.severe(this, "run",
 								interruptedException);
 					}
 				} catch (Exception exception)
 				{
 					ThreadedEventProducer.DIJKSTRASEMAPHORE.release();
 					nl.tudelft.simulation.logger.Logger.severe(this, "run",
 							exception);
 				}
 			}
 		}
 	}
 }