SMACK-341 Updated collectors to use concurrent classes.

git-svn-id: http://svn.igniterealtime.org/svn/repos/smack/trunk@13452 b35dd754-fafc-0310-a699-88a17e54d16e

source/org/jivesoftware/smack/PacketCollector.java

@@ -20,11 +20,12 @@
 
 package org.jivesoftware.smack;
 
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
 import org.jivesoftware.smack.filter.PacketFilter;
 import org.jivesoftware.smack.packet.Packet;
 
-import java.util.LinkedList;
-
 /**
  * Provides a mechanism to collect packets into a result queue that pass a
  * specified filter. The collector lets you perform blocking and polling
@@ -41,16 +42,9 @@ import java.util.LinkedList;
  */
 public class PacketCollector {
 
-    /**
-     * Max number of packets that any one collector can hold. After the max is
-     * reached, older packets will be automatically dropped from the queue as
-     * new packets are added.
-     */
-    private int maxPackets = SmackConfiguration.getPacketCollectorSize();
-
     private PacketFilter packetFilter;
-    private LinkedList<Packet> resultQueue;
+    private ArrayBlockingQueue<Packet> resultQueue;
-    private Connection conection;
+    private Connection connection;
     private boolean cancelled = false;
 
     /**
@@ -61,9 +55,7 @@ public class PacketCollector {
      * @param packetFilter determines which packets will be returned by this collector.
      */
     protected PacketCollector(Connection conection, PacketFilter packetFilter) {
-        this.conection = conection;
+    	this(conection, packetFilter, SmackConfiguration.getPacketCollectorSize());
-        this.packetFilter = packetFilter;
-        this.resultQueue = new LinkedList<Packet>();
     }
 
     /**
@@ -75,8 +67,9 @@ public class PacketCollector {
      * @param maxSize the maximum number of packets that will be stored in the collector.
      */
     protected PacketCollector(Connection conection, PacketFilter packetFilter, int maxSize) {
-        this(conection, packetFilter);
+        this.connection = conection;
-        maxPackets = maxSize;
+        this.packetFilter = packetFilter;
+        this.resultQueue = new ArrayBlockingQueue<Packet>(maxSize);
     }
 
     /**
@@ -88,7 +81,7 @@ public class PacketCollector {
         // If the packet collector has already been cancelled, do nothing.
         if (!cancelled) {
             cancelled = true;
-            conection.removePacketCollector(this);
+            connection.removePacketCollector(this);
         }
     }
 
@@ -110,13 +103,8 @@ public class PacketCollector {
      * @return the next packet result, or <tt>null</tt> if there are no more
      *      results.
      */
-    public synchronized Packet pollResult() {
+    public Packet pollResult() {
-        if (resultQueue.isEmpty()) {
+    	return resultQueue.poll();
-            return null;
-        }
-        else {
-            return resultQueue.removeLast();
-        }
     }
 
     /**
@@ -125,18 +113,14 @@ public class PacketCollector {
      *
      * @return the next available packet.
      */
-    public synchronized Packet nextResult() {
+    public Packet nextResult() {
-        // Wait indefinitely until there is a result to return.
-        while (resultQueue.isEmpty()) {
         try {
-                wait();
+			return resultQueue.take();
 		}
-            catch (InterruptedException ie) {
+		catch (InterruptedException e) {
-                // Ignore.
+			throw new RuntimeException(e);
 		}
     }
-        return resultQueue.removeLast();
-    }
 
     /**
      * Returns the next available packet. The method call will block (not return)
@@ -146,39 +130,12 @@ public class PacketCollector {
      * @param timeout the amount of time to wait for the next packet (in milleseconds).
      * @return the next available packet.
      */
-    public synchronized Packet nextResult(long timeout) {
+    public Packet nextResult(long timeout) {
-        // Wait up to the specified amount of time for a result.
-        if (resultQueue.isEmpty()) {
-            long waitTime = timeout;
-            long start = System.currentTimeMillis();
     	try {
-                // Keep waiting until the specified amount of time has elapsed, or
+			return resultQueue.poll(timeout, TimeUnit.MILLISECONDS);
-                // a packet is available to return.
-                while (resultQueue.isEmpty()) {
-                    if (waitTime <= 0) {
-                        break;
 		}
-                    wait(waitTime);
+		catch (InterruptedException e) {
-                    long now = System.currentTimeMillis();
+			throw new RuntimeException(e);
-                    waitTime -= (now - start);
-                    start = now;
-                }
-            }
-            catch (InterruptedException ie) {
-                // Ignore.
-            }
-            // Still haven't found a result, so return null.
-            if (resultQueue.isEmpty()) {
-                return null;
-            }
-            // Return the packet that was found.
-            else {
-                return resultQueue.removeLast();
-            }
-        }
-        // There's already a packet waiting, so return it.
-        else {
-            return resultQueue.removeLast();
 		}
     }
 
@@ -188,19 +145,16 @@ public class PacketCollector {
      *
      * @param packet the packet to process.
      */
-    protected synchronized void processPacket(Packet packet) {
+    protected void processPacket(Packet packet) {
         if (packet == null) {
             return;
         }
+        
         if (packetFilter == null || packetFilter.accept(packet)) {
-            // If the max number of packets has been reached, remove the oldest one.
+        	while (!resultQueue.offer(packet)) {
-            if (resultQueue.size() == maxPackets) {
+        		// Since we know the queue is full, this poll should never actually block.
-                resultQueue.removeLast();
+        		resultQueue.poll();
         	}
-            // Add the new packet.
-            resultQueue.addFirst(packet);
-            // Notify waiting threads a result is available.
-            notifyAll();
         }
     }
 }

source/org/jivesoftware/smackx/muc/ConnectionDetachedPacketCollector.java

@@ -20,11 +20,12 @@
 
 package org.jivesoftware.smackx.muc;
 
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
 import org.jivesoftware.smack.SmackConfiguration;
 import org.jivesoftware.smack.packet.Packet;
 
-import java.util.LinkedList;
-
 /**
  * A variant of the {@link org.jivesoftware.smack.PacketCollector} class
  * that does not force attachment to a <code>Connection</code>
@@ -41,14 +42,14 @@ class ConnectionDetachedPacketCollector {
      */
     private int maxPackets = SmackConfiguration.getPacketCollectorSize();
 
-    private LinkedList<Packet> resultQueue;
+    private ArrayBlockingQueue<Packet> resultQueue;
 
     /**
      * Creates a new packet collector. If the packet filter is <tt>null</tt>, then
      * all packets will match this collector.
      */
     public ConnectionDetachedPacketCollector() {
-        this.resultQueue = new LinkedList<Packet>();
+        this(SmackConfiguration.getPacketCollectorSize());
     }
 
     /**
@@ -56,8 +57,7 @@ class ConnectionDetachedPacketCollector {
      * all packets will match this collector.
      */
     public ConnectionDetachedPacketCollector(int maxSize) {
-        this.resultQueue = new LinkedList<Packet>();
+        this.resultQueue = new ArrayBlockingQueue<Packet>(maxSize);
-        maxPackets = maxSize;
     }
 
     /**
@@ -68,13 +68,8 @@ class ConnectionDetachedPacketCollector {
      * @return the next packet result, or <tt>null</tt> if there are no more
      *      results.
      */
-    public synchronized Packet pollResult() {
+    public Packet pollResult() {
-        if (resultQueue.isEmpty()) {
+    	return resultQueue.poll();
-            return null;
-        }
-        else {
-            return resultQueue.removeLast();
-        }
     }
 
     /**
@@ -83,18 +78,14 @@ class ConnectionDetachedPacketCollector {
      *
      * @return the next available packet.
      */
-    public synchronized Packet nextResult() {
+    public Packet nextResult() {
-        // Wait indefinitely until there is a result to return.
-        while (resultQueue.isEmpty()) {
         try {
-                wait();
+			return resultQueue.take();
 		}
-            catch (InterruptedException ie) {
+		catch (InterruptedException e) {
-                // Ignore.
+			throw new RuntimeException(e);
 		}
     }
-        return resultQueue.removeLast();
-    }
 
     /**
      * Returns the next available packet. The method call will block (not return)
@@ -104,22 +95,12 @@ class ConnectionDetachedPacketCollector {
      * @param timeout the amount of time to wait for the next packet (in milleseconds).
      * @return the next available packet.
      */
-    public synchronized Packet nextResult(long timeout) {
+    public Packet nextResult(long timeout) {
-        // Wait up to the specified amount of time for a result.
-        if (resultQueue.isEmpty()) {
         try {
-                wait(timeout);
+        	return resultQueue.poll(timeout, TimeUnit.MILLISECONDS);
 		}
-            catch (InterruptedException ie) {
+		catch (InterruptedException e) {
-                // Ignore.
+			throw new RuntimeException(e);
-            }
-        }
-        // If still no result, return null.
-        if (resultQueue.isEmpty()) {
-            return null;
-        }
-        else {
-            return resultQueue.removeLast();
 		}
     }
 
@@ -129,17 +110,14 @@ class ConnectionDetachedPacketCollector {
      *
      * @param packet the packet to process.
      */
-    protected synchronized void processPacket(Packet packet) {
+    protected void processPacket(Packet packet) {
         if (packet == null) {
             return;
         }
-        // If the max number of packets has been reached, remove the oldest one.
+        
-        if (resultQueue.size() == maxPackets) {
+    	while (!resultQueue.offer(packet)) {
-            resultQueue.removeLast();
+    		// Since we know the queue is full, this poll should never actually block.
+    		resultQueue.poll();
     	}
-        // Add the new packet.
-        resultQueue.addFirst(packet);
-        // Notify waiting threads a result is available.
-        notifyAll();
     }
 }

test-unit/org/jivesoftware/smack/PacketCollectorTest.java

@@ -0,0 +1,198 @@
+package org.jivesoftware.smack;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import org.jivesoftware.smack.filter.PacketFilter;
+import org.jivesoftware.smack.packet.Packet;
+import org.junit.Test;
+
+public class PacketCollectorTest
+{
+
+	@Test
+	public void verifyRollover()
+	{
+		TestPacketCollector collector = new TestPacketCollector(null, new OKEverything(), 5);
+		
+		for (int i=0; i<6; i++)
+		{
+			Packet testPacket = new TestPacket(i);
+			collector.processPacket(testPacket);
+		}
+		
+		// Assert that '0' has rolled off
+		assertEquals("1", collector.nextResult().getPacketID());
+		assertEquals("2", collector.nextResult().getPacketID());
+		assertEquals("3", collector.nextResult().getPacketID());
+		assertEquals("4", collector.nextResult().getPacketID());
+		assertEquals("5", collector.pollResult().getPacketID());
+		assertNull(collector.pollResult());
+		
+		for (int i=10; i<15; i++)
+		{
+			Packet testPacket = new TestPacket(i);
+			collector.processPacket(testPacket);
+		}
+		
+		assertEquals("10", collector.nextResult().getPacketID());
+		assertEquals("11", collector.nextResult().getPacketID());
+		assertEquals("12", collector.nextResult().getPacketID());
+		assertEquals("13", collector.nextResult().getPacketID());
+		assertEquals("14", collector.pollResult().getPacketID());
+		assertNull(collector.pollResult());
+		
+		assertNull(collector.nextResult(1000));
+	}
+
+	/**
+	 * Although this doesn't guarentee anything due to the nature of threading, it can 
+	 * potentially catch problems.
+	 */
+	@Test
+	public void verifyThreadSafety()
+	{
+		int insertCount = 500;
+		final TestPacketCollector collector = new TestPacketCollector(null, new OKEverything(), insertCount);
+		
+		Thread consumer1 = new Thread(new Runnable()
+		{
+			@Override
+			public void run()
+			{
+				try
+				{
+					while (true)
+					{
+						try
+						{
+							Thread.sleep(3);
+						}
+						catch (InterruptedException e)
+						{
+						}
+						Packet packet = collector.nextResult();
+//						System.out.println(Thread.currentThread().getName() + "  packet: " + packet);
+					}
+				}
+				catch (RuntimeException re)
+				{
+					if (re.getCause() instanceof InterruptedException)
+					{
+//						System.out.println(Thread.currentThread().getName() + " has been interupted");
+					}
+				}
+			}
+		});
+		consumer1.setName("consumer 1");
+
+		Thread consumer2 = new Thread(new Runnable()
+		{
+			@Override
+			public void run()
+			{
+				Packet p = null;
+				
+				do
+				{
+					try
+					{
+						Thread.sleep(3);
+					}
+					catch (InterruptedException e)
+					{
+					}
+					p = collector.nextResult(1);
+//					System.out.println(Thread.currentThread().getName() + "  packet: " + p);
+				}
+				while (p != null);
+			}
+		});
+		consumer2.setName("consumer 2");
+
+		Thread consumer3 = new Thread(new Runnable()
+		{
+			@Override
+			public void run()
+			{
+				Packet p = null;
+				
+				do
+				{
+					try
+					{
+						Thread.sleep(3);
+					}
+					catch (InterruptedException e)
+					{
+					}
+					p = collector.pollResult();
+//					System.out.println(Thread.currentThread().getName() + "  packet: " + p);
+				}
+				while (p != null);
+			}
+		});
+		consumer3.setName("consumer 3");
+
+		consumer1.start();
+		consumer2.start();
+		consumer3.start();
+		
+		for(int i=0; i<insertCount; i++)
+		{
+			collector.processPacket(new TestPacket(i));
+		}
+
+		try
+		{
+			Thread.sleep(5000);
+			consumer3.join();
+			consumer2.join();
+			consumer1.interrupt();
+		}
+		catch (InterruptedException e)
+		{
+		}
+		//We cannot guarantee that this is going to pass due to the possible issue of timing between consumer 1 
+		// and main, but the probability is extremely remote.
+		assertNull(collector.pollResult());
+	}
+
+	class OKEverything implements PacketFilter
+	{
+		@Override
+		public boolean accept(Packet packet)
+		{
+			return true;
+		}
+		
+	}
+	
+	class TestPacketCollector extends PacketCollector 
+	{
+		protected TestPacketCollector(Connection conection, PacketFilter packetFilter, int size)
+		{
+			super(conection, packetFilter, size);
+		}
+	}
+
+	class TestPacket extends Packet
+	{
+		public TestPacket(int i)
+		{
+			setPacketID(String.valueOf(i));
+		}
+
+		@Override
+		public String toString()
+		{
+			return toXML();
+		}
+
+		@Override
+		public String toXML()
+		{
+			return "<packetId>" + getPacketID() + "</packetId>";
+		}
+	}
+}

test-unit/org/jivesoftware/smackx/muc/ConnectionDetachedPacketCollectorTest.java

@@ -0,0 +1,179 @@
+package org.jivesoftware.smackx.muc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import org.jivesoftware.smack.packet.Packet;
+import org.junit.Test;
+
+public class ConnectionDetachedPacketCollectorTest
+{
+
+	@Test
+	public void verifyRollover()
+	{
+		ConnectionDetachedPacketCollector collector = new ConnectionDetachedPacketCollector(5);
+		
+		for (int i=0; i<6; i++)
+		{
+			Packet testPacket = new TestPacket(i);
+			collector.processPacket(testPacket);
+		}
+		
+		// Assert that '0' has rolled off
+		assertEquals("1", collector.nextResult().getPacketID());
+		assertEquals("2", collector.nextResult().getPacketID());
+		assertEquals("3", collector.nextResult().getPacketID());
+		assertEquals("4", collector.nextResult().getPacketID());
+		assertEquals("5", collector.pollResult().getPacketID());
+		assertNull(collector.pollResult());
+		
+		for (int i=10; i<15; i++)
+		{
+			Packet testPacket = new TestPacket(i);
+			collector.processPacket(testPacket);
+		}
+		
+		assertEquals("10", collector.nextResult().getPacketID());
+		assertEquals("11", collector.nextResult().getPacketID());
+		assertEquals("12", collector.nextResult().getPacketID());
+		assertEquals("13", collector.nextResult().getPacketID());
+		assertEquals("14", collector.pollResult().getPacketID());
+		assertNull(collector.pollResult());
+		
+		assertNull(collector.nextResult(1000));
+	}
+
+	/**
+	 * Although this doesn't guarentee anything due to the nature of threading, it can 
+	 * potentially catch problems.
+	 */
+	@Test
+	public void verifyThreadSafety()
+	{
+		int insertCount = 500;
+		final ConnectionDetachedPacketCollector collector = new ConnectionDetachedPacketCollector(insertCount);
+		
+		Thread consumer1 = new Thread(new Runnable()
+		{
+			@Override
+			public void run()
+			{
+				try
+				{
+					while (true)
+					{
+						try
+						{
+							Thread.sleep(3);
+						}
+						catch (InterruptedException e)
+						{
+						}
+						Packet packet = collector.nextResult();
+//						System.out.println(Thread.currentThread().getName() + "  packet: " + packet);
+					}
+				}
+				catch (RuntimeException re)
+				{
+					if (re.getCause() instanceof InterruptedException)
+					{
+//						System.out.println(Thread.currentThread().getName() + " has been interupted");
+					}
+				}
+			}
+		});
+		consumer1.setName("consumer 1");
+
+		Thread consumer2 = new Thread(new Runnable()
+		{
+			@Override
+			public void run()
+			{
+				Packet p = null;
+				
+				do
+				{
+					try
+					{
+						Thread.sleep(3);
+					}
+					catch (InterruptedException e)
+					{
+					}
+					p = collector.nextResult(1);
+//					System.out.println(Thread.currentThread().getName() + "  packet: " + p);
+				}
+				while (p != null);
+			}
+		});
+		consumer2.setName("consumer 2");
+
+		Thread consumer3 = new Thread(new Runnable()
+		{
+			@Override
+			public void run()
+			{
+				Packet p = null;
+				
+				do
+				{
+					try
+					{
+						Thread.sleep(3);
+					}
+					catch (InterruptedException e)
+					{
+					}
+					p = collector.pollResult();
+//					System.out.println(Thread.currentThread().getName() + "  packet: " + p);
+				}
+				while (p != null);
+			}
+		});
+		consumer3.setName("consumer 3");
+
+		consumer1.start();
+		consumer2.start();
+		consumer3.start();
+		
+		for(int i=0; i<insertCount; i++)
+		{
+			collector.processPacket(new TestPacket(i));
+		}
+
+		try
+		{
+			Thread.sleep(5000);
+			consumer3.join();
+			consumer2.join();
+			consumer1.interrupt();
+		}
+		catch (InterruptedException e)
+		{
+		}
+		//We cannot guarantee that this is going to pass due to the possible issue of timing between consumer 1 
+		// and main, but the probability is extremely remote.
+		assertNull(collector.pollResult());
+	}
+	
+	class TestPacket extends Packet
+	{
+		public TestPacket(int i)
+		{
+			setPacketID(String.valueOf(i));
+		}
+
+		@Override
+		public String toString()
+		{
+			return toXML();
+		}
+
+		@Override
+		public String toXML()
+		{
+			return "<packetId>" + getPacketID() + "</packetId>";
+		}
+	}
+}