SMACK-412 Abstracted the keepalive implementation and set the thread to start and stop on demand.

git-svn-id: http://svn.igniterealtime.org/svn/repos/smack/branches/smack_3_3_0@13610 b35dd754-fafc-0310-a699-88a17e54d16e

build/resources/META-INF/smack-config.xml

@@ -24,7 +24,7 @@
     <startupClasses>
         <className>org.jivesoftware.smackx.ServiceDiscoveryManager</className>
         <className>org.jivesoftware.smack.PrivacyListManager</className>        
-        <className>org.jivesoftware.smack.ping.ServerPingManager</className>        
+        <className>org.jivesoftware.smack.keepalive.KeepAliveManager</className>        
         <className>org.jivesoftware.smackx.XHTMLManager</className>
         <className>org.jivesoftware.smackx.muc.MultiUserChat</className>
         <className>org.jivesoftware.smackx.bytestreams.ibb.InBandBytestreamManager</className>
@@ -34,7 +34,7 @@
         <className>org.jivesoftware.smack.ReconnectionManager</className>    
         <className>org.jivesoftware.smackx.commands.AdHocCommandManager</className>    
         <className>org.jivesoftware.smack.util.dns.JavaxResolver</className>
+        <className>org.jivesoftware.smackx.ping.PingManager</className>
     </startupClasses>
 
-
 </smack>

source/org/jivesoftware/smack/keepalive/KeepAliveManager.java

@@ -14,13 +14,14 @@
  * limitations under the License.
  */
 
-package org.jivesoftware.smack.ping;
+package org.jivesoftware.smack.keepalive;
 
 import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
-import java.util.WeakHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -33,17 +34,11 @@ import org.jivesoftware.smack.ConnectionListener;
 import org.jivesoftware.smack.PacketCollector;
 import org.jivesoftware.smack.PacketListener;
 import org.jivesoftware.smack.SmackConfiguration;
-import org.jivesoftware.smack.XMPPException;
-import org.jivesoftware.smack.filter.AndFilter;
-import org.jivesoftware.smack.filter.IQTypeFilter;
 import org.jivesoftware.smack.filter.PacketFilter;
 import org.jivesoftware.smack.filter.PacketIDFilter;
-import org.jivesoftware.smack.filter.PacketTypeFilter;
-import org.jivesoftware.smack.packet.IQ;
-import org.jivesoftware.smack.packet.IQ.Type;
 import org.jivesoftware.smack.packet.Packet;
+import org.jivesoftware.smack.ping.PingFailedListener;
 import org.jivesoftware.smack.ping.packet.Ping;
-import org.jivesoftware.smackx.ServiceDiscoveryManager;
 
 /**
  * Using an implementation of <a href="http://www.xmpp.org/extensions/xep-0199.html">XMPP Ping (XEP-0199)</a>. This
@@ -56,25 +51,15 @@ import org.jivesoftware.smackx.ServiceDiscoveryManager;
  * 
  * @author Florian Schmaus
  */
-public class ServerPingManager {
+public class KeepAliveManager {
-    private static Map<Connection, ServerPingManager> instances = Collections
+    private static Map<Connection, KeepAliveManager> instances = new HashMap<Connection, KeepAliveManager>();
-            .synchronizedMap(new WeakHashMap<Connection, ServerPingManager>());
+    private static volatile ScheduledExecutorService periodicPingExecutorService;
-    private static long defaultPingInterval = SmackConfiguration.getKeepAliveInterval(); 
-    
-    private static ScheduledExecutorService periodicPingExecutorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
-        @Override
-        public Thread newThread(Runnable runnable) {
-            Thread pingThread = new Thread(runnable, "Smack Server Ping");
-            pingThread.setDaemon(true);
-            return pingThread;
-        }
-    });
     
     static {
-        if (defaultPingInterval > 0) {
+        if (SmackConfiguration.getKeepAliveInterval() > 0) {
             Connection.addConnectionCreationListener(new ConnectionCreationListener() {
                 public void connectionCreated(Connection connection) {
-                    new ServerPingManager(connection);
+                    new KeepAliveManager(connection);
                 }
             });
         }
@@ -87,56 +72,94 @@ public class ServerPingManager {
     private volatile long lastSuccessfulContact = -1;
 
     /**
-     * Retrieves a {@link ServerPingManager} for the specified {@link Connection}, creating one if it doesn't already
+     * Retrieves a {@link KeepAliveManager} for the specified {@link Connection}, creating one if it doesn't already
      * exist.
      * 
      * @param connection
      * The connection the manager is attached to.
      * @return The new or existing manager.
      */
-    public synchronized static ServerPingManager getInstanceFor(Connection connection) {
+    public synchronized static KeepAliveManager getInstanceFor(Connection connection) {
-        ServerPingManager pingManager = instances.get(connection);
+        KeepAliveManager pingManager = instances.get(connection);
 
         if (pingManager == null) {
-            pingManager = new ServerPingManager(connection);
+            pingManager = new KeepAliveManager(connection);
+            instances.put(connection, pingManager);
         }
         return pingManager;
     }
 
-    private ServerPingManager(Connection connection) {
+    /*
-        ServiceDiscoveryManager sdm = ServiceDiscoveryManager.getInstanceFor(connection);
+     * Start the executor service if it hasn't been started yet.
-        sdm.addFeature(Ping.NAMESPACE);
+     */
+    private synchronized static void enableExecutorService() {
+        if (periodicPingExecutorService == null) {
+            periodicPingExecutorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
+                @Override
+                public Thread newThread(Runnable runnable) {
+                    Thread pingThread = new Thread(runnable, "Smack Keepalive");
+                    pingThread.setDaemon(true);
+                    return pingThread;
+                }
+            });
+        }
+    }
+
+    /*
+     * Stop the executor service if all monitored connections are disconnected.
+     */
+    private synchronized static void handleDisconnect(Connection con) {
+        if (periodicPingExecutorService != null) {
+            instances.remove(con);
+            
+            if (instances.isEmpty()) {
+                periodicPingExecutorService.shutdownNow();
+                periodicPingExecutorService = null;
+            }
+        }
+    }
+    
+    private KeepAliveManager(Connection connection) {
         this.connection = connection;
         init();
+        handleConnect();
+    }
+
+    /*
+     * Call after every connection to add the packet listener.
+     */
+    private void handleConnect() {
+        // Listen for all incoming packets and reset the scheduled ping whenever
+        // one arrives.
+        connection.addPacketListener(new PacketListener() {
+
+            @Override
+            public void processPacket(Packet packet) {
+                // reschedule the ping based on this last server contact
+                lastSuccessfulContact = System.currentTimeMillis();
+                schedulePingServerTask();
+            }
+        }, null);
     }
 
     private void init() {
-        PacketFilter pingPacketFilter = new AndFilter(new PacketTypeFilter(Ping.class), new IQTypeFilter(Type.GET));
-        
-        connection.addPacketListener(new PacketListener() {
-            /**
-             * Sends a Pong for every Ping
-             */
-            public void processPacket(Packet packet) {
-                IQ pong = IQ.createResultIQ((Ping) packet);
-                connection.sendPacket(pong);
-            }
-        }, pingPacketFilter);
-
         connection.addConnectionListener(new ConnectionListener() {
 
             @Override
             public void connectionClosed() {
                 stopPingServerTask();
+                handleDisconnect(connection);
             }
 
             @Override
             public void connectionClosedOnError(Exception arg0) {
                 stopPingServerTask();
+                handleDisconnect(connection);
             }
 
             @Override
             public void reconnectionSuccessful() {
+                handleConnect();
                 schedulePingServerTask();
             }
 
@@ -149,17 +172,6 @@ public class ServerPingManager {
             }
         });
 
-        // Listen for all incoming packets and reset the scheduled ping whenever
-        // one arrives.
-        connection.addPacketListener(new PacketListener() {
-
-            @Override
-            public void processPacket(Packet packet) {
-                // reschedule the ping based on this last server contact
-                lastSuccessfulContact = System.currentTimeMillis();
-                schedulePingServerTask();
-            }
-        }, null);
         instances.put(connection, this);
         schedulePingServerTask();
     }
@@ -171,15 +183,20 @@ public class ServerPingManager {
      * The new ping time interval in milliseconds.
      */
     public void setPingInterval(long newPingInterval) {
-        if (pingInterval != newPingInterval) {
+        if (pingInterval == newPingInterval) 
-            pingInterval = newPingInterval;
+            return;
 
-            if (pingInterval < 0) {
+        // Enable the executor service
-                stopPinging();
+        if (newPingInterval > 0)
-            }
+            enableExecutorService();
-            else {
+        
-                schedulePingServerTask();
+        pingInterval = newPingInterval;
-            }
+            
+        if (pingInterval < 0) {
+            stopPinging();
+        }
+        else {
+            schedulePingServerTask();
         }
     }
 
@@ -227,12 +244,20 @@ public class ServerPingManager {
     }
 
     /**
-     * Returns the time of the last successful contact with the server. (i.e. the last time any message was received).
+     * Returns the elapsed time (in milliseconds) since the last successful contact with the server 
+     * (i.e. the last time any message was received).
+     * <p>
+     * <b>Note</b>: Result is -1 if no message has been received since manager was created and 
+     * 0 if the elapsed time is negative due to a clock reset. 
      * 
-     * @return Time of last message or -1 if none has been received since manager was created.
+     * @return Elapsed time since last message was received.  
      */
-    public long getLastSuccessfulContact() {
+    public long getTimeSinceLastContact() {
-        return lastSuccessfulContact;
+        if (lastSuccessfulContact == -1)
+            return lastSuccessfulContact;
+        long delta = System.currentTimeMillis() - lastSuccessfulContact;
+        
+        return (delta < 0) ? 0 : delta;
     }
 
     /**
@@ -242,6 +267,7 @@ public class ServerPingManager {
      * This is designed so only one executor is used for scheduling all pings on all connections.  This results in only 1 thread used for pinging.
      */
     private synchronized void schedulePingServerTask() {
+        enableExecutorService();
         stopPingServerTask();
         
         if (pingInterval > 0) {

source/org/jivesoftware/smackx/ping/PingManager.java

@@ -16,11 +16,25 @@
 
 package org.jivesoftware.smackx.ping;
 
+import java.util.Collections;
+import java.util.Map;
+import java.util.WeakHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+
 import org.jivesoftware.smack.Connection;
+import org.jivesoftware.smack.ConnectionCreationListener;
+import org.jivesoftware.smack.PacketListener;
 import org.jivesoftware.smack.SmackConfiguration;
 import org.jivesoftware.smack.SmackError;
 import org.jivesoftware.smack.XMPPException;
-import org.jivesoftware.smack.ping.ServerPingManager;
+import org.jivesoftware.smack.filter.AndFilter;
+import org.jivesoftware.smack.filter.IQTypeFilter;
+import org.jivesoftware.smack.filter.PacketFilter;
+import org.jivesoftware.smack.filter.PacketTypeFilter;
+import org.jivesoftware.smack.keepalive.KeepAliveManager;
+import org.jivesoftware.smack.packet.IQ;
+import org.jivesoftware.smack.packet.Packet;
+import org.jivesoftware.smack.packet.IQ.Type;
 import org.jivesoftware.smack.ping.packet.Ping;
 import org.jivesoftware.smack.util.SyncPacketSend;
 import org.jivesoftware.smackx.ServiceDiscoveryManager;
@@ -31,7 +45,7 @@ import org.jivesoftware.smackx.packet.DiscoverInfo;
  * allows one entity to 'ping' any other entity by simply sending a ping to 
  * the appropriate JID.
  * <p>
- * NOTE: The {@link ServerPingManager} already provides a keepalive functionality 
+ * NOTE: The {@link KeepAliveManager} already provides a keepalive functionality 
  * for regularly pinging the server to keep the underlying transport connection
  * alive.  This class is specifically intended to do manual pings of other 
  * entities.  
@@ -41,12 +55,57 @@ import org.jivesoftware.smackx.packet.DiscoverInfo;
  *      Ping</a>
  */
 public class PingManager {
+    private static Map<Connection, PingManager> instances = Collections
+            .synchronizedMap(new WeakHashMap<Connection, PingManager>());
+    
+    static {
+        Connection.addConnectionCreationListener(new ConnectionCreationListener() {
+            public void connectionCreated(Connection connection) {
+                new PingManager(connection);
+            }
+        });
+    }
+
     private Connection connection;
 
-    public PingManager(Connection connection) {
+    /**
+     * Retrieves a {@link PingManager} for the specified {@link Connection}, creating one if it doesn't already
+     * exist.
+     * 
+     * @param connection
+     * The connection the manager is attached to.
+     * @return The new or existing manager.
+     */
+    public synchronized static PingManager getInstanceFor(Connection connection) {
+        PingManager pingManager = instances.get(connection);
+
+        if (pingManager == null) {
+            pingManager = new PingManager(connection);
+        }
+        return pingManager;
+    }
+
+    private PingManager(Connection con) {
+        this.connection = con;
         ServiceDiscoveryManager sdm = ServiceDiscoveryManager.getInstanceFor(connection);
+        
+        // The ServiceDiscoveryManager was not pre-initialized
+        if (sdm == null)
+            sdm = new ServiceDiscoveryManager(connection);
+        
         sdm.addFeature(Ping.NAMESPACE);
-        this.connection = connection;
+        
+        PacketFilter pingPacketFilter = new AndFilter(new PacketTypeFilter(Ping.class), new IQTypeFilter(Type.GET));
+        
+        connection.addPacketListener(new PacketListener() {
+            /**
+             * Sends a Pong for every Ping
+             */
+            public void processPacket(Packet packet) {
+                IQ pong = IQ.createResultIQ((Ping) packet);
+                connection.sendPacket(pong);
+            }
+        }, pingPacketFilter);
     }
 
     /**

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

@@ -49,6 +49,7 @@ public class DummyConnection extends Connection {
 
     private boolean authenticated = false;
     private boolean anonymous = false;
+    private boolean reconnect = false;
 
     private String user;
     private String connectionID;
@@ -71,6 +72,12 @@ public class DummyConnection extends Connection {
     @Override
     public void connect() throws XMPPException {
         connectionID = "dummy-" + new Random(new Date().getTime()).nextInt();
+
+        if (reconnect) {
+            for (ConnectionListener listener : getConnectionListeners()) {
+                listener.reconnectionSuccessful();
+            }
+        }
     }
 
     @Override
@@ -80,6 +87,11 @@ public class DummyConnection extends Connection {
         roster = null;
         authenticated = false;
         anonymous = false;
+        
+        for (ConnectionListener listener : getConnectionListeners()) {
+            listener.connectionClosed();
+        }
+        reconnect = true;
     }
 
     @Override

test-unit/org/jivesoftware/smack/keepalive/KeepaliveTest.java

@@ -1,4 +1,4 @@
-package org.jivesoftware.smack.ping;
+package org.jivesoftware.smack.keepalive;
 
 import static org.custommonkey.xmlunit.XMLAssert.assertXMLEqual;
 import static org.junit.Assert.assertEquals;
@@ -17,8 +17,10 @@ import org.jivesoftware.smack.TestUtils;
 import org.jivesoftware.smack.ThreadedDummyConnection;
 import org.jivesoftware.smack.filter.IQTypeFilter;
 import org.jivesoftware.smack.filter.PacketTypeFilter;
+import org.jivesoftware.smack.keepalive.KeepAliveManager;
 import org.jivesoftware.smack.packet.IQ;
 import org.jivesoftware.smack.packet.Packet;
+import org.jivesoftware.smack.ping.PingFailedListener;
 import org.jivesoftware.smack.ping.packet.Ping;
 import org.jivesoftware.smack.util.PacketParserUtils;
 import org.junit.After;
@@ -40,6 +42,7 @@ public class KeepaliveTest {
     @Before
     public void resetProperties()
     {
+        SmackConfiguration.setKeepAliveInterval(-1);
         originalTimeout = SmackConfiguration.getPacketReplyTimeout();
         SmackConfiguration.setPacketReplyTimeout(1000);
     }
@@ -56,7 +59,7 @@ public class KeepaliveTest {
     public void validatePingStanzaXML() throws Exception {
         // @formatter:off
         String control = "<iq to='juliet@capulet.lit/balcony' id='s2c1' type='get'>"
-                + "<ping xmlns='urn:xmpp:ping'/>" + "</iq>";
+                + "<ping xmlns='urn:xmpp:ping'/></iq>";
         // @formatter:on
 
         Ping ping = new Ping(TO);
@@ -65,29 +68,6 @@ public class KeepaliveTest {
         assertXMLEqual(control, ping.toXML());
     }
 
-    @Test
-    public void checkProvider() throws Exception {
-        // @formatter:off
-        String control = "<iq from='capulet.lit' to='juliet@capulet.lit/balcony' id='s2c1' type='get'>"
-                + "<ping xmlns='urn:xmpp:ping'/>" + "</iq>";
-        // @formatter:on
-        DummyConnection con = new DummyConnection();
-        IQ pingRequest = PacketParserUtils.parseIQ(TestUtils.getIQParser(control), con);
-
-        assertTrue(pingRequest instanceof Ping);
-
-        con.processPacket(pingRequest);
-
-        Packet pongPacket = con.getSentPacket();
-        assertTrue(pongPacket instanceof IQ);
-
-        IQ pong = (IQ) pongPacket;
-        assertEquals("juliet@capulet.lit/balcony", pong.getFrom());
-        assertEquals("capulet.lit", pong.getTo());
-        assertEquals("s2c1", pong.getPacketID());
-        assertEquals(IQ.Type.RESULT, pong.getType());
-    }
-
     @Test
     public void serverPingFailSingleConnection() throws Exception {
         DummyConnection connection = getConnection();
@@ -138,7 +118,7 @@ public class KeepaliveTest {
     }
 
     private void addPingFailedListener(DummyConnection con, final CountDownLatch latch) {
-        ServerPingManager manager = ServerPingManager.getInstanceFor(con);
+        KeepAliveManager manager = KeepAliveManager.getInstanceFor(con);
         manager.addPingFailedListener(new PingFailedListener() {
             @Override
             public void pingFailed() {
@@ -149,7 +129,7 @@ public class KeepaliveTest {
 
     private DummyConnection getConnection() {
         DummyConnection con = new DummyConnection();
-        ServerPingManager mgr = ServerPingManager.getInstanceFor(con);
+        KeepAliveManager mgr = KeepAliveManager.getInstanceFor(con);
         mgr.setPingInterval(PING_MINIMUM);
 
         return con;
@@ -157,7 +137,7 @@ public class KeepaliveTest {
     
     private ThreadedDummyConnection getThreadedConnection() {
         ThreadedDummyConnection con = new ThreadedDummyConnection();
-        ServerPingManager mgr = ServerPingManager.getInstanceFor(con);
+        KeepAliveManager mgr = KeepAliveManager.getInstanceFor(con);
         mgr.setPingInterval(PING_MINIMUM);
 
         return con;

test-unit/org/jivesoftware/smackx/ping/PingTest.java

@@ -38,10 +38,37 @@ public class PingTest {
         threadedCon = new ThreadedDummyConnection();
     }
     
+    @Test
+    public void checkProvider() throws Exception {
+        // @formatter:off
+        String control = "<iq from='capulet.lit' to='juliet@capulet.lit/balcony' id='s2c1' type='get'>"
+                + "<ping xmlns='urn:xmpp:ping'/>" 
+                + "</iq>";
+        // @formatter:on
+        DummyConnection con = new DummyConnection();
+        
+        // Enable ping for this connection
+        PingManager.getInstanceFor(con);
+        IQ pingRequest = PacketParserUtils.parseIQ(TestUtils.getIQParser(control), con);
+
+        assertTrue(pingRequest instanceof Ping);
+
+        con.processPacket(pingRequest);
+
+        Packet pongPacket = con.getSentPacket();
+        assertTrue(pongPacket instanceof IQ);
+
+        IQ pong = (IQ) pongPacket;
+        assertEquals("juliet@capulet.lit/balcony", pong.getFrom());
+        assertEquals("capulet.lit", pong.getTo());
+        assertEquals("s2c1", pong.getPacketID());
+        assertEquals(IQ.Type.RESULT, pong.getType());
+    }
+
     @Test
     public void checkSendingPing() throws Exception {
         dummyCon = new DummyConnection();
-        PingManager pinger = new PingManager(dummyCon);
+        PingManager pinger = PingManager.getInstanceFor(dummyCon);
         pinger.ping("test@myserver.com");
 
         Packet sentPacket = dummyCon.getSentPacket();
@@ -54,7 +81,7 @@ public class PingTest {
     public void checkSuccessfulPing() throws Exception {
         threadedCon = new ThreadedDummyConnection();
         
-        PingManager pinger = new PingManager(threadedCon);
+        PingManager pinger = PingManager.getInstanceFor(threadedCon);
 
         boolean pingSuccess = pinger.ping("test@myserver.com");
         
@@ -69,7 +96,7 @@ public class PingTest {
     @Test
     public void checkFailedPingOnTimeout() throws Exception {
         dummyCon = new DummyConnection();
-        PingManager pinger = new PingManager(dummyCon);
+        PingManager pinger = PingManager.getInstanceFor(dummyCon);
 
         boolean pingSuccess = pinger.ping("test@myserver.com");
         
@@ -96,7 +123,7 @@ public class PingTest {
         IQ serviceUnavailable = PacketParserUtils.parseIQ(TestUtils.getIQParser(reply), threadedCon);
         threadedCon.addIQReply(serviceUnavailable);
 
-        PingManager pinger = new PingManager(threadedCon);
+        PingManager pinger = PingManager.getInstanceFor(threadedCon);
 
         boolean pingSuccess = pinger.ping("test@myserver.com");
         
@@ -106,7 +133,7 @@ public class PingTest {
     @Test
     public void checkPingToServerSuccess() throws Exception {
         ThreadedDummyConnection con = new ThreadedDummyConnection();
-        PingManager pinger = new PingManager(con);
+        PingManager pinger = PingManager.getInstanceFor(con);
 
         boolean pingSuccess = pinger.pingMyServer();
         
@@ -132,7 +159,7 @@ public class PingTest {
         IQ serviceUnavailable = PacketParserUtils.parseIQ(TestUtils.getIQParser(reply), con);
         con.addIQReply(serviceUnavailable);
 
-        PingManager pinger = new PingManager(con);
+        PingManager pinger = PingManager.getInstanceFor(con);
 
         boolean pingSuccess = pinger.pingMyServer();
         
@@ -142,7 +169,7 @@ public class PingTest {
     @Test
     public void checkPingToServerTimeout() throws Exception {
         DummyConnection con = new DummyConnection();
-        PingManager pinger = new PingManager(con);
+        PingManager pinger = PingManager.getInstanceFor(con);
 
         boolean pingSuccess = pinger.pingMyServer();
         
@@ -165,7 +192,7 @@ public class PingTest {
         IQ discoReply = PacketParserUtils.parseIQ(TestUtils.getIQParser(reply), con);
         con.addIQReply(discoReply);
 
-        PingManager pinger = new PingManager(con);
+        PingManager pinger = PingManager.getInstanceFor(con);
         boolean pingSupported = pinger.isPingSupported("test@myserver.com");
         
         assertTrue(pingSupported);
@@ -187,7 +214,7 @@ public class PingTest {
         IQ discoReply = PacketParserUtils.parseIQ(TestUtils.getIQParser(reply), con);
         con.addIQReply(discoReply);
 
-        PingManager pinger = new PingManager(con);
+        PingManager pinger = PingManager.getInstanceFor(con);
         boolean pingSupported = pinger.isPingSupported("test@myserver.com");
         
         assertFalse(pingSupported);