Updates to add queuing for jmdns discovery.

2 files changed, 154 insertions, 1 deletions

diff --git a/providers/bundles/org.eclipse.ecf.provider.jmdns/src/org/eclipse/ecf/internal/provider/jmdns/SimpleFIFOQueue.java b/providers/bundles/org.eclipse.ecf.provider.jmdns/src/org/eclipse/ecf/internal/provider/jmdns/SimpleFIFOQueue.java
new file mode 100644
index 000000000..1b6a11e81
--- /dev/null
+++ b/providers/bundles/org.eclipse.ecf.provider.jmdns/src/org/eclipse/ecf/internal/provider/jmdns/SimpleFIFOQueue.java
@@ -0,0 +1,116 @@
+/****************************************************************************
+ * Copyright (c) 2007 Composent, Inc. and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    Composent, Inc. - initial API and implementation
+ *****************************************************************************/
+
+package org.eclipse.ecf.internal.provider.jmdns;
+
+import java.util.LinkedList;
+import java.util.List;
+
+public class SimpleFIFOQueue {
+	List list;
+
+	boolean stopped;
+
+	public SimpleFIFOQueue() {
+		list = new LinkedList();
+		stopped = false;
+	}
+
+	public synchronized boolean enqueue(Object obj) {
+		if (isStopped() || obj == null) {
+			return false;
+		}
+		// Add item to the list
+		list.add(obj);
+		// Notify waiting thread. Dequeue should only be read by one thread, so
+		// only need
+		// notify() rather than notifyAll().
+		notify();
+		return true;
+	}
+
+	public synchronized Object dequeue() {
+		Object val = peekQueue();
+		if (val != null) {
+			removeHead();
+		}
+		return val;
+	}
+
+	public synchronized Object peekQueue() {
+		while (isEmpty()) {
+			if (stopped)
+				return null;
+			try {
+				wait();
+			} catch (Exception e) {
+				return null;
+			}
+		}
+		return list.get(0);
+	}
+
+	public synchronized Object peekQueue(long waitMS) {
+		if (waitMS == 0)
+			return peekQueue();
+		if (stopped) {
+			return null;
+		}
+		try {
+			wait(waitMS);
+		} catch (Exception e) {
+			return null;
+		}
+		if (isEmpty())
+			return null;
+		return list.get(0);
+	}
+
+	public synchronized Object removeHead() {
+		if (list.isEmpty())
+			return null;
+		return list.remove(0);
+	}
+
+	public synchronized boolean isEmpty() {
+		return list.isEmpty();
+	}
+
+	public synchronized void stop() {
+		stopped = true;
+	}
+
+	public synchronized boolean isStopped() {
+		return stopped;
+	}
+
+	public synchronized int size() {
+		return list.size();
+	}
+
+	public synchronized Object[] flush() {
+		Object[] out = list.toArray();
+		list.clear();
+		close();
+		return out;
+	}
+
+	public synchronized void close() {
+		stop();
+		notifyAll();
+	}
+
+	public String toString() {
+		StringBuffer sb = new StringBuffer("SimpleFIFOQueue["); //$NON-NLS-1$
+		sb.append(list).append("]"); //$NON-NLS-1$
+		return sb.toString();
+	}
+}
diff --git a/providers/bundles/org.eclipse.ecf.provider.jmdns/src/org/eclipse/ecf/provider/jmdns/container/JMDNSDiscoveryContainer.java b/providers/bundles/org.eclipse.ecf.provider.jmdns/src/org/eclipse/ecf/provider/jmdns/container/JMDNSDiscoveryContainer.java
index f2e1a03f0..d1d3ecc13 100644
--- a/providers/bundles/org.eclipse.ecf.provider.jmdns/src/org/eclipse/ecf/provider/jmdns/container/JMDNSDiscoveryContainer.java
+++ b/providers/bundles/org.eclipse.ecf.provider.jmdns/src/org/eclipse/ecf/provider/jmdns/container/JMDNSDiscoveryContainer.java
@@ -42,6 +42,9 @@ public class JMDNSDiscoveryContainer extends AbstractDiscoveryContainerAdapter i
 	boolean disposed = false;
 	Object lock = new Object();
 
+	SimpleFIFOQueue queue = null;
+	Thread notificationThread = null;
+
 	public JMDNSDiscoveryContainer(InetAddress addr) throws IDCreateException {
 		super(JMDNSNamespace.NAME, new DiscoveryContainerConfig(IDFactory.getDefault().createStringID(JMDNSDiscoveryContainer.class.getName() + ";" + addr.toString() + ";" + instanceCount++))); //$NON-NLS-1$  //$NON-NLS-2$
 		intf = addr;
@@ -78,6 +81,7 @@ public class JMDNSDiscoveryContainer extends AbstractDiscoveryContainerAdapter i
 				throw new ContainerConnectException(Messages.JMDNSDiscoveryContainer_EXCEPTION_ALREADY_CONNECTED);
 			this.targetID = (targetID1 == null) ? getConfig().getID() : targetID1;
 			fireContainerEvent(new ContainerConnectingEvent(this.getID(), this.targetID, joinContext));
+			initializeQueue();
 			try {
 				this.jmdns = new JmDNS(intf);
 				jmdns.addServiceTypeListener(this);
@@ -92,6 +96,31 @@ public class JMDNSDiscoveryContainer extends AbstractDiscoveryContainerAdapter i
 		}
 	}
 
+	private void initializeQueue() {
+		queue = new SimpleFIFOQueue();
+		notificationThread = new Thread(new Runnable() {
+			public void run() {
+				for (;;) {
+					if (Thread.currentThread().isInterrupted())
+						break;
+					Runnable runnable = (Runnable) queue.dequeue();
+					if (Thread.currentThread().isInterrupted() || runnable == null)
+						break;
+					try {
+						runnable.run();
+					} catch (Throwable t) {
+						handleRuntimeException(t);
+					}
+				}
+			}
+		}, "JMDNS Discovery Thread"); //$NON-NLS-1$
+		notificationThread.start();
+	}
+
+	protected void handleRuntimeException(Throwable t) {
+		// Nothing to do except log
+	}
+
 	/* (non-Javadoc)
 	 * @see org.eclipse.ecf.core.IContainer#disconnect()
 	 */
@@ -102,6 +131,9 @@ public class JMDNSDiscoveryContainer extends AbstractDiscoveryContainerAdapter i
 				fireContainerEvent(new ContainerDisconnectingEvent(this.getID(), connectedID));
 				jmdns.close();
 				jmdns = null;
+				queue.close();
+				queue = null;
+				notificationThread = null;
 				this.targetID = null;
 				fireContainerEvent(new ContainerDisconnectedEvent(this.getID(), connectedID));
 			}
@@ -204,7 +236,12 @@ public class JMDNSDiscoveryContainer extends AbstractDiscoveryContainerAdapter i
 	/**************************** JMDNS listeners ***********************************/
 
 	private void runInThread(Runnable runnable) {
-		new Thread(runnable).start();
+		SimpleFIFOQueue localQueue = null;
+		synchronized (lock) {
+			localQueue = queue;
+		}
+		if (localQueue != null)
+			localQueue.enqueue(runnable);
 	}
 
 	/* (non-Javadoc)