This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a4ff25a844c0 CAMEL-25841: camel-xmpp - do not hand out the endpoint of
another participant or user (#28262)
a4ff25a844c0 is described below
commit a4ff25a844c00b69fb81f96997a761b198cfba4a
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 02:19:15 2026 +0530
CAMEL-25841: camel-xmpp - do not hand out the endpoint of another
participant or user (#28262)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/xmpp-component.adoc | 7 +
components/camel-xmpp/pom.xml | 6 +
.../camel-xmpp/src/main/docs/xmpp-component.adoc | 7 +
.../apache/camel/component/xmpp/XmppComponent.java | 166 +++++++++--
.../apache/camel/component/xmpp/XmppConsumer.java | 27 +-
.../camel/component/xmpp/XmppDirectProducer.java | 22 +-
.../apache/camel/component/xmpp/XmppEndpoint.java | 65 +++--
.../component/xmpp/XmppGroupChatProducer.java | 34 +--
.../component/xmpp/XmppPrivateChatProducer.java | 17 +-
.../camel/component/xmpp/XmppPubSubProducer.java | 20 +-
.../camel/component/xmpp/FakeXmppServer.java | 305 +++++++++++++++++++++
.../xmpp/XmppComponentEndpointPerUriTest.java | 71 +++++
.../component/xmpp/XmppSharedConnectionTest.java | 289 +++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 +
14 files changed, 968 insertions(+), 82 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 98b432855851..7354f9322955 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/pom.xml b/components/camel-xmpp/pom.xml
index 130814b73dbd..928b74a4b185 100644
--- a/components/camel-xmpp/pom.xml
+++ b/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>
diff --git a/components/camel-xmpp/src/main/docs/xmpp-component.adoc
b/components/camel-xmpp/src/main/docs/xmpp-component.adoc
index 98b432855851..7354f9322955 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 96e6524b33ae..861a0d9a70ee 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<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 @@ public class XmppComponent extends DefaultComponent {
}
}
- endpointCache.put(cacheKey, endpoint);
-
return endpoint;
}
@@ -92,14 +93,145 @@ public class XmppComponent extends DefaultComponent {
@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();
+ }
+ }
+ }
+
}
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 bd4b9906a159..7db77ce9963d 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 class XmppConsumer extends DefaultConsumer implements
IncomingChatMessage
@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 @@ public class XmppConsumer extends DefaultConsumer
implements IncomingChatMessage
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;
}
}
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 20fad1159d3d..87cb1d8d37e0 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,17 +42,11 @@ public class XmppDirectProducer extends DefaultProducer {
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 class XmppDirectProducer extends DefaultProducer {
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 12ed1e40ffa6..87c5409dea0d 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 @@ package org.apache.camel.component.xmpp;
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,51 @@ public class XmppEndpoint extends DefaultEndpoint
implements HeaderFilterStrateg
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);
+ }
+
+ /**
+ * 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 {
+ getComponent().reconnect(this, connection);
+ }
+
+ /**
+ * 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.
+ * 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<Object> 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 +262,9 @@ public class XmppEndpoint extends DefaultEndpoint
implements HeaderFilterStrateg
// 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 +572,13 @@ public class XmppEndpoint extends DefaultEndpoint
implements HeaderFilterStrateg
// 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 c6c1dc7a16fc..2d6869a941d7 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 class XmppGroupChatProducer extends DefaultProducer {
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);
@@ -101,12 +101,8 @@ public class XmppGroupChatProducer extends DefaultProducer
{
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();
}
@@ -116,7 +112,7 @@ public class XmppGroupChatProducer extends DefaultProducer {
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(
@@ -160,16 +156,20 @@ public class XmppGroupChatProducer extends
DefaultProducer {
@Override
protected void doStop() throws Exception {
- if (chat != null) {
- LOG.info("Leaving room: {}", room);
- chat.leave();
- }
- chat = null;
-
- if (connection != null && connection.isConnected()) {
- connection.disconnect();
+ 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/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java
b/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java
index fee149fcd521..950d9242da6f 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 class XmppPrivateChatProducer extends DefaultProducer {
// make sure we are connected
try {
if (connection == null) {
- connection = endpoint.createConnection();
+ connection = endpoint.acquireConnection(this);
}
if (!connection.isConnected()) {
@@ -113,12 +113,8 @@ public class XmppPrivateChatProducer extends
DefaultProducer {
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();
}
@@ -128,7 +124,7 @@ public class XmppPrivateChatProducer extends
DefaultProducer {
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 +140,9 @@ public class XmppPrivateChatProducer extends
DefaultProducer {
@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 778d2ec6ac25..e71848572dbe 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,16 +40,11 @@ public class XmppPubSubProducer extends DefaultProducer {
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: "
@@ -76,4 +71,13 @@ public class XmppPubSubProducer extends DefaultProducer {
}
}
+ @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 000000000000..d1591b45ccac
--- /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("<stream:stream[^>]*>");
+ private static final Pattern AUTH =
Pattern.compile("<auth[^>]*?(/>|>[^<]*</auth>)");
+ private static final Pattern IQ =
Pattern.compile("<iq[^>]*?(/>|>.*?</iq>)", Pattern.DOTALL);
+ private static final Pattern PRESENCE =
Pattern.compile("<presence[^>]*?(/>|>.*?</presence>)", Pattern.DOTALL);
+ private static final Pattern MESSAGE =
Pattern.compile("<message[^>]*?(/>|>.*?</message>)", 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(">([^<]*)</auth>");
+ private static final Pattern RESOURCE =
Pattern.compile("<resource>([^<]+)</resource>");
+ private static final Pattern BODY =
Pattern.compile("<body[^>]*>(.*?)</body>", Pattern.DOTALL);
+
+ private final ServerSocket serverSocket;
+ private final List<Session> sessions = new CopyOnWriteArrayList<>();
+ private final List<String[]> 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<String> 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("<message from='" + from + "/phone' to='" +
session.jid + "' type='chat' id='m"
+ + counter.incrementAndGet() + "'><body>" + body
+ "</body></message>");
+ }
+ }
+ }
+
+ /**
+ * 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("<message from='" + room + "/" + nick + "' to='"
+ session.jid + "' type='groupchat' id='g"
+ + counter.incrementAndGet() + "'><body>" + body
+ "</body></message>");
+ }
+ }
+ }
+
+ 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
+ ? "<bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'/>"
+ : "<mechanisms
xmlns='urn:ietf:params:xml:ns:xmpp-sasl'><mechanism>PLAIN</mechanism>"
+ + "<mechanism>ANONYMOUS</mechanism></mechanisms>";
+ write("<?xml version='1.0'?><stream:stream
xmlns='jabber:client' "
+ + "xmlns:stream='http://etherx.jabber.org/streams'
id='s" + counter.incrementAndGet()
+ + "' from='" + DOMAIN + "' version='1.0'
xml:lang='en'><stream:features>" + features
+ + "</stream: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("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'/>");
+ } 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("<iq type='result' id='" + id + "'><bind
xmlns='urn:ietf:params:xml:ns:xmpp-bind'><jid>" + jid
+ + "</jid></bind></iq>");
+ } else if (element.contains("jabber:iq:roster")) {
+ write("<iq type='result' id='" + id + "' to='" + jid +
"'><query xmlns='jabber:iq:roster'/></iq>");
+ } 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("<iq type='result' id='" + id + "'" + from + "
to='" + jid + "'>"
+ + "<query
xmlns='http://jabber.org/protocol/disco#info'>"
+ + "<identity category='conference' type='text'
name='Rooms'/>"
+ + "<feature
var='http://jabber.org/protocol/muc'/></query></iq>");
+ } else {
+ write("<iq type='result' id='" + id + "'" + from + "
to='" + jid + "'/>");
+ }
+ }
+ } 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("<presence from='" + to + "' to='" + jid + "'
type='unavailable'>"
+ + "<x
xmlns='http://jabber.org/protocol/muc#user'><item affiliation='none'
role='none'/>"
+ + "<status code='110'/></x></presence>");
+ return;
+ }
+ // entering a room: the room answers with the occupant's own
presence (status 110)
+ room = to.substring(0, to.indexOf('/'));
+ write("<presence from='" + to + "' to='" + jid + "' id='" +
attribute(ID, element) + "'>"
+ + "<x xmlns='http://jabber.org/protocol/muc#user'><item
affiliation='none' role='participant'/>"
+ + "<status code='110'/></x></presence>");
+ } 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("<stream:error><conflict
xmlns='urn:ietf:params:xml:ns:xmpp-streams'/></stream:error>"
+ + "</stream:stream>");
+ 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 000000000000..2618d9c1b862
--- /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 000000000000..20d157b331b7
--- /dev/null
+++
b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppSharedConnectionTest.java
@@ -0,0 +1,289 @@
+/*
+ * 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.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;
+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
+ * 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 static final String ROOM = "dev@conference." +
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");
+ }
+
+ @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<Future<?>> 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();
+ }
+
+ // 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
+ 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();
+ }
+
+ @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));
+ }
+
+ private String uri(String participant) {
+ return "xmpp://[email protected]:" + 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");
+ }
+ };
+ }
+}
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 72132ccb57ed..984cd5ca20dd 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
@@ -1476,6 +1476,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
