CAMEL-25841: camel-xmpp - do not hand out the endpoint of another participant or user

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/xmpp-component.adoc

@@ -48,6 +48,13 @@ include::partial$component-endpoint-headers.adoc[]
 
 == Usage
 
+=== Connections
+
+The consumers and producers of one account (the same host, port, service name, user, password, resource and login
+settings) share one XMPP connection, also across endpoints with different participants or rooms. With the
+`connectionConfig` option, endpoints share a connection only when they refer to the same `ConnectionConfiguration`
+instance, for example the same bean in the registry, because that class has no `equals()`.
+
 === Headers and setting Subject or Language
 
 Camel sets the message IN headers as properties on the XMPP message. You

components/camel-xmpp/pom.xml

@@ -49,6 +49,12 @@
             <artifactId>jxmpp-jid</artifactId>
             <version>${jxmpp-version}</version>
         </dependency>
+        <!-- smack-core needs it, but the resolver used on CI can leave it out (smack asks for jxmpp in a version range) -->
+        <dependency>
+            <groupId>org.jxmpp</groupId>
+            <artifactId>jxmpp-util-cache</artifactId>
+            <version>${jxmpp-version}</version>
+        </dependency>
         <dependency>
             <groupId>org.igniterealtime.smack</groupId>
             <artifactId>smack-core</artifactId>

components/camel-xmpp/src/main/docs/xmpp-component.adoc

@@ -48,6 +48,13 @@ include::partial$component-endpoint-headers.adoc[]
 
 == Usage
 
+=== Connections
+
+The consumers and producers of one account (the same host, port, service name, user, password, resource and login
+settings) share one XMPP connection, also across endpoints with different participants or rooms. With the
+`connectionConfig` option, endpoints share a connection only when they refer to the same `ConnectionConfiguration`
+instance, for example the same bean in the registry, because that class has no `equals()`.
+
 === Headers and setting Subject or Language
 
 Camel sets the message IN headers as properties on the XMPP message. You

components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java

@@ -16,17 +16,25 @@
  */
 package org.apache.camel.component.xmpp;
 
+import java.io.IOException;
 import java.net.URI;
-import java.net.URISyntaxException;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.camel.Endpoint;
 import org.apache.camel.spi.annotations.Component;
 import org.apache.camel.support.DefaultComponent;
-import org.apache.camel.support.service.ServiceHelper;
 import org.apache.camel.util.URISupport;
 import org.jivesoftware.smack.ReconnectionManager;
+import org.jivesoftware.smack.SmackException;
+import org.jivesoftware.smack.XMPPException;
+import org.jivesoftware.smack.tcp.XMPPTCPConnection;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,20 +43,15 @@ public class XmppComponent extends DefaultComponent {
 
     private static final Logger LOG = LoggerFactory.getLogger(XmppComponent.class);
 
-    // keep a cache of endpoints so they can be properly cleaned up
-    private final Map<String, XmppEndpoint> endpointCache = new HashMap<>();
+    // the connections of the endpoints, one per account: a server keeps one session per account and resource, and
+    // closes the older session (conflict) when another one logs in, so the endpoints of an account share its connection
+    private final Map<List<Object>, SharedConnection> connections = new HashMap<>();
 
     public XmppComponent() {
     }
 
     @Override
     protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
-        String cacheKey = extractCacheKeyFromUri(uri);
-        if (endpointCache.containsKey(cacheKey)) {
-            LOG.debug("Using cached endpoint for URI {}", URISupport.sanitizeUri(uri));
-            return endpointCache.get(cacheKey);
-        }
-
         LOG.debug("Creating new endpoint for URI {}", URISupport.sanitizeUri(uri));
         XmppEndpoint endpoint = new XmppEndpoint(uri, this);
 
@@ -78,8 +81,6 @@ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Obje
             }
         }
 
-        endpointCache.put(cacheKey, endpoint);
-
         return endpoint;
     }
 
@@ -92,14 +93,145 @@ protected void doStart() throws Exception {
 
     @Override
     protected void doStop() throws Exception {
-        ServiceHelper.stopService(endpointCache.values());
-        endpointCache.clear();
+        List<SharedConnection> remaining;
+        synchronized (connections) {
+            remaining = List.copyOf(connections.values());
+            connections.clear();
+        }
+        for (SharedConnection shared : remaining) {
+            shared.disconnect();
+        }
 
         super.doStop();
     }
 
-    private String extractCacheKeyFromUri(String uri) throws URISyntaxException {
-        URI u = new URI(uri);
-        return u.getScheme() + "://" + u.getHost() + u.getPort() + u.getQuery();
+    /**
+     * Gets the connection of the endpoint's account for the given consumer or producer, and creates it when the account
+     * has no connection yet. A connection that is no longer connected is connected and logged in again, not replaced:
+     * the consumers and producers that got it before keep using that object (and Smack's ReconnectionManager keeps
+     * reconnecting it), so a second connection of the account would log in the same JID and resource again and the
+     * server would close one of the two (conflict). The connection is shared with the other consumers and producers of
+     * the account, also of other endpoints; release it with {@link #releaseConnection(XmppEndpoint, Object)}.
+     */
+    XMPPTCPConnection acquireConnection(XmppEndpoint endpoint, Object user)
+            throws InterruptedException, IOException, SmackException, XMPPException {
+        List<Object> key = endpoint.getConnectionKey();
+        SharedConnection shared;
+        synchronized (connections) {
+            shared = connections.computeIfAbsent(key, k -> new SharedConnection());
+            shared.users.add(user);
+        }
+        boolean acquired = false;
+        shared.lock.lock();
+        try {
+            if (shared.connection == null) {
+                shared.connection = endpoint.createConnection();
+            } else {
+                connectAndLogin(endpoint, shared.connection);
+            }
+            acquired = true;
+            return shared.connection;
+        } finally {
+            shared.lock.unlock();
+            if (!acquired) {
+                releaseConnection(endpoint, user);
+            }
+        }
     }
+
+    /**
+     * Connects and logs in the shared connection of the endpoint's account again when it is not connected or not logged
+     * in. The consumers and producers of an account share one connection, so the reconnect is done under the account's
+     * lock.
+     */
+    void reconnect(XmppEndpoint endpoint, XMPPTCPConnection connection)
+            throws InterruptedException, IOException, SmackException, XMPPException {
+        SharedConnection shared;
+        synchronized (connections) {
+            shared = connections.get(endpoint.getConnectionKey());
+        }
+        Lock lock = shared != null ? shared.lock : null;
+        if (lock != null) {
+            lock.lock();
+        }
+        try {
+            connectAndLogin(endpoint, connection);
+        } finally {
+            if (lock != null) {
+                lock.unlock();
+            }
+        }
+    }
+
+    /**
+     * Connects the connection when it is not connected, and logs it in again when it is not logged in (a connection
+     * that is connected again is not logged in, and the server delivers nothing to it until it is). Smack's
+     * {@code connect()} and {@code login()} throw {@link SmackException.AlreadyConnectedException} and
+     * {@link SmackException.AlreadyLoggedInException} to a second caller (such as Smack's ReconnectionManager), which
+     * only means that it was done already. {@code login()} logs in with the user, password and resource the connection
+     * logged in with before.
+     */
+    private static void connectAndLogin(XmppEndpoint endpoint, XMPPTCPConnection connection)
+            throws InterruptedException, IOException, SmackException, XMPPException {
+        if (!connection.isConnected()) {
+            LOG.debug("Reconnecting to: {}", XmppEndpoint.getConnectionMessage(connection));
+            try {
+                connection.connect();
+            } catch (SmackException.AlreadyConnectedException e) {
+                LOG.debug("Already connected again: {}", XmppEndpoint.getConnectionMessage(connection));
+            }
+        }
+        if (!connection.isAuthenticated() && (endpoint.getUser() == null || endpoint.isLogin())) {
+            LOG.debug("Logging in again to: {}", XmppEndpoint.getConnectionMessage(connection));
+            try {
+                connection.login();
+            } catch (SmackException.AlreadyLoggedInException e) {
+                LOG.debug("Already logged in again: {}", XmppEndpoint.getConnectionMessage(connection));
+            }
+        }
+    }
+
+    /**
+     * Releases the connection that the given consumer or producer acquired, and disconnects it when no consumer or
+     * producer of the account uses it any more.
+     */
+    void releaseConnection(XmppEndpoint endpoint, Object user) {
+        List<Object> key = endpoint.getConnectionKey();
+        SharedConnection unused = null;
+        synchronized (connections) {
+            SharedConnection shared = connections.get(key);
+            if (shared != null && shared.users.remove(user) && shared.users.isEmpty()) {
+                connections.remove(key);
+                unused = shared;
+            }
+        }
+        if (unused != null) {
+            unused.disconnect();
+        }
+    }
+
+    /**
+     * The number of consumers and producers that use the connection of the endpoint's account. For tests.
+     */
+    int connectionUsers(XmppEndpoint endpoint) {
+        synchronized (connections) {
+            SharedConnection shared = connections.get(endpoint.getConnectionKey());
+            return shared != null ? shared.users.size() : 0;
+        }
+    }
+
+    private static final class SharedConnection {
+        private final Lock lock = new ReentrantLock();
+        private final Set<Object> users = Collections.newSetFromMap(new IdentityHashMap<>());
+        private volatile XMPPTCPConnection connection;
+
+        void disconnect() {
+            XMPPTCPConnection current = connection;
+            if (current != null) {
+                // also when it is not connected: that stops Smack's ReconnectionManager from connecting it again
+                current.disconnect();
+            }
+        }
+    }
+
 }

components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java

@@ -69,7 +69,7 @@ public XmppConsumer(XmppEndpoint endpoint, Processor processor) {
     @Override
     protected void doStart() throws Exception {
         try {
-            connection = endpoint.createConnection();
+            connection = endpoint.acquireConnection(this);
         } catch (SmackException e) {
             if (endpoint.isTestConnectionOnStartup()) {
                 throw new RuntimeCamelException("Could not connect to XMPP server.", e);
@@ -184,11 +184,28 @@ protected void doStop() throws Exception {
         if (muc != null) {
             LOG.info("Leaving room: {}", muc.getRoom());
             muc.removeMessageListener(this);
-            muc.leave();
-            muc = null;
+            try {
+                muc.leave();
+            } finally {
+                muc = null;
+                // also when leaving failed: the connection is down, or Smack no longer has us in the room
+                releaseConnection();
+            }
         }
-        if (connection != null && connection.isConnected()) {
-            connection.disconnect();
+        releaseConnection();
+    }
+
+    private void releaseConnection() {
+        if (connection != null) {
+            // the connection is shared with the other consumers and producers of the account: stop listening on it,
+            // and release it (it is closed when nobody uses it any more)
+            if (chatManager != null) {
+                chatManager.removeIncomingListener(this);
+                chatManager = null;
+            }
+            connection.removeSyncStanzaListener(this);
+            endpoint.releaseConnection(this);
+            connection = null;
         }
     }
 

components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java

@@ -42,17 +42,11 @@ public XmppDirectProducer(XmppEndpoint endpoint) {
     public void process(Exchange exchange) throws Exception {
         try {
             if (connection == null) {
-                connection = endpoint.createConnection();
+                connection = endpoint.acquireConnection(this);
             }
 
-            // make sure we are connected
-            if (!connection.isConnected()) {
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("Reconnecting to: {}", XmppEndpoint.getConnectionMessage(connection));
-                }
-
-                connection.connect();
-            }
+            // make sure we are connected (the connection is shared by the consumers and producers of the account)
+            endpoint.reconnect(connection);
         } catch (XMPPException e) {
             throw new RuntimeExchangeException(
                     "Cannot connect to XMPP Server: "
@@ -89,4 +83,14 @@ public void process(Exchange exchange) throws Exception {
                     exchange, e);
         }
     }
+
+    @Override
+    protected void doStop() throws Exception {
+        if (connection != null) {
+            // the connection is shared with the other consumers and producers of the account
+            endpoint.releaseConnection(this);
+        }
+        connection = null;
+        super.doStop();
+    }
 }