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/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 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 96e6524b33aec..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 @@ -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. 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 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 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 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 bd4b9906a159e..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 @@ -69,7 +69,7 @@ public XmppConsumer(XmppEndpoint endpoint, Processor processor) { @Override protected void doStart() throws Exception { try { - connection = endpoint.createConnection(); + connection = endpoint.acquireConnection(this); } catch (SmackException e) { if (endpoint.isTestConnectionOnStartup()) { throw new RuntimeCamelException("Could not connect to XMPP server.", e); @@ -184,11 +184,28 @@ protected void doStop() throws Exception { if (muc != null) { LOG.info("Leaving room: {}", muc.getRoom()); muc.removeMessageListener(this); - muc.leave(); - muc = null; + try { + muc.leave(); + } finally { + muc = null; + // also when leaving failed: the connection is down, or Smack no longer has us in the room + releaseConnection(); + } } - if (connection != null && connection.isConnected()) { - connection.disconnect(); + releaseConnection(); + } + + private void releaseConnection() { + if (connection != null) { + // the connection is shared with the other consumers and producers of the account: stop listening on it, + // and release it (it is closed when nobody uses it any more) + if (chatManager != null) { + chatManager.removeIncomingListener(this); + chatManager = null; + } + connection.removeSyncStanzaListener(this); + endpoint.releaseConnection(this); + connection = null; } } 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..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 @@ -42,17 +42,11 @@ public XmppDirectProducer(XmppEndpoint endpoint) { public void process(Exchange exchange) throws Exception { try { if (connection == null) { - connection = endpoint.createConnection(); + connection = endpoint.acquireConnection(this); } - // make sure we are connected - if (!connection.isConnected()) { - if (LOG.isDebugEnabled()) { - LOG.debug("Reconnecting to: {}", XmppEndpoint.getConnectionMessage(connection)); - } - - connection.connect(); - } + // make sure we are connected (the connection is shared by the consumers and producers of the account) + endpoint.reconnect(connection); } catch (XMPPException e) { throw new RuntimeExchangeException( "Cannot connect to XMPP Server: " @@ -89,4 +83,14 @@ public void process(Exchange exchange) throws Exception { exchange, e); } } + + @Override + protected void doStop() throws Exception { + if (connection != null) { + // the connection is shared with the other consumers and producers of the account + endpoint.releaseConnection(this); + } + connection = null; + super.doStop(); + } } 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..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 @@ -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,51 @@ 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); + } + + /** + * 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 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 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 +572,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..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 @@ -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); @@ -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(); } @@ -116,7 +112,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( @@ -160,16 +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 && 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 fee149fcd521e..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 @@ -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()) { @@ -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(); } @@ -128,7 +124,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 +140,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..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 @@ -40,16 +40,11 @@ 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 - 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 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..20d157b331b7c --- /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> 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://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"); + } + }; + } +} 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