package jas2.util.rmi;

import java.rmi.*;
import java.rmi.server.*;
import java.io.*;
import java.util.*;


/** This class deals with delivering events to a remote process using RMI.
 * Since RMI does not have any asychronous method of delivering events, 
 * they are queued up and delivered in a chunk by an event delivery thread.
 * Events placed in the queue can be "Preemptable", in which case a later
 * event may preempt an earlier one, making it unnecessary to deliver the
 * earlier events.
 */
public class RMIEventQueue extends Thread
{
	public RMIEventQueue(RemoteEventReceiver receiver)
	{
		this.receiver = receiver;
		start();
	}
	public void dispose()
	{
		Enumeration e = m_obs.elements();
		while (e.hasMoreElements())
		{
			RMIObserver obs = (RMIObserver) e.nextElement();
			obs.stop();
		}
		interrupt(); // stop the event delivery thread, and allow it to clean up
	}
	public Observer setEventSource(RMIDestination destination)
	{
		RMIObserver obs = new RMIObserver(destination);
		m_obs.addElement(obs);
		return obs;
	}
	public void run()
	{
		try
		{
			for (;;)
			{
				boolean messageWaiting = false;
				synchronized (this)
				{
					Enumeration e = m_obs.elements();		
					while (e.hasMoreElements())
					{
						RMIObserver a = (RMIObserver) e.nextElement();
						messageWaiting = a.hasMessages();
						if (messageWaiting) break;
					}
					if (!messageWaiting) wait();
				}
				
				int n = 0;
				Enumeration e = m_obs.elements();		
				while (e.hasMoreElements())
				{
					RMIObserver a = (RMIObserver) e.nextElement();
					int i = a.getNUpdates();
					if (i>0) n += i+1;
				}

				if (n == 0) continue;
				else
				{
					Serializable[] s = new Serializable[n];
					int i = 0;

					e = m_obs.elements();
					while (e.hasMoreElements())
					{
						RMIObserver a = (RMIObserver) e.nextElement();
						Vector v = a.getUpdates();
						if (!v.isEmpty())
						{
							s[i++] = a.getDestination();
							Enumeration e2 = v.elements();
							while (e2.hasMoreElements())
							{
								s[i++] = (Serializable) e2.nextElement();
							}
						}	
					}
					receiver.remoteUpdate(s);
				}
			}
		}
		catch (InterruptedException e) 
		{
			System.out.println("RMIEventQueue Thread terminating");
			receiver = null;
		}
		catch (RemoteException e)
		{
			System.out.println("Error dispatching events "+e);
		}
		finally
		{
			receiver = null;
			m_obs = null;
		}
	}
	private RemoteEventReceiver receiver;
	private Vector m_obs = new Vector();

	private class RMIObserver implements Observer
	{
		RMIObserver(RMIDestination dest)
		{
			this.dest = dest;
		}
		RMIDestination getDestination()
		{
			return dest;
		}
		void stop()
		{
			stopped = true;
		}
		public void update(Observable obs, Object arg)
		{
			if (stopped) return;
			if (arg instanceof Serializable)
			{
				boolean wasEmpty = m_active.isEmpty();
				m_active.addElement(arg);
				if (wasEmpty) synchronized (RMIEventQueue.this) { RMIEventQueue.this.notifyAll(); }
			}
			else throw new RuntimeException("Argument for RMIObserver not serializable");
		}
		boolean hasMessages()
		{
			return !stopped && !m_active.isEmpty();
		}
		int getNUpdates() 
		{
			m_result.removeAllElements();
			if (m_active.isEmpty()) return 0;

			Vector v = m_active;
			m_queue.removeAllElements();
			m_active = m_queue; // Updates after this go to new queue;	
			m_queue = v;

			int size = v.size();

			outer: for (int i=0; i