Merge branch '4.3'

config/checkstyle.xml

@@ -6,6 +6,11 @@
 	<module name="SuppressionFilter">
 		<property name="file" value="config/suppressions.xml"/>
 	</module>
+	<module name="SuppressWithPlainTextCommentFilter">
+		<property name="offCommentFormat" value="CHECKSTYLE\:OFF\:(\w+)"/>
+		<property name="onCommentFormat" value="CHECKSTYLE\:ON\:(\w+)"/>
+		<property name="checkFormat" value="$1"/>
+	</module>
 	<module name="Header">
 		<property name="headerFile" value="config/${checkstyleLicenseHeader}.txt"/>
 		<property name="ignoreLines" value="3"/>
@@ -61,6 +66,14 @@
 		<property name="format" value="^\s*//[^\s]"/>
 		<property name="message" value="Comment start ('//') followed by non-space character. You would not continue after a punctuation without a space, would you?"/>
 	</module>
+	<!-- Check for synchronized keyword on Manager's static
+	     getInstanceFor() method. Note that if XMPPConnection is every
+	     replaced with something else, then we need to change it here
+	     too. -->
+	<module name="RegexpSingleline">
+		<property name="format" value="^\s*public(?!.*synchronized).*getInstanceFor\(XMPPConnection.*$"/>
+		<property name="message" value="getInstanceFor() should be synchronized"/>
+	</module>
 	<module name="JavadocPackage"/>
 	<module name="TreeWalker">
 		<module name="SuppressionCommentFilter"/>

smack-core/src/main/java/org/jivesoftware/smack/AbstractXMPPConnection.java

@@ -41,6 +41,7 @@ import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CopyOnWriteArraySet;
@@ -343,6 +344,17 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
 
     protected final AsyncButOrdered<StanzaListener> inOrderListeners = new AsyncButOrdered<>();
 
+    /**
+     * An executor which uses {@link #asyncGoLimited(Runnable)} to limit the number of asynchronously processed runnables
+     * per connection.
+     */
+    private final Executor limitedExcutor = new Executor() {
+        @Override
+        public void execute(Runnable runnable) {
+            asyncGoLimited(runnable);
+        }
+    };
+
     /**
      * The used host to establish the connection to
      */
@@ -1336,7 +1348,7 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
                         executorService = ASYNC_BUT_ORDERED.asExecutorFor(this);
                         break;
                     case async:
-                        executorService = CACHED_EXECUTOR_SERVICE;
+                        executorService = limitedExcutor;
                         break;
                     }
                     final IQRequestHandler finalIqRequestHandler = iqRequestHandler;
@@ -1379,7 +1391,7 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
         final Collection<StanzaListener> listenersToNotify = new LinkedList<>();
         extractMatchingListeners(packet, asyncRecvListeners, listenersToNotify);
         for (final StanzaListener listener : listenersToNotify) {
-            asyncGo(new Runnable() {
+            asyncGoLimited(new Runnable() {
                 @Override
                 public void run() {
                     try {
@@ -1875,6 +1887,75 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
         return getClass().getSimpleName() + '[' + localEndpointString + "] (" + getConnectionCounter() + ')';
     }
 
+    /**
+     * A queue of deferred runnables that where not executed immediately because {@link #currentAsyncRunnables} reached
+     * {@link #maxAsyncRunnables}. Note that we use a {@code LinkedList} in order to avoid space blowups in case the
+     * list ever becomes very big and shrinks again.
+     */
+    private final Queue<Runnable> deferredAsyncRunnables = new LinkedList<>();
+
+    private int deferredAsyncRunnablesCount;
+
+    private int deferredAsyncRunnablesCountPrevious;
+
+    private int maxAsyncRunnables = SmackConfiguration.getDefaultConcurrencyLevelLimit();
+
+    private int currentAsyncRunnables;
+
+    protected void asyncGoLimited(final Runnable runnable) {
+        Runnable wrappedRunnable = new Runnable() {
+            @Override
+            public void run() {
+                runnable.run();
+
+                synchronized (deferredAsyncRunnables) {
+                    Runnable defferredRunnable = deferredAsyncRunnables.poll();
+                    if (defferredRunnable == null) {
+                        currentAsyncRunnables--;
+                    } else {
+                        deferredAsyncRunnablesCount--;
+                        asyncGo(defferredRunnable);
+                    }
+                }
+            }
+        };
+
+        synchronized (deferredAsyncRunnables) {
+            if (currentAsyncRunnables < maxAsyncRunnables) {
+                currentAsyncRunnables++;
+                asyncGo(wrappedRunnable);
+            } else {
+                deferredAsyncRunnablesCount++;
+                deferredAsyncRunnables.add(wrappedRunnable);
+            }
+
+            final int HIGH_WATERMARK = 100;
+            final int INFORM_WATERMARK = 20;
+
+            final int deferredAsyncRunnablesCount = this.deferredAsyncRunnablesCount;
+
+            if (deferredAsyncRunnablesCount >= HIGH_WATERMARK
+                    && deferredAsyncRunnablesCountPrevious < HIGH_WATERMARK) {
+                LOGGER.log(Level.WARNING, "High watermark of " + HIGH_WATERMARK + " simultaneous executing runnables reached");
+            } else if (deferredAsyncRunnablesCount >= INFORM_WATERMARK
+                    && deferredAsyncRunnablesCountPrevious < INFORM_WATERMARK) {
+                LOGGER.log(Level.INFO, INFORM_WATERMARK + " simultaneous executing runnables reached");
+            }
+
+            deferredAsyncRunnablesCountPrevious = deferredAsyncRunnablesCount;
+        }
+    }
+
+    public void setMaxAsyncOperations(int maxAsyncOperations) {
+        if (maxAsyncOperations < 1) {
+            throw new IllegalArgumentException("Max async operations must be greater than 0");
+        }
+
+        synchronized (deferredAsyncRunnables) {
+            maxAsyncRunnables = maxAsyncOperations;
+        }
+    }
+
     protected static void asyncGo(Runnable runnable) {
         CACHED_EXECUTOR_SERVICE.execute(runnable);
     }

smack-core/src/main/java/org/jivesoftware/smack/AsyncButOrdered.java

@@ -55,6 +55,16 @@ public class AsyncButOrdered<K> {
 
     private final Map<K, Boolean> threadActiveMap = new WeakHashMap<>();
 
+    private final Executor executor;
+
+    public AsyncButOrdered() {
+        this(null);
+    }
+
+    public AsyncButOrdered(Executor executor) {
+        this.executor = executor;
+    }
+
     /**
      * Invoke the given {@link Runnable} asynchronous but ordered in respect to the given key.
      *
@@ -86,7 +96,11 @@ public class AsyncButOrdered<K> {
             if (newHandler) {
                 Handler handler = new Handler(keyQueue, key);
                 threadActiveMap.put(key, true);
+                if (executor == null) {
                     AbstractXMPPConnection.asyncGo(handler);
+                } else {
+                    executor.execute(handler);
+                }
             }
         }
 

smack-core/src/main/java/org/jivesoftware/smack/SmackConfiguration.java

@@ -365,4 +365,19 @@ public final class SmackConfiguration {
     public static void setUnknownIqRequestReplyMode(UnknownIqRequestReplyMode unknownIqRequestReplyMode) {
         SmackConfiguration.unknownIqRequestReplyMode = Objects.requireNonNull(unknownIqRequestReplyMode, "Must set mode");
     }
+
+    private static final int defaultConcurrencyLevelLimit;
+
+    static {
+        int availableProcessors = Runtime.getRuntime().availableProcessors();
+        if (availableProcessors < 8) {
+            defaultConcurrencyLevelLimit = 8;
+        } else {
+            defaultConcurrencyLevelLimit = (int) (availableProcessors * 1.1);
+        }
+    }
+
+    public static int getDefaultConcurrencyLevelLimit() {
+        return defaultConcurrencyLevelLimit;
+    }
 }

smack-experimental/src/main/java/org/jivesoftware/smackx/jingle_filetransfer/JingleFileTransferManager.java

@@ -32,7 +32,7 @@ public final class JingleFileTransferManager extends Manager {
         super(connection);
     }
 
-    public static JingleFileTransferManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized JingleFileTransferManager getInstanceFor(XMPPConnection connection) {
         JingleFileTransferManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new JingleFileTransferManager(connection);

smack-experimental/src/main/java/org/jivesoftware/smackx/mam/MamManager.java

@@ -178,7 +178,9 @@ public final class MamManager extends Manager {
      * @param connection the XMPP connection to get the archive for.
      * @return the instance of MamManager.
      */
+    // CHECKSTYLE:OFF:RegexpSingleline
     public static MamManager getInstanceFor(XMPPConnection connection) {
+    // CHECKSTYLE:ON:RegexpSingleline
         return getInstanceFor(connection, (Jid) null);
     }
 

smack-experimental/src/main/java/org/jivesoftware/smackx/reference/ReferenceManager.java

@@ -51,7 +51,7 @@ public final class ReferenceManager extends Manager {
      * @param connection xmpp connection
      * @return reference manager instance
      */
-    public static ReferenceManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized ReferenceManager getInstanceFor(XMPPConnection connection) {
         ReferenceManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new ReferenceManager(connection);

smack-experimental/src/main/java/org/jivesoftware/smackx/sid/StableUniqueStanzaIdManager.java

@@ -78,7 +78,7 @@ public final class StableUniqueStanzaIdManager extends Manager {
      * @param connection xmpp-connection
      * @return manager instance for the connection
      */
-    public static StableUniqueStanzaIdManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized StableUniqueStanzaIdManager getInstanceFor(XMPPConnection connection) {
         StableUniqueStanzaIdManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new StableUniqueStanzaIdManager(connection);

smack-experimental/src/main/java/org/jivesoftware/smackx/spoiler/SpoilerManager.java

@@ -61,7 +61,7 @@ public final class SpoilerManager extends Manager {
      * @param connection xmpp connection
      * @return SpoilerManager
      */
-    public static SpoilerManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized SpoilerManager getInstanceFor(XMPPConnection connection) {
         SpoilerManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new SpoilerManager(connection);

smack-extensions/src/main/java/org/jivesoftware/smackx/disco/ServiceDiscoveryManager.java

@@ -1,6 +1,6 @@
 /**
  *
- * Copyright 2003-2007 Jive Software, 2018 Florian Schmaus.
+ * Copyright 2003-2007 Jive Software, 2018-2019 Florian Schmaus.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -227,7 +227,7 @@ public final class ServiceDiscoveryManager extends Manager {
     /**
      * Returns the type of client that will be returned when asked for the client identity in a
      * disco request. The valid types are defined by the category client. Follow this link to learn
-     * the possible types: <a href="http://xmpp.org/registrar/disco-categories.html#client">Jabber::Registrar</a>.
+     * the possible types: <a href="https://xmpp.org/registrar/disco-categories.html">XMPP Registry for Service Discovery Identities</a>
      *
      * @return the type of client that will be returned when asked for the client identity in a
      *          disco request.
@@ -271,8 +271,8 @@ public final class ServiceDiscoveryManager extends Manager {
      */
     public Set<DiscoverInfo.Identity> getIdentities() {
         Set<Identity> res = new HashSet<>(identities);
-        // Add the default identity that must exist
+        // Add the main identity that must exist
-        res.add(defaultIdentity);
+        res.add(identity);
         return Collections.unmodifiableSet(res);
     }
 

smack-extensions/src/main/java/org/jivesoftware/smackx/disco/packet/DiscoverInfo.java

@@ -263,7 +263,7 @@ public class DiscoverInfo extends IQ implements TypedCloneable<DiscoverInfo> {
      * Represents the identity of a given XMPP entity. An entity may have many identities but all
      * the identities SHOULD have the same name.<p>
      *
-     * Refer to <a href="http://www.jabber.org/registrar/disco-categories.html">Jabber::Registrar</a>
+     * Refer to <a href="https://xmpp.org/registrar/disco-categories.html">XMPP Registry for Service Discovery Identities</a>
      * in order to get the official registry of values for the <i>category</i> and <i>type</i>
      * attributes.
      *
@@ -327,7 +327,7 @@ public class DiscoverInfo extends IQ implements TypedCloneable<DiscoverInfo> {
 
         /**
          * Returns the entity's category. To get the official registry of values for the
-         * 'category' attribute refer to <a href="http://www.jabber.org/registrar/disco-categories.html">Jabber::Registrar</a>
+         * 'category' attribute refer to <a href="https://xmpp.org/registrar/disco-categories.html">XMPP Registry for Service Discovery Identities</a>.
          *
          * @return the entity's category.
          */
@@ -346,7 +346,7 @@ public class DiscoverInfo extends IQ implements TypedCloneable<DiscoverInfo> {
 
         /**
          * Returns the entity's type. To get the official registry of values for the
-         * 'type' attribute refer to <a href="http://www.jabber.org/registrar/disco-categories.html">Jabber::Registrar</a>
+         * 'type' attribute refer to <a href="https://xmpp.org/registrar/disco-categories.html">XMPP Registry for Service Discovery Identities</a>.
          *
          * @return the entity's type.
          */

smack-extensions/src/main/java/org/jivesoftware/smackx/jingle/JingleTransportMethodManager.java

@@ -48,7 +48,7 @@ public final class JingleTransportMethodManager extends Manager {
         super(connection);
     }
 
-    public static JingleTransportMethodManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized JingleTransportMethodManager getInstanceFor(XMPPConnection connection) {
         JingleTransportMethodManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new JingleTransportMethodManager(connection);

smack-extensions/src/main/java/org/jivesoftware/smackx/jingle/transports/jingle_ibb/JingleIBBTransportManager.java

@@ -38,7 +38,7 @@ public final class JingleIBBTransportManager extends JingleTransportManager<Jing
         JingleContentProviderManager.addJingleContentTransportProvider(getNamespace(), new JingleIBBTransportProvider());
     }
 
-    public static JingleIBBTransportManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized JingleIBBTransportManager getInstanceFor(XMPPConnection connection) {
         JingleIBBTransportManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new JingleIBBTransportManager(connection);

smack-extensions/src/main/java/org/jivesoftware/smackx/jingle/transports/jingle_s5b/JingleS5BTransportManager.java

@@ -63,7 +63,7 @@ public final class JingleS5BTransportManager extends JingleTransportManager<Jing
         JingleContentProviderManager.addJingleContentTransportProvider(getNamespace(), new JingleS5BTransportProvider());
     }
 
-    public static JingleS5BTransportManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized JingleS5BTransportManager getInstanceFor(XMPPConnection connection) {
         JingleS5BTransportManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new JingleS5BTransportManager(connection);

smack-extensions/src/main/java/org/jivesoftware/smackx/mood/MoodManager.java

@@ -102,7 +102,7 @@ public final class MoodManager extends Manager {
         });
     }
 
-    public static MoodManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized MoodManager getInstanceFor(XMPPConnection connection) {
         MoodManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new MoodManager(connection);

smack-jingle-old/src/main/java/org/jivesoftware/smackx/jingleold/JingleSession.java

@@ -620,7 +620,7 @@ public final class JingleSession extends JingleNegotiator implements MediaReceiv
      *            A XMPP connection
      * @return a Jingle session
      */
-    public static JingleSession getInstanceFor(XMPPConnection con) {
+    public static synchronized JingleSession getInstanceFor(XMPPConnection con) {
         if (con == null) {
             throw new IllegalArgumentException("XMPPConnection cannot be null");
         }

smack-openpgp/src/main/java/org/jivesoftware/smackx/ox/OpenPgpManager.java

@@ -1,6 +1,6 @@
 /**
  *
- * Copyright 2017 Florian Schmaus, 2018 Paul Schaub.
+ * Copyright 2017-2019 Florian Schmaus, 2018 Paul Schaub.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -189,7 +189,7 @@ public final class OpenPgpManager extends Manager {
      * @param connection xmpp connection.
      * @return instance of the manager.
      */
-    public static OpenPgpManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized OpenPgpManager getInstanceFor(XMPPConnection connection) {
         OpenPgpManager manager = INSTANCES.get(connection);
         if (manager == null) {
             manager = new OpenPgpManager(connection);

smack-openpgp/src/main/java/org/jivesoftware/smackx/ox_im/OXInstantMessagingManager.java

@@ -135,7 +135,7 @@ public final class OXInstantMessagingManager extends Manager {
      * @param connection XMPP connection
      * @return manager instance
      */
-    public static OXInstantMessagingManager getInstanceFor(XMPPConnection connection) {
+    public static synchronized OXInstantMessagingManager getInstanceFor(XMPPConnection connection) {
         OXInstantMessagingManager manager = INSTANCES.get(connection);
 
         if (manager == null) {