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/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..e5fb4e422a145 --- /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 (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(); + } + } + + @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"); + } + }; + } +}