From 83d17345834fb5b1e98b1524aedb6e35b97e8a13 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Tue, 20 Oct 2026 07:30:22 +0530 Subject: [PATCH 1/3] CAMEL-25842: camel-xmpp - the consumer logs in and enters its room again after its connection was lost When the connection of XmppConsumer is lost, its connection monitor and Smack's ReconnectionManager (enabled by XmppComponent) both try to connect it again. The ReconnectionManager logs in after it connected, but gives up when the monitor connected first, and the monitor only called connection.connect(), which does not log in. An XMPP server delivers nothing to a stream that did not log in and bind a resource, so the consumer then reported a working connection and consumed nothing again until the route was restarted. And whoever logged in again, the consumer did not enter its room again, although the server removed it from the room when the connection was lost. The monitor now logs the connection in again when it is not authenticated (XmppEndpoint.login, split out of createConnection; the account is only created with the first connection; nothing is done when login=false). The consumer listens for every fresh login of its connection, by the monitor or by the ReconnectionManager, logs it and enters its room again in the monitor thread; the monitor tries again if entering fails. The monitor and the ReconnectionManager can connect and log in the same connection at the same time: Smack throws AlreadyConnectedException or AlreadyLoggedInException to the second one. The monitor now takes either as done by the other one (before, it logged a WARN with a stack trace for either, and skipped the login after AlreadyConnectedException). Co-Authored-By: Claude Opus 5.5 --- .../camel/component/xmpp/XmppConsumer.java | 93 +++++- .../camel/component/xmpp/XmppEndpoint.java | 61 ++-- .../camel/component/xmpp/FakeXmppServer.java | 305 ++++++++++++++++++ .../xmpp/XmppConsumerConnectionRaceTest.java | 113 +++++++ .../xmpp/XmppConsumerReconnectTest.java | 124 +++++++ .../xmpp/XmppConsumerSmackReconnectTest.java | 111 +++++++ 6 files changed, 769 insertions(+), 38 deletions(-) create mode 100644 components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/FakeXmppServer.java create mode 100644 components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerConnectionRaceTest.java create mode 100644 components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerReconnectTest.java create mode 100644 components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerSmackReconnectTest.java 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..d020696e2898a 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 @@ -24,9 +24,11 @@ import org.apache.camel.RuntimeCamelException; import org.apache.camel.support.DefaultConsumer; import org.apache.camel.util.URISupport; +import org.jivesoftware.smack.ConnectionListener; import org.jivesoftware.smack.MessageListener; import org.jivesoftware.smack.SmackException; import org.jivesoftware.smack.StanzaListener; +import org.jivesoftware.smack.XMPPConnection; import org.jivesoftware.smack.chat2.Chat; import org.jivesoftware.smack.chat2.ChatManager; import org.jivesoftware.smack.chat2.IncomingChatMessageListener; @@ -57,9 +59,22 @@ public class XmppConsumer extends DefaultConsumer implements IncomingChatMessage private final XmppEndpoint endpoint; private MultiUserChat muc; + // set when the connection logged in again after it was lost, until the consumer entered its room again + private volatile boolean rejoinRoom; + // the connection logs in again after it was lost: by the connection monitor of this consumer, or by Smack's + // ReconnectionManager (enabled by the component), whichever connects first + private final ConnectionListener connectionListener = new ConnectionListener() { + @Override + public void authenticated(XMPPConnection connection, boolean resumed) { + if (!resumed) { + loggedInAgain(); + } + } + }; private ChatManager chatManager; - private XMPPTCPConnection connection; - private ScheduledExecutorService scheduledExecutor; + // package-private for tests + XMPPTCPConnection connection; + private volatile ScheduledExecutorService scheduledExecutor; public XmppConsumer(XmppEndpoint endpoint, Processor processor) { super(endpoint, processor); @@ -102,19 +117,14 @@ protected void doStart() throws Exception { final AndFilter packetFilter = new AndFilter(new StanzaTypeFilter(Presence.class)); connection.addSyncStanzaListener(this, packetFilter); - String roomPassword = endpoint.getRoomPassword(); MultiUserChatManager mucm = MultiUserChatManager.getInstanceFor(connection); muc = mucm.getMultiUserChat(JidCreate.entityBareFrom(endpoint.resolveRoom(connection))); muc.addMessageListener(this); - MucEnterConfiguration.Builder mucc = muc.getEnterConfigurationBuilder(Resourcepart.from(endpoint.getNickname())) - .requestNoHistory(); - if (roomPassword != null) { - mucc.withPassword(roomPassword); - } - muc.join(mucc.build()); + joinRoom(); LOG.info("Joined room: {} as: {}", muc.getRoom(), endpoint.getNickname()); } + connection.addConnectionListener(connectionListener); this.startRobustConnectionMonitor(); super.doStart(); } @@ -151,16 +161,76 @@ public void run() { endpoint.getConnectionPollDelay(), TimeUnit.SECONDS); } - private void checkConnection() throws Exception { + /** + * Connects and logs in the connection again when it was lost, and enters the room again when needed. Smack's + * ReconnectionManager can connect and log in the same connection at the same time: whichever comes second gets + * AlreadyConnectedException or AlreadyLoggedInException, which only means that the other one did it. + * Package-private for tests. + */ + void checkConnection() throws Exception { if (!connection.isConnected()) { LOG.info("Attempting to reconnect to: {}", XmppEndpoint.getConnectionMessage(connection)); try { connection.connect(); LOG.debug("Successfully connected to XMPP server through: {}", connection); + } catch (SmackException.AlreadyConnectedException e) { + LOG.debug("Already connected again to XMPP server through: {}", connection); } catch (SmackException e) { LOG.warn("Connection to XMPP server failed. Will try to reconnect later again.", e); + return; + } + } + if (!connection.isAuthenticated() && (endpoint.getUser() == null || endpoint.isLogin())) { + // connecting again does not log in: the server delivers nothing to the connection until it logs in again + // (Smack's ReconnectionManager only logs in when its attempt found the connection not connected) + try { + endpoint.login(connection); + } catch (SmackException.AlreadyLoggedInException e) { + // its login calls authenticated, which enters the room again + LOG.debug("Already logged in again to XMPP server through: {}", connection); } } + rejoinRoomIfNeeded(); + } + + /** + * Called when the connection logged in again after it was lost, by whoever connected it again. + */ + private void loggedInAgain() { + LOG.info("Logged in again to: {}", XmppEndpoint.getConnectionMessage(connection)); + if (muc != null) { + // the server removed the consumer from its room when the connection was lost. Enter it again in the thread of + // the connection monitor, not in the thread that logged in; the monitor also tries again if it fails + rejoinRoom = true; + ScheduledExecutorService executor = scheduledExecutor; + if (executor != null && !executor.isShutdown()) { + executor.execute(() -> { + try { + rejoinRoomIfNeeded(); + } catch (Exception e) { + LOG.warn("Could not enter the room again. Will try again at the next connection check.", e); + } + }); + } + } + } + + private void rejoinRoomIfNeeded() throws Exception { + if (rejoinRoom && muc != null && connection.isAuthenticated()) { + joinRoom(); + rejoinRoom = false; + LOG.info("Joined room again: {} as: {}", muc.getRoom(), endpoint.getNickname()); + } + } + + private void joinRoom() throws Exception { + MucEnterConfiguration.Builder mucc = muc.getEnterConfigurationBuilder(Resourcepart.from(endpoint.getNickname())) + .requestNoHistory(); + String roomPassword = endpoint.getRoomPassword(); + if (roomPassword != null) { + mucc.withPassword(roomPassword); + } + muc.join(mucc.build()); } private ScheduledExecutorService getExecutor() { @@ -180,6 +250,9 @@ protected void doStop() throws Exception { getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(scheduledExecutor); scheduledExecutor = null; } + if (connection != null) { + connection.removeConnectionListener(connectionListener); + } if (muc != null) { LOG.info("Leaving room: {}", muc.getRoom()); 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..9a9451e191c02 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 @@ -198,35 +198,11 @@ public XMPPTCPConnection createConnection() newConnection.addSyncStanzaListener(new XmppLogger("OUTBOUND"), stanza -> true); if (!newConnection.isAuthenticated()) { - if (user != null) { - if (LOG.isDebugEnabled()) { - LOG.debug("Logging in to XMPP as user: {} on connection: {}", user, - getConnectionMessage(newConnection)); - } - if (password == null) { - LOG.warn("No password configured for user: {} on connection: {}", user, - getConnectionMessage(newConnection)); - } - - if (createAccount) { - AccountManager accountManager = AccountManager.getInstance(newConnection); - accountManager.createAccount(Localpart.from(user), password); - } - if (login) { - if (resource != null) { - newConnection.login(user, password, Resourcepart.from(resource)); - } else { - newConnection.login(user, password); - } - } - } else { - if (LOG.isDebugEnabled()) { - LOG.debug("Logging in anonymously to XMPP on connection: {}", getConnectionMessage(newConnection)); - } - newConnection.login(); + if (user != null && createAccount) { + AccountManager accountManager = AccountManager.getInstance(newConnection); + accountManager.createAccount(Localpart.from(user), password); } - - // presence is not needed to be sent after login + login(newConnection); } // okay new connection was created successfully so assign it as the connection @@ -238,6 +214,35 @@ public XMPPTCPConnection createConnection() } } + /** + * Logs the connection in: as the configured user, or anonymously. A connection that was lost and is connected again + * must log in again before the server delivers anything to it. + */ + public void login(XMPPTCPConnection connection) + throws InterruptedException, IOException, SmackException, XMPPException { + if (user != null) { + if (LOG.isDebugEnabled()) { + LOG.debug("Logging in to XMPP as user: {} on connection: {}", user, getConnectionMessage(connection)); + } + if (password == null) { + LOG.warn("No password configured for user: {} on connection: {}", user, getConnectionMessage(connection)); + } + if (login) { + if (resource != null) { + connection.login(user, password, Resourcepart.from(resource)); + } else { + connection.login(user, password); + } + } + } else { + if (LOG.isDebugEnabled()) { + LOG.debug("Logging in anonymously to XMPP on connection: {}", getConnectionMessage(connection)); + } + connection.login(); + } + // presence is not needed to be sent after login + } + private XMPPTCPConnection createConnectionInternal() throws UnknownHostException, XmppStringprepException { if (connectionConfig != null) { return new XMPPTCPConnection(ObjectHelper.cast(XMPPTCPConnectionConfiguration.class, connectionConfig)); 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/XmppConsumerConnectionRaceTest.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerConnectionRaceTest.java new file mode 100644 index 0000000000000..2c115ab947a99 --- /dev/null +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerConnectionRaceTest.java @@ -0,0 +1,113 @@ +/* + * 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.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.camel.impl.DefaultCamelContext; +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.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * The connection monitor of the consumer and Smack's ReconnectionManager can connect and log in the same connection at + * the same time. Smack 4.4 throws AlreadyConnectedException from connect() and AlreadyLoggedInException from login() to + * whichever comes second (both methods are synchronized and check the state first): the monitor must go on in the first + * case and do nothing more in the second, instead of reporting a failed connection. + */ +class XmppConsumerConnectionRaceTest { + + private final List logins = new CopyOnWriteArrayList<>(); + private DefaultCamelContext context; + + @BeforeEach + void startContext() { + context = new DefaultCamelContext(); + context.start(); + } + + @AfterEach + void stopContext() { + context.stop(); + } + + @Test + void connectionConnectedMeanwhileByTheReconnectionManagerIsLoggedIn() throws Exception { + // the ReconnectionManager connected the connection between isConnected() and connect() of the monitor + XMPPTCPConnection connection = new XMPPTCPConnection(config()) { + @Override + public synchronized AbstractXMPPConnection connect() throws SmackException { + throw new SmackException.AlreadyConnectedException(); + } + }; + XmppConsumer consumer = new XmppConsumer(endpoint(null), exchange -> { + }); + consumer.connection = connection; + + consumer.checkConnection(); + + assertEquals(List.of(connection), logins, "The monitor did not log the connection in"); + } + + @Test + void connectionLoggedInMeanwhileByTheReconnectionManagerIsNotAFailure() throws Exception { + XMPPTCPConnection connection = new XMPPTCPConnection(config()) { + @Override + public synchronized AbstractXMPPConnection connect() { + return this; + } + }; + // the ReconnectionManager logged the connection in between isAuthenticated() and login() of the monitor + XmppConsumer consumer = new XmppConsumer(endpoint(new SmackException.AlreadyLoggedInException()), exchange -> { + }); + consumer.connection = connection; + + assertDoesNotThrow(consumer::checkConnection); + assertEquals(List.of(connection), logins); + } + + private XmppEndpoint endpoint(SmackException loginFailure) { + XmppComponent component = context.getComponent("xmpp", XmppComponent.class); + XmppEndpoint endpoint = new XmppEndpoint("xmpp://localhost:5222", component) { + @Override + public void login(XMPPTCPConnection connection) throws SmackException { + logins.add(connection); + if (loginFailure != null) { + throw loginFailure; + } + } + }; + endpoint.setHost("localhost"); + return endpoint; + } + + private static XMPPTCPConnectionConfiguration config() throws Exception { + return XMPPTCPConnectionConfiguration.builder() + .setXmppDomain("localhost") + .setSecurityMode(ConnectionConfiguration.SecurityMode.disabled) + .build(); + } +} diff --git a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerReconnectTest.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerReconnectTest.java new file mode 100644 index 0000000000000..4d2de3a0041b6 --- /dev/null +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerReconnectTest.java @@ -0,0 +1,124 @@ +/* + * 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.concurrent.TimeUnit; + +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.jivesoftware.smack.ConnectionConfiguration; +import org.jivesoftware.smack.tcp.XMPPTCPConnectionConfiguration; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * The consumer must consume again after its connection to the XMPP server was lost: it connects again, logs in again + * (the server only delivers stanzas to a session that logged in and bound a resource) and enters its room again (the + * server removed it from the room when the connection was lost). + */ +class XmppConsumerReconnectTest extends CamelTestSupport { + + 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) + .performSaslAnonymousAuthentication() + .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 privateChatIsConsumedAfterTheConnectionWasLost() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:chat"); + mock.expectedBodiesReceived("before", "after"); + + // both consumers (chat and room) are logged in + await().atMost(20, TimeUnit.SECONDS).until(() -> server.occupants(ROOM) == 1); + long sessions = server.boundSessions(); + server.sendChat("bob@" + FakeXmppServer.DOMAIN, "before"); + await().atMost(10, TimeUnit.SECONDS).until(() -> mock.getReceivedCounter() == 1); + + server.dropConnections(); + + // the consumer connects again (connectionPollDelay=1); it must also log in again to receive anything + await().atMost(20, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(sessions, server.boundSessions(), "The consumers did not log in again")); + server.sendChat("bob@" + FakeXmppServer.DOMAIN, "after"); + + mock.assertIsSatisfied(); + } + + @Test + void roomIsConsumedAfterTheConnectionWasLost() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:room"); + mock.expectedBodiesReceived("before", "after"); + + await().atMost(20, TimeUnit.SECONDS).until(() -> server.occupants(ROOM) == 1); + server.sendGroupChat(ROOM, "alice", "before"); + await().atMost(10, TimeUnit.SECONDS).until(() -> mock.getReceivedCounter() == 1); + + server.dropConnections(); + + await().atMost(20, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(1, server.occupants(ROOM), "The consumer did not enter the room again")); + server.sendGroupChat(ROOM, "alice", "after"); + + mock.assertIsSatisfied(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + String server = "xmpp://127.0.0.1:" + XmppConsumerReconnectTest.this.server.getPort(); + String options = "connectionConfig=#connectionConfig&connectionPollDelay=1"; + + from(server + "/bob@" + FakeXmppServer.DOMAIN + "?" + options).routeId("chat") + .to("mock:chat"); + + from(server + "?room=" + ROOM + "&nickname=camel&" + options).routeId("room") + .to("mock:room"); + } + }; + } +} diff --git a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerSmackReconnectTest.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerSmackReconnectTest.java new file mode 100644 index 0000000000000..d7ed32a9a9448 --- /dev/null +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerSmackReconnectTest.java @@ -0,0 +1,111 @@ +/* + * 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.concurrent.TimeUnit; + +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.jivesoftware.smack.ConnectionConfiguration; +import org.jivesoftware.smack.ReconnectionManager; +import org.jivesoftware.smack.ReconnectionManager.ReconnectionPolicy; +import org.jivesoftware.smack.tcp.XMPPTCPConnectionConfiguration; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Smack's ReconnectionManager (enabled by the component) connects and logs in again before the connection monitor of + * the consumer runs: the consumer must still enter its room again, as the server removed it from the room when the + * connection was lost. + */ +class XmppConsumerSmackReconnectTest extends CamelTestSupport { + + private static final String ROOM = "dev@conference." + FakeXmppServer.DOMAIN; + + private FakeXmppServer server; + + @Override + protected CamelContext createCamelContext() throws Exception { + // Smack's ReconnectionManager reconnects after one second, long before the connection monitor (one hour) + ReconnectionManager.setDefaultReconnectionPolicy(ReconnectionPolicy.FIXED_DELAY); + ReconnectionManager.setDefaultFixedDelay(1); + 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) + .performSaslAnonymousAuthentication() + .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(); + // the defaults of Smack + ReconnectionManager.setDefaultReconnectionPolicy(ReconnectionPolicy.RANDOM_INCREASING_DELAY); + ReconnectionManager.setDefaultFixedDelay(15); + if (server != null) { + server.close(); + } + } + + @Test + void roomIsConsumedAfterSmackReconnected() throws Exception { + MockEndpoint room = getMockEndpoint("mock:room"); + room.expectedBodiesReceived("before", "after"); + + await().atMost(20, TimeUnit.SECONDS).until(() -> server.occupants(ROOM) == 1); + server.sendGroupChat(ROOM, "alice", "before"); + await().atMost(10, TimeUnit.SECONDS).until(() -> room.getReceivedCounter() == 1); + + server.dropConnections(); + + // Smack's ReconnectionManager connects and logs in again; the consumer must enter its room again + await().atMost(20, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(1, server.boundSessions(), "Smack did not log in again")); + await().atMost(20, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(1, server.occupants(ROOM), "The consumer did not enter the room again")); + server.sendGroupChat(ROOM, "alice", "after"); + + room.assertIsSatisfied(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + String server = "xmpp://127.0.0.1:" + XmppConsumerSmackReconnectTest.this.server.getPort(); + String options = "connectionConfig=#connectionConfig&connectionPollDelay=3600"; + + from(server + "?room=" + ROOM + "&nickname=camel&" + options).routeId("room") + .to("mock:room"); + } + }; + } +} From 53d929c02f687d1002fd9cad70c5cf2b27a42025 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:18:17 +0530 Subject: [PATCH 2/3] CAMEL-25842: camel-xmpp - restore both Smack reconnection defaults in the test cleanup (setDefaultFixedDelay first, as it also sets the policy) Co-Authored-By: Claude Opus 5.5 --- .../camel/component/xmpp/XmppConsumerSmackReconnectTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerSmackReconnectTest.java b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerSmackReconnectTest.java index d7ed32a9a9448..e5fb4e422a145 100644 --- a/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerSmackReconnectTest.java +++ b/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppConsumerSmackReconnectTest.java @@ -66,9 +66,9 @@ protected CamelContext createCamelContext() throws Exception { 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(); - // the defaults of Smack - ReconnectionManager.setDefaultReconnectionPolicy(ReconnectionPolicy.RANDOM_INCREASING_DELAY); + // the defaults of Smack (setDefaultFixedDelay also sets the policy to FIXED_DELAY, so it comes first) ReconnectionManager.setDefaultFixedDelay(15); + ReconnectionManager.setDefaultReconnectionPolicy(ReconnectionPolicy.RANDOM_INCREASING_DELAY); if (server != null) { server.close(); } From d59f575d8234ae0f4cff827e15bd49a0e0fb89a9 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 22 Oct 2026 13:08:09 +0530 Subject: [PATCH 3/3] CAMEL-25842: camel-xmpp - declare jxmpp-util-cache, which Smack needs on the test classpath smack-core 4.4.8 asks for jxmpp-core and jxmpp-jid in the range [1.0.0,1.0.999], while camel-xmpp pins jxmpp-jid to ${jxmpp-version} (1.1.0). The default resolver then uses jxmpp 1.0.3 everywhere, but the CI build (mvnd) left jxmpp-util-cache off the test classpath, so Smack failed to initialize (NoClassDefFoundError: org/jxmpp/util/cache/Cache). Declaring it explicitly keeps it there for any resolver. Co-Authored-By: Claude Opus 5.5 --- components/camel-xmpp/pom.xml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/components/camel-xmpp/pom.xml b/components/camel-xmpp/pom.xml index 130814b73dbda..928b74a4b185a 100644 --- a/components/camel-xmpp/pom.xml +++ b/components/camel-xmpp/pom.xml @@ -49,6 +49,12 @@ jxmpp-jid ${jxmpp-version} + + + org.jxmpp + jxmpp-util-cache + ${jxmpp-version} + org.igniterealtime.smack smack-core