From 2eb3dfba7e5e07c5e6fb8a07ba0c94361c5e5e5d Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Tue, 20 Oct 2026 07:21:37 +0530 Subject: [PATCH 1/5] CAMEL-25841: camel-xmpp - do not hand out the endpoint of another participant or user XmppComponent kept the endpoints it created in a map keyed by scheme, host, port and query only, and returned a cached endpoint for any URI with the same key. So xmpp://alice@host/bob@host and xmpp://alice@host/carol@host were the same endpoint: messages for carol were sent to bob. Endpoints of different users on the same server were also shared, so messages went out from the first user's account. The component now creates a new endpoint for each URI, as other components do; the CamelContext already reuses the endpoint of the same URI and stops the endpoints it holds. The endpoints of one account must still share its connection: a server keeps one session per account and resource, and closes the older session with a conflict error when another one logs in (Smack does not reconnect after that). The component now keeps one connection per account (host, port, service name, user, password, resource, login, connection configuration), used by the consumers and producers of all its endpoints, and closes it when the last of them stops. Each endpoint keeps its own participant, room and other settings. A consumer that stops removes its listeners from the shared connection. Co-Authored-By: Claude Opus 5.5 --- .../camel/component/xmpp/XmppComponent.java | 99 +++++- .../camel/component/xmpp/XmppConsumer.java | 14 +- .../component/xmpp/XmppDirectProducer.java | 12 +- .../camel/component/xmpp/XmppEndpoint.java | 53 ++- .../component/xmpp/XmppGroupChatProducer.java | 9 +- .../xmpp/XmppPrivateChatProducer.java | 9 +- .../component/xmpp/XmppPubSubProducer.java | 11 +- .../camel/component/xmpp/FakeXmppServer.java | 305 ++++++++++++++++++ .../xmpp/XmppComponentEndpointPerUriTest.java | 71 ++++ .../xmpp/XmppSharedConnectionTest.java | 139 ++++++++ 10 files changed, 676 insertions(+), 46 deletions(-) create mode 100644 components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/FakeXmppServer.java create mode 100644 components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppComponentEndpointPerUriTest.java create mode 100644 components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java index 96e6524b33aec..3a352efbc37d6 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java +++ b/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 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, SharedConnection> connections = new HashMap<>(); public XmppComponent() { } @Override protected Endpoint createEndpoint(String uri, String remaining, Map 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 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 or its connection is no longer connected. 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 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.isConnected()) { + shared.connection = endpoint.createConnection(); + } + acquired = true; + return shared.connection; + } finally { + shared.lock.unlock(); + if (!acquired) { + releaseConnection(endpoint, user); + } + } } + + /** + * 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 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(); + } + } + + private static final class SharedConnection { + private final Lock lock = new ReentrantLock(); + private final Set users = Collections.newSetFromMap(new IdentityHashMap<>()); + private volatile XMPPTCPConnection connection; + + void disconnect() { + XMPPTCPConnection current = connection; + if (current != null && current.isConnected()) { + current.disconnect(); + } + } + } + } diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java index bd4b9906a159e..47788a3285686 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java +++ b/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); @@ -187,8 +187,16 @@ protected void doStop() throws Exception { muc.leave(); muc = null; } - if (connection != null && connection.isConnected()) { - connection.disconnect(); + 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; } } diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java index 20fad1159d3db..76de6366648d3 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java @@ -42,7 +42,7 @@ 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 @@ -89,4 +89,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(); + } } diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java index 12ed1e40ffa69..3ac4c6cad8733 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java @@ -19,6 +19,7 @@ import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; +import java.util.Arrays; import java.util.List; import java.util.Map; @@ -65,7 +66,6 @@ public class XmppEndpoint extends DefaultEndpoint implements HeaderFilterStrateg private static final Logger LOG = LoggerFactory.getLogger(XmppEndpoint.class); - private volatile XMPPTCPConnection connection; private XmppBinding binding; @UriPath @@ -177,18 +177,39 @@ protected String createEndpointUri() { return "xmpp://" + host + ":" + port + "/" + getParticipant() + "?serviceName=" + serviceName; } + /** + * Gets the connection of this endpoint's account for the given consumer or producer. The endpoints of one account + * share its connection (a server keeps one session per account and resource); each endpoint keeps its own + * participant, room and other settings. Release it with {@link #releaseConnection(Object)} when the consumer or + * producer stops: the connection is closed when no consumer or producer of the account uses it any more. + */ + public XMPPTCPConnection acquireConnection(Object user) + throws InterruptedException, IOException, SmackException, XMPPException { + return getComponent().acquireConnection(this, user); + } + + /** + * Releases the connection that the given consumer or producer acquired with {@link #acquireConnection(Object)}. + */ + public void releaseConnection(Object user) { + getComponent().releaseConnection(this, user); + } + + /** + * The settings that make up the connection of an account: the endpoints with the same settings share a connection. + */ + List getConnectionKey() { + return Arrays.asList(host, port == 0 ? 5222 : port, serviceName, user, password, resource, login, connectionConfig); + } + + /** + * Creates a new connection, connected and logged in. The consumers and producers use the connection of the account + * instead, see {@link #acquireConnection(Object)}. + */ public XMPPTCPConnection createConnection() throws InterruptedException, IOException, SmackException, XMPPException { lock.lock(); try { - if (connection != null && connection.isConnected()) { - // use existing working connection - return connection; - } - - // prepare for creating new connection - connection = null; - LOG.trace("Creating new connection ..."); XMPPTCPConnection newConnection = createConnectionInternal(); @@ -229,10 +250,9 @@ public XMPPTCPConnection createConnection() // presence is not needed to be sent after login } - // okay new connection was created successfully so assign it as the connection + // okay new connection was created successfully LOG.debug("Created new connection successfully: {}", newConnection); - connection = newConnection; - return connection; + return newConnection; } finally { lock.unlock(); } @@ -540,12 +560,13 @@ public boolean isDoc() { // Implementation methods // ------------------------------------------------------------------------- + @Override + public XmppComponent getComponent() { + return (XmppComponent) super.getComponent(); + } + @Override protected void doStop() throws Exception { - if (connection != null) { - connection.disconnect(); - } - connection = null; binding = null; } diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java index c6c1dc7a16fc8..a09b367cc3ce7 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java @@ -51,7 +51,7 @@ public XmppGroupChatProducer(XmppEndpoint endpoint) { public void process(Exchange exchange) { if (connection == null) { try { - connection = endpoint.createConnection(); + connection = endpoint.acquireConnection(this); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeExchangeException("Interrupted while connecting to XMPP server.", exchange, e); @@ -116,7 +116,7 @@ private void reconnect() throws InterruptedException, IOException, SmackExceptio protected void doStart() throws Exception { if (connection == null) { try { - connection = endpoint.createConnection(); + connection = endpoint.acquireConnection(this); } catch (SmackException e) { if (endpoint.isTestConnectionOnStartup()) { throw new RuntimeCamelException( @@ -166,8 +166,9 @@ protected void doStop() throws Exception { } chat = null; - if (connection != null && connection.isConnected()) { - connection.disconnect(); + if (connection != null) { + // the connection is shared with the other consumers and producers of the account + endpoint.releaseConnection(this); } connection = null; diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java index fee149fcd521e..2e8014d4852a4 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java @@ -55,7 +55,7 @@ public void process(Exchange exchange) { // make sure we are connected try { if (connection == null) { - connection = endpoint.createConnection(); + connection = endpoint.acquireConnection(this); } if (!connection.isConnected()) { @@ -128,7 +128,7 @@ private void reconnect() throws InterruptedException, IOException, SmackExceptio protected void doStart() throws Exception { if (connection == null) { try { - connection = endpoint.createConnection(); + connection = endpoint.acquireConnection(this); } catch (SmackException e) { if (endpoint.isTestConnectionOnStartup()) { throw new RuntimeCamelException( @@ -144,8 +144,9 @@ protected void doStart() throws Exception { @Override protected void doStop() throws Exception { - if (connection != null && connection.isConnected()) { - connection.disconnect(); + if (connection != null) { + // the connection is shared with the other consumers and producers of the account + endpoint.releaseConnection(this); } connection = null; super.doStop(); diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPubSubProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPubSubProducer.java index 778d2ec6ac25e..b2b151833e00b 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPubSubProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPubSubProducer.java @@ -40,7 +40,7 @@ public XmppPubSubProducer(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 @@ -76,4 +76,13 @@ public void process(Exchange exchange) throws Exception { } } + @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(); + } } diff --git a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/FakeXmppServer.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/FakeXmppServer.java new file mode 100644 index 0000000000000..d1591b45ccac0 --- /dev/null +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/FakeXmppServer.java @@ -0,0 +1,305 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.xmpp; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * A small XMPP server for tests, on plain TCP, that does what Smack needs to connect, log in (SASL PLAIN, any password, + * or ANONYMOUS), bind a resource, send messages and enter a multi-user chat room. Like a real server it only delivers + * stanzas to sessions that logged in and bound a resource, and to room occupants that entered the room on their current + * session, and it closes the older session with a conflict stream error when a session binds the full JID (account and + * resource) of another session. + */ +final class FakeXmppServer implements AutoCloseable { + + static final String DOMAIN = "example.com"; + + private static final Pattern STREAM = Pattern.compile("]*>"); + private static final Pattern AUTH = Pattern.compile("]*?(/>|>[^<]*)"); + private static final Pattern IQ = Pattern.compile("]*?(/>|>.*?)", Pattern.DOTALL); + private static final Pattern PRESENCE = Pattern.compile("]*?(/>|>.*?)", Pattern.DOTALL); + private static final Pattern MESSAGE = Pattern.compile("]*?(/>|>.*?)", Pattern.DOTALL); + private static final Pattern ID = Pattern.compile("\\sid=['\"]([^'\"]+)['\"]"); + private static final Pattern TO = Pattern.compile("\\sto=['\"]([^'\"]+)['\"]"); + private static final Pattern MECHANISM = Pattern.compile("\\smechanism=['\"]([^'\"]+)['\"]"); + private static final Pattern AUTH_DATA = Pattern.compile(">([^<]*)"); + private static final Pattern RESOURCE = Pattern.compile("([^<]+)"); + private static final Pattern BODY = Pattern.compile("]*>(.*?)", Pattern.DOTALL); + + private final ServerSocket serverSocket; + private final List sessions = new CopyOnWriteArrayList<>(); + private final List received = new CopyOnWriteArrayList<>(); + private final AtomicInteger counter = new AtomicInteger(); + private final AtomicInteger binds = new AtomicInteger(); + private final AtomicInteger conflicts = new AtomicInteger(); + private volatile boolean running = true; + + FakeXmppServer() throws IOException { + serverSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Thread acceptor = new Thread(this::accept, "fake-xmpp-acceptor"); + acceptor.setDaemon(true); + acceptor.start(); + } + + int getPort() { + return serverSocket.getLocalPort(); + } + + /** + * The number of sessions that logged in and bound a resource. + */ + long boundSessions() { + return sessions.stream().filter(s -> s.jid != null && !s.socket.isClosed()).count(); + } + + /** + * The number of times a session bound a resource since the server started. + */ + int binds() { + return binds.get(); + } + + /** + * The number of sessions closed with a conflict stream error, as another session bound the same full JID. + */ + int conflicts() { + return conflicts.get(); + } + + /** + * The number of sessions that are occupants of the given room. + */ + long occupants(String room) { + return sessions.stream().filter(s -> room.equals(s.room) && !s.socket.isClosed()).count(); + } + + /** + * The bodies of the messages the clients sent to the given JID. + */ + List messagesTo(String jid) { + return received.stream().filter(m -> jid.equals(m[0])).map(m -> m[1]).toList(); + } + + /** + * Drops every connection, as a network failure or a server restart does. + */ + void dropConnections() throws IOException { + for (Session session : sessions) { + session.socket.close(); + } + sessions.clear(); + } + + /** + * Sends a chat message to every bound session. + */ + void sendChat(String from, String body) throws IOException { + for (Session session : sessions) { + if (session.jid != null && !session.socket.isClosed()) { + session.write("" + body + ""); + } + } + } + + /** + * Sends a groupchat message to every occupant of the room. + */ + void sendGroupChat(String room, String nick, String body) throws IOException { + for (Session session : sessions) { + if (room.equals(session.room) && !session.socket.isClosed()) { + session.write("" + body + ""); + } + } + } + + private void accept() { + while (running) { + try { + Socket socket = serverSocket.accept(); + Session session = new Session(socket); + sessions.add(session); + Thread reader = new Thread(session::read, "fake-xmpp-session"); + reader.setDaemon(true); + reader.start(); + } catch (IOException e) { + // closed + } + } + } + + @Override + public void close() throws IOException { + running = false; + serverSocket.close(); + dropConnections(); + } + + private final class Session { + private final Socket socket; + private final StringBuilder buffer = new StringBuilder(); + private volatile String user; + private volatile String jid; + private volatile String room; + + Session(Socket socket) { + this.socket = socket; + } + + synchronized void write(String xml) throws IOException { + OutputStream out = socket.getOutputStream(); + out.write(xml.getBytes(StandardCharsets.UTF_8)); + out.flush(); + } + + void read() { + byte[] bytes = new byte[8192]; + try (InputStream in = socket.getInputStream()) { + int n; + while ((n = in.read(bytes)) != -1) { + buffer.append(new String(bytes, 0, n, StandardCharsets.UTF_8)); + handle(); + } + } catch (IOException e) { + // the connection was closed + } + } + + private void handle() throws IOException { + boolean progress = true; + while (progress) { + progress = false; + for (Pattern pattern : List.of(STREAM, AUTH, IQ, PRESENCE, MESSAGE)) { + Matcher m = pattern.matcher(buffer); + if (m.find() && buffer.substring(0, m.start()).trim().replaceAll("<\\?xml[^>]*\\?>", "").isEmpty()) { + String element = m.group(); + buffer.delete(0, m.end()); + onElement(pattern, element); + progress = true; + break; + } + } + } + } + + private void onElement(Pattern pattern, String element) throws IOException { + if (pattern == STREAM) { + String features = user != null + ? "" + : "PLAIN" + + "ANONYMOUS"; + write("" + features + + ""); + } else if (pattern == AUTH) { + user = "PLAIN".equals(attribute(MECHANISM, element)) + ? new String(Base64.getDecoder().decode(content(AUTH_DATA, element)), StandardCharsets.UTF_8) + .split("\0")[1] + : "anon" + counter.incrementAndGet(); + write(""); + } else if (pattern == IQ) { + String id = attribute(ID, element); + if (user == null) { + return; + } + if (element.contains("urn:ietf:params:xml:ns:xmpp-bind")) { + String resource = content(RESOURCE, element); + bind(user + "@" + DOMAIN + "/" + (resource != null ? resource : "camel")); + write("" + jid + + ""); + } else if (element.contains("jabber:iq:roster")) { + write(""); + } else if (element.contains("type='get'") || element.contains("type=\"get\"") + || element.contains("type='set'") || element.contains("type=\"set\"")) { + String to = attribute(TO, element); + String from = to != null ? " from='" + to + "'" : ""; + if (to != null && to.startsWith("conference.") && element.contains("disco#info")) { + // the multi-user chat service + write("" + + "" + + "" + + ""); + } else { + write(""); + } + } + } else if (pattern == PRESENCE) { + String to = attribute(TO, element); + if (jid == null || to == null || !element.contains("http://jabber.org/protocol/muc")) { + return; + } + if (element.contains("type='unavailable'") || element.contains("type=\"unavailable\"")) { + room = null; + write("" + + "" + + ""); + return; + } + // entering a room: the room answers with the occupant's own presence (status 110) + room = to.substring(0, to.indexOf('/')); + write("" + + "" + + ""); + } else if (pattern == MESSAGE && jid != null) { + String to = attribute(TO, element); + String body = content(BODY, element); + if (to != null && body != null) { + received.add(new String[] { to, body }); + } + } + } + + private void bind(String fullJid) throws IOException { + // a server keeps one session per full JID: the older session is closed + for (Session other : sessions) { + if (other != this && fullJid.equals(other.jid) && !other.socket.isClosed()) { + conflicts.incrementAndGet(); + other.write("" + + ""); + other.socket.close(); + } + } + jid = fullJid; + binds.incrementAndGet(); + } + + private String attribute(Pattern pattern, String element) { + Matcher m = pattern.matcher(element.substring(0, element.indexOf('>') + 1)); + return m.find() ? m.group(1) : null; + } + + private String content(Pattern pattern, String element) { + Matcher m = pattern.matcher(element); + return m.find() ? m.group(1) : null; + } + } +} diff --git a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppComponentEndpointPerUriTest.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppComponentEndpointPerUriTest.java new file mode 100644 index 0000000000000..2618d9c1b8628 --- /dev/null +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppComponentEndpointPerUriTest.java @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.xmpp; + +import org.apache.camel.CamelContext; +import org.apache.camel.impl.DefaultCamelContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; + +/** + * Endpoints whose URIs differ only in the participant (the path) or in the user must be different endpoints: the + * component must not hand out the endpoint of another participant or user. + */ +class XmppComponentEndpointPerUriTest { + + private final CamelContext context = new DefaultCamelContext(); + + @AfterEach + void stop() { + context.stop(); + } + + @Test + void endpointPerParticipant() { + XmppEndpoint bob = context.getEndpoint("xmpp://camel-user@localhost:123/bob@localhost?password=secret", + XmppEndpoint.class); + XmppEndpoint carol = context.getEndpoint("xmpp://camel-user@localhost:123/carol@localhost?password=secret", + XmppEndpoint.class); + + assertEquals("bob@localhost", bob.getParticipant()); + assertEquals("carol@localhost", carol.getParticipant()); + } + + @Test + void endpointPerUser() { + XmppEndpoint alice = context.getEndpoint("xmpp://alice@localhost:123/bob@localhost?password=secret", + XmppEndpoint.class); + XmppEndpoint dave = context.getEndpoint("xmpp://dave@localhost:123/bob@localhost?password=secret", + XmppEndpoint.class); + + assertEquals("alice", alice.getUser()); + assertEquals("dave", dave.getUser()); + } + + @Test + void sameUriSameEndpoint() { + XmppEndpoint one = context.getEndpoint("xmpp://camel-user@localhost:123/bob@localhost?password=secret", + XmppEndpoint.class); + XmppEndpoint two = context.getEndpoint("xmpp://camel-user@localhost:123/bob@localhost?password=secret", + XmppEndpoint.class); + + assertSame(one, two); + } +} diff --git a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java new file mode 100644 index 0000000000000..b57e7ed94f4a9 --- /dev/null +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java @@ -0,0 +1,139 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.xmpp; + +import java.net.InetAddress; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.jivesoftware.smack.ConnectionConfiguration; +import org.jivesoftware.smack.tcp.XMPPTCPConnectionConfiguration; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Endpoints of one account that differ in the participant share the account's connection: a server keeps one session + * per account and resource and closes the older one when another logs in. Each endpoint still sends to its own + * participant. + */ +class XmppSharedConnectionTest extends CamelTestSupport { + + private static final String ALICE = "alice@" + FakeXmppServer.DOMAIN; + private static final String CAROL = "carol@" + FakeXmppServer.DOMAIN; + + private FakeXmppServer server; + + @Override + protected CamelContext createCamelContext() throws Exception { + server = new FakeXmppServer(); + CamelContext context = super.createCamelContext(); + // the test server has no TLS + context.getRegistry().bind("connectionConfig", XMPPTCPConnectionConfiguration.builder() + .setHostAddress(InetAddress.getLoopbackAddress()) + .setPort(server.getPort()) + .setXmppDomain(FakeXmppServer.DOMAIN) + .setSecurityMode(ConnectionConfiguration.SecurityMode.disabled) + .build()); + return context; + } + + @AfterEach + void closeServer() throws Exception { + // stop the routes before the server, so the consumers do not try to connect to a server that is closed + context.stop(); + if (server != null) { + server.close(); + } + } + + @Test + void endpointsOfOneAccountShareItsConnection() { + template.sendBody("direct:alice", "hello alice"); + template.sendBody("direct:carol", "hello carol"); + + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(List.of("hello alice"), server.messagesTo(ALICE)); + assertEquals(List.of("hello carol"), server.messagesTo(CAROL)); + }); + assertEquals(0, server.conflicts(), "A session of the account was closed by another one"); + assertEquals(1, server.binds(), "The account logged in more than once"); + assertEquals(1, server.boundSessions()); + } + + @Test + void stoppingOneEndpointKeepsTheConnectionOfTheOther() throws Exception { + context.getRouteController().stopRoute("alice"); + context.removeEndpoint(context.getEndpoint(uri(ALICE))); + + template.sendBody("direct:carol", "hello carol"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(List.of("hello carol"), server.messagesTo(CAROL))); + assertEquals(1, server.binds(), "The account logged in more than once"); + assertEquals(1, server.boundSessions()); + + // the connection is closed when its last consumer or producer stops + context.getRouteController().stopRoute("carol"); + assertEquals(1, server.boundSessions()); + context.getRouteController().stopRoute("in"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(0, server.boundSessions())); + } + + @Test + void consumerRestartedOnASharedConnectionReceivesEachMessageOnce() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:in"); + mock.expectedBodiesReceived("first", "second"); + + template.sendBody("direct:carol", "hello carol"); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.boundSessions() == 1); + server.sendChat(CAROL, "first"); + await().atMost(10, TimeUnit.SECONDS).until(() -> mock.getReceivedCounter() == 1); + + // the producer of the account keeps the connection while the consumer restarts + context.getRouteController().stopRoute("in"); + context.getRouteController().startRoute("in"); + server.sendChat(CAROL, "second"); + + // the restarted consumer must not receive a message twice + mock.setAssertPeriod(500); + mock.assertIsSatisfied(); + assertEquals(1, server.binds(), "The account logged in more than once"); + } + + private String uri(String participant) { + return "xmpp://bot@127.0.0.1:" + server.getPort() + "/" + participant + + "?password=secret&connectionConfig=#connectionConfig"; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:alice").routeId("alice").to(uri(ALICE)); + from("direct:carol").routeId("carol").to(uri(CAROL)); + from(uri(CAROL)).routeId("in").to("mock:in"); + } + }; + } +} From e9807505d6f161ee16bdfd87926f5e7e35742227 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 22 Oct 2026 06:44:00 +0530 Subject: [PATCH 2/5] CAMEL-25841: camel-xmpp - address review: reconnect the shared connection once per account From gnodet-bot's review of #28262: now that the consumers and producers of an account share one connection, producers that find it disconnected at the same time each called connect() on it under their own lock (or none), and Smack 4.4.8 throws AlreadyConnectedException to the second caller, which failed that exchange. The producers now reconnect through XmppEndpoint.reconnect(), which holds the account's lock and treats AlreadyConnectedException as "another user of the account connected it". The javadoc of the connection key and the component page say that endpoints that set connectionConfig share a connection only when they refer to the same ConnectionConfiguration instance (it has no equals()). New tests: producersOfTheAccountReconnectTheSharedConnectionConcurrently (40 sends from 8 threads right after the server dropped the session; fails 3 of 3 runs without the change with "AlreadyConnectedException: Client is already connected") and reconnectAcceptsAConnectionThatAnotherUserOfTheAccountConnected. Co-Authored-By: Claude Opus 5.5 --- .../camel/catalog/docs/xmpp-component.adoc | 7 +++ .../src/main/docs/xmpp-component.adoc | 7 +++ .../camel/component/xmpp/XmppComponent.java | 33 +++++++++++ .../component/xmpp/XmppDirectProducer.java | 10 +--- .../camel/component/xmpp/XmppEndpoint.java | 12 ++++ .../component/xmpp/XmppGroupChatProducer.java | 8 +-- .../xmpp/XmppPrivateChatProducer.java | 8 +-- .../component/xmpp/XmppPubSubProducer.java | 9 +-- .../xmpp/XmppSharedConnectionTest.java | 58 +++++++++++++++++++ 9 files changed, 125 insertions(+), 27 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/xmpp-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/xmpp-component.adoc index 98b4328558511..7354f93229552 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/xmpp-component.adoc +++ b/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 diff --git a/components/camel-xmpp/src/main/docs/xmpp-component.adoc b/components/camel-xmpp/src/main/docs/xmpp-component.adoc index 98b4328558511..7354f93229552 100644 --- a/components/camel-xmpp/src/main/docs/xmpp-component.adoc +++ b/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 diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java index 3a352efbc37d6..79b4d013c9fbc 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java @@ -135,6 +135,39 @@ XMPPTCPConnection acquireConnection(XmppEndpoint endpoint, Object user) } } + /** + * Connects the shared connection of the endpoint's account again when it is not connected. The consumers and + * producers of an account share one connection, so the reconnect is done under the account's lock; Smack's + * {@code connect()} throws {@link SmackException.AlreadyConnectedException} to a second caller, which here only + * means that another consumer or producer of the account connected it first. + */ + 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 { + if (!connection.isConnected()) { + LOG.debug("Reconnecting to: {}", XmppEndpoint.getConnectionMessage(connection)); + try { + connection.connect(); + } catch (SmackException.AlreadyConnectedException e) { + LOG.debug("Already connected by another consumer or producer of the account: {}", + XmppEndpoint.getConnectionMessage(connection)); + } + } + } finally { + if (lock != null) { + lock.unlock(); + } + } + } + /** * 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. diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java index 76de6366648d3..87cb1d8d37e06 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppDirectProducer.java @@ -45,14 +45,8 @@ public void process(Exchange exchange) throws Exception { 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: " diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java index 3ac4c6cad8733..495869b7d238a 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java @@ -188,6 +188,15 @@ public XMPPTCPConnection acquireConnection(Object user) return getComponent().acquireConnection(this, user); } + /** + * Connects the given connection, acquired with {@link #acquireConnection(Object)}, again when it is not connected; + * safe when other consumers or producers of the account do the same at the same time. + */ + public void reconnect(XMPPTCPConnection connection) + throws InterruptedException, IOException, SmackException, XMPPException { + getComponent().reconnect(this, connection); + } + /** * Releases the connection that the given consumer or producer acquired with {@link #acquireConnection(Object)}. */ @@ -197,6 +206,9 @@ public void releaseConnection(Object user) { /** * The settings that make up the connection of an account: the endpoints with the same settings share a connection. + * Smack's {@code ConnectionConfiguration} does not implement {@code equals()}, so endpoints that set the + * {@code connectionConfig} option share a connection only when they refer to the same instance (for example the + * same bean in the registry). */ List getConnectionKey() { return Arrays.asList(host, port == 0 ? 5222 : port, serviceName, user, password, resource, login, connectionConfig); diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java index a09b367cc3ce7..84731c50816a9 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java @@ -101,12 +101,8 @@ public void process(Exchange exchange) { private void reconnect() throws InterruptedException, IOException, SmackException, XMPPException { lock.lock(); try { - if (!connection.isConnected()) { - if (LOG.isDebugEnabled()) { - LOG.debug("Reconnecting to: {}", XmppEndpoint.getConnectionMessage(connection)); - } - connection.connect(); - } + // the connection is shared by the consumers and producers of the account + endpoint.reconnect(connection); } finally { lock.unlock(); } diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java index 2e8014d4852a4..950d9242da6f4 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java @@ -113,12 +113,8 @@ private Chat getOrCreateChat(ChatManager chatManager, final String participant) private void reconnect() throws InterruptedException, IOException, SmackException, XMPPException { lock.lock(); try { - if (!connection.isConnected()) { - if (LOG.isDebugEnabled()) { - LOG.debug("Reconnecting to: {}", XmppEndpoint.getConnectionMessage(connection)); - } - connection.connect(); - } + // the connection is shared by the consumers and producers of the account + endpoint.reconnect(connection); } finally { lock.unlock(); } diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPubSubProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPubSubProducer.java index b2b151833e00b..e71848572dbef 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPubSubProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPubSubProducer.java @@ -43,13 +43,8 @@ public void process(Exchange exchange) throws Exception { 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: " diff --git a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java index b57e7ed94f4a9..4e48c4333f3f2 100644 --- a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java @@ -17,19 +17,28 @@ package org.apache.camel.component.xmpp; import java.net.InetAddress; +import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import org.apache.camel.CamelContext; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.test.junit6.CamelTestSupport; +import org.jivesoftware.smack.AbstractXMPPConnection; import org.jivesoftware.smack.ConnectionConfiguration; +import org.jivesoftware.smack.SmackException; +import org.jivesoftware.smack.tcp.XMPPTCPConnection; import org.jivesoftware.smack.tcp.XMPPTCPConnectionConfiguration; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; /** @@ -120,6 +129,55 @@ void consumerRestartedOnASharedConnectionReceivesEachMessageOnce() throws Except assertEquals(1, server.binds(), "The account logged in more than once"); } + @Test + void producersOfTheAccountReconnectTheSharedConnectionConcurrently() throws Exception { + template.sendBody("direct:alice", "before alice"); + template.sendBody("direct:carol", "before carol"); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.messagesTo(CAROL).size() == 1); + + // the server closes the session: every producer of the account finds the shared connection disconnected + server.dropConnections(); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.boundSessions() == 0); + + int sends = 40; + ExecutorService pool = Executors.newFixedThreadPool(8); + try { + CountDownLatch start = new CountDownLatch(1); + List> results = new ArrayList<>(); + for (int i = 0; i < sends; i++) { + String route = i % 2 == 0 ? "direct:alice" : "direct:carol"; + String body = "after " + i; + results.add(pool.submit(() -> { + start.await(); + template.sendBody(route, body); + return null; + })); + } + start.countDown(); + for (Future result : results) { + // a producer that lost the race to connect() must not fail its exchange + result.get(20, TimeUnit.SECONDS); + } + } finally { + pool.shutdownNow(); + } + } + + @Test + void reconnectAcceptsAConnectionThatAnotherUserOfTheAccountConnected() throws Exception { + XmppEndpoint endpoint = context.getEndpoint(uri(ALICE), XmppEndpoint.class); + XMPPTCPConnectionConfiguration config + = context.getRegistry().lookupByNameAndType("connectionConfig", XMPPTCPConnectionConfiguration.class); + XMPPTCPConnection connection = new XMPPTCPConnection(config) { + @Override + public synchronized AbstractXMPPConnection connect() throws SmackException { + // what Smack 4.4.8 connect() throws to a caller when another thread connected the connection first + throw new SmackException.AlreadyConnectedException(); + } + }; + assertDoesNotThrow(() -> endpoint.reconnect(connection)); + } + private String uri(String participant) { return "xmpp://bot@127.0.0.1:" + server.getPort() + "/" + participant + "?password=secret&connectionConfig=#connectionConfig"; From 81f6fe21a53806be1c124840885f4ed8221a0c63 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:57:35 +0530 Subject: [PATCH 3/5] CAMEL-25841: camel-xmpp - reconnect the account's existing connection instead of replacing it, and release it when leaving a room fails Per review: - acquireConnection creates a connection only when the account has none; a connection that is not connected is connected and logged in again under the account's lock (connectAndLogin, also used by reconnect, which now logs in again too), so the account never has two connection objects with the same JID and resource. - Releasing the last user disconnects the connection also when it is not connected, which stops Smack's ReconnectionManager from connecting it again. - XmppConsumer and XmppGroupChatProducer remove their listeners and release the connection in a finally after leaving their room, so a failing leave() does not keep the account's connection open. - Upgrade guide 4.24: separate endpoints per participant and user, one connection per account, and XmppEndpoint.createConnection() creating a new connection on every call. - Tests: acquire during an outage gets the same connection logged in again; a consumer and a group chat producer that cannot leave their room release the connection; delivery after the concurrent reconnects; the fake connection also throws AlreadyLoggedInException. Co-Authored-By: Claude Opus 5.5 --- .../camel/component/xmpp/XmppComponent.java | 69 ++++++++++---- .../camel/component/xmpp/XmppConsumer.java | 13 ++- .../component/xmpp/XmppGroupChatProducer.java | 23 +++-- .../xmpp/XmppSharedConnectionTest.java | 92 +++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 14 +++ 5 files changed, 181 insertions(+), 30 deletions(-) diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java index 79b4d013c9fbc..6d307ec3a3f1d 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java @@ -107,9 +107,11 @@ protected void doStop() throws Exception { /** * Gets the connection of the endpoint's account for the given consumer or producer, and creates it when the account - * has no connection yet or its connection is no longer connected. The connection is shared with the other consumers - * and producers of the account, also of other endpoints; release it with - * {@link #releaseConnection(XmppEndpoint, Object)}. + * 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 { @@ -122,8 +124,10 @@ XMPPTCPConnection acquireConnection(XmppEndpoint endpoint, Object user) boolean acquired = false; shared.lock.lock(); try { - if (shared.connection == null || !shared.connection.isConnected()) { + if (shared.connection == null) { shared.connection = endpoint.createConnection(); + } else { + connectAndLogin(endpoint, shared.connection); } acquired = true; return shared.connection; @@ -136,10 +140,8 @@ XMPPTCPConnection acquireConnection(XmppEndpoint endpoint, Object user) } /** - * Connects the shared connection of the endpoint's account again when it is not connected. The consumers and - * producers of an account share one connection, so the reconnect is done under the account's lock; Smack's - * {@code connect()} throws {@link SmackException.AlreadyConnectedException} to a second caller, which here only - * means that another consumer or producer of the account connected it first. + * Connects and logs in the shared connection of the endpoint's account again when it is not connected. 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 { @@ -152,15 +154,7 @@ void reconnect(XmppEndpoint endpoint, XMPPTCPConnection connection) lock.lock(); } try { - if (!connection.isConnected()) { - LOG.debug("Reconnecting to: {}", XmppEndpoint.getConnectionMessage(connection)); - try { - connection.connect(); - } catch (SmackException.AlreadyConnectedException e) { - LOG.debug("Already connected by another consumer or producer of the account: {}", - XmppEndpoint.getConnectionMessage(connection)); - } - } + connectAndLogin(endpoint, connection); } finally { if (lock != null) { lock.unlock(); @@ -168,6 +162,34 @@ void reconnect(XmppEndpoint endpoint, XMPPTCPConnection connection) } } + /** + * 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. @@ -187,6 +209,16 @@ void releaseConnection(XmppEndpoint endpoint, Object user) { } } + /** + * 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 users = Collections.newSetFromMap(new IdentityHashMap<>()); @@ -194,7 +226,8 @@ private static final class SharedConnection { void disconnect() { XMPPTCPConnection current = connection; - if (current != null && current.isConnected()) { + if (current != null) { + // also when it is not connected: that stops Smack's ReconnectionManager from connecting it again current.disconnect(); } } diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java index 47788a3285686..7db77ce9963d4 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java @@ -184,9 +184,18 @@ 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(); + } } + 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) diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java index 84731c50816a9..2d6869a941d76 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java @@ -156,17 +156,20 @@ protected void initializeChat() @Override protected void doStop() throws Exception { - if (chat != null) { - LOG.info("Leaving room: {}", room); - chat.leave(); - } - chat = null; - - if (connection != null) { - // the connection is shared with the other consumers and producers of the account - endpoint.releaseConnection(this); + try { + if (chat != null) { + LOG.info("Leaving room: {}", room); + chat.leave(); + } + } finally { + chat = null; + // also when leaving failed: the connection is down, or Smack no longer has us in the room + if (connection != null) { + // the connection is shared with the other consumers and producers of the account + endpoint.releaseConnection(this); + } + connection = null; } - connection = null; super.doStop(); } diff --git a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java index 4e48c4333f3f2..20d157b331b7c 100644 --- a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java @@ -40,6 +40,8 @@ import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * Endpoints of one account that differ in the participant share the account's connection: a server keeps one session @@ -50,6 +52,7 @@ class XmppSharedConnectionTest extends CamelTestSupport { private static final String ALICE = "alice@" + FakeXmppServer.DOMAIN; private static final String CAROL = "carol@" + FakeXmppServer.DOMAIN; + private static final String ROOM = "dev@conference." + FakeXmppServer.DOMAIN; private FakeXmppServer server; @@ -161,6 +164,89 @@ void producersOfTheAccountReconnectTheSharedConnectionConcurrently() throws Exce } finally { pool.shutdownNow(); } + + // the reconnect also logged the connection in again, so the server delivers what the producers send now + template.sendBody("direct:alice", "later alice"); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.messagesTo(ALICE).contains("later alice")); + assertEquals(0, server.conflicts(), "A session of the account was closed by another one"); + } + + @Test + void connectionAcquiredDuringAnOutageIsTheConnectionOfTheAccount() throws Exception { + template.sendBody("direct:alice", "before"); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.messagesTo(ALICE).size() == 1); + + XmppEndpoint endpoint = context.getEndpoint(uri(ALICE), XmppEndpoint.class); + XmppComponent component = endpoint.getComponent(); + Object first = new Object(); + Object second = new Object(); + XMPPTCPConnection connection = component.acquireConnection(endpoint, first); + try { + server.dropConnections(); + await().atMost(10, TimeUnit.SECONDS).until(() -> !connection.isConnected()); + + // a consumer or producer of the account starts during the outage (a route that starts, the first message + // of a toD producer): it gets the connection the others hold, connected and logged in again + XMPPTCPConnection acquired = component.acquireConnection(endpoint, second); + assertSame(connection, acquired, "A second connection of the account was created"); + assertTrue(acquired.isAuthenticated(), "The connection was not logged in again"); + } finally { + component.releaseConnection(endpoint, second); + component.releaseConnection(endpoint, first); + } + + // the producer that got the connection before the outage sends on it again + template.sendBody("direct:alice", "after"); + await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(List.of("before", "after"), server.messagesTo(ALICE))); + assertEquals(0, server.conflicts(), "A session of the account was closed by another one"); + assertEquals(1, server.boundSessions()); + } + + @Test + void consumerThatCannotLeaveItsRoomReleasesTheConnection() throws Exception { + assertReleasedWhenLeavingFails("roomConsumer", roomUri("listener")); + } + + @Test + void groupChatProducerThatCannotLeaveItsRoomReleasesTheConnection() throws Exception { + assertReleasedWhenLeavingFails("roomProducer", roomUri("speaker")); + } + + private void assertReleasedWhenLeavingFails(String routeId, String uri) throws Exception { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + if ("roomConsumer".equals(routeId)) { + from(uri).routeId(routeId).to("mock:room"); + } else { + // the group chat producer enters the room when it starts + from("direct:room").routeId(routeId).to(uri); + } + } + }); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.occupants(ROOM) == 1); + XmppEndpoint endpoint = context.getEndpoint(uri, XmppEndpoint.class); + XmppComponent component = endpoint.getComponent(); + assertEquals(1, component.connectionUsers(endpoint)); + + // the connection is lost: leaving the room fails with NotConnectedException when the route stops + Object probe = new Object(); + XMPPTCPConnection connection = component.acquireConnection(endpoint, probe); + component.releaseConnection(endpoint, probe); + server.dropConnections(); + await().atMost(10, TimeUnit.SECONDS).until(() -> !connection.isConnected()); + + context.getRouteController().stopRoute(routeId); + + // the connection is released all the same (it is closed, as nothing else of the account uses it) + assertEquals(0, component.connectionUsers(endpoint)); + context.removeRoute(routeId); + } + + private String roomUri(String user) { + return "xmpp://" + user + "@127.0.0.1:" + server.getPort() + "?room=" + ROOM + "&nickname=" + user + + "&password=secret&connectionConfig=#connectionConfig"; } @Test @@ -174,6 +260,12 @@ public synchronized AbstractXMPPConnection connect() throws SmackException { // what Smack 4.4.8 connect() throws to a caller when another thread connected the connection first throw new SmackException.AlreadyConnectedException(); } + + @Override + public synchronized void login() throws SmackException { + // what Smack 4.4.8 login() throws to a caller when another thread logged the connection in first + throw new SmackException.AlreadyLoggedInException(); + } }; assertDoesNotThrow(() -> endpoint.reconnect(connection)); } diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 33070e296cf32..ae35e13df1580 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -1144,6 +1144,20 @@ subscription are not routed. `REPLAY_BLOCKS_OBSERVABLE`, `REPLAY_TRANSACTIONS_OB `ETH_LOG_OBSERVABLE` with `fromBlock` set to a block number are not subscribed again, as that would route the same blocks or logs again. +=== camel-xmpp - one endpoint per participant and user, and one connection per account + +XMPP URIs that differ only in the participant or the user are now separate endpoints. Before, the component could +return an endpoint that was created for another participant or user, and messages went to the wrong participant. + +All the consumers and producers of one account (the same host, port, service name, user, password, resource and login +settings) now share one XMPP connection, also across endpoints. So an account has one session on the server instead +of one per endpoint, and the server no longer closes one of them with a `conflict` error. The connection is closed when +the last route that uses the account stops. 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). + +`XmppEndpoint.createConnection()` now creates a new connection on every call, instead of returning the connection of +the endpoint, and an endpoint no longer disconnects a connection when it stops. + === camel-vertx-http - a header with several values is sent as several headers A message header whose value is a `java.util.Collection` or an object array (as the HTTP consumers map a request From 377d4629079c9ae1005e0b5a23e98763fbb40f13 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:53:16 +0530 Subject: [PATCH 4/5] CAMEL-25841: camel-xmpp - javadoc of reconnect says it also logs in again Co-Authored-By: Claude Opus 5.5 --- .../java/org/apache/camel/component/xmpp/XmppComponent.java | 5 +++-- .../java/org/apache/camel/component/xmpp/XmppEndpoint.java | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java index 6d307ec3a3f1d..861a0d9a70ee2 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java @@ -140,8 +140,9 @@ XMPPTCPConnection acquireConnection(XmppEndpoint endpoint, Object user) } /** - * Connects and logs in the shared connection of the endpoint's account again when it is not connected. The - * consumers and producers of an account share one connection, so the reconnect is done under the account's lock. + * 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 { diff --git a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java index 495869b7d238a..87c5409dea0d3 100644 --- a/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java +++ b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java @@ -189,8 +189,8 @@ public XMPPTCPConnection acquireConnection(Object user) } /** - * Connects the given connection, acquired with {@link #acquireConnection(Object)}, again when it is not connected; - * safe when other consumers or producers of the account do the same at the same time. + * Connects and logs in the given connection, acquired with {@link #acquireConnection(Object)}, again when it is not + * connected or not logged in; safe when other consumers or producers of the account do the same at the same time. */ public void reconnect(XMPPTCPConnection connection) throws InterruptedException, IOException, SmackException, XMPPException { From 252c279e3b397402fe00fd65643ee47b718a5f47 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 22 Oct 2026 13:09:41 +0530 Subject: [PATCH 5/5] CAMEL-25841: camel-xmpp - declare jxmpp-util-cache, which Smack needs on the test classpath smack-core 4.4.8 asks for jxmpp-core and jxmpp-jid in the range [1.0.0,1.0.999], while camel-xmpp pins jxmpp-jid to ${jxmpp-version} (1.1.0). The default resolver then uses jxmpp 1.0.3 everywhere, but the CI build (mvnd) left jxmpp-util-cache off the test classpath of #28263, so Smack failed to initialize (NoClassDefFoundError: org/jxmpp/util/cache/Cache). This PR adds Smack tests too; declaring it explicitly keeps it there for any resolver. Co-Authored-By: Claude Opus 5.5 --- components/camel-xmpp/pom.xml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/components/camel-xmpp/pom.xml b/components/camel-xmpp/pom.xml index 130814b73dbda..928b74a4b185a 100644 --- a/components/camel-xmpp/pom.xml +++ b/components/camel-xmpp/pom.xml @@ -49,6 +49,12 @@ jxmpp-jid ${jxmpp-version} + + + org.jxmpp + jxmpp-util-cache + ${jxmpp-version} + org.igniterealtime.smack smack-core