Skip to content

CAMEL-25842: camel-xmpp - the consumer logs in and enters its room again after its connection was lost #28263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davsclaus merged 3 commits into from
Oct 22, 2026
Merged

CAMEL-25842: camel-xmpp - the consumer logs in and enters its room again after its connection was lost #28263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
83d1734
CAMEL-25842: camel-xmpp - the consumer logs in and enters its room ag…
allthingssecurity Oct 20, 2026
53d929c
CAMEL-25842: camel-xmpp - restore both Smack reconnection defaults in…
allthingssecurity Oct 22, 2026
d59f575
CAMEL-25842: camel-xmpp - declare jxmpp-util-cache, which Smack needs…
allthingssecurity Oct 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions components/camel-xmpp/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,12 @@
<artifactId>jxmpp-jid</artifactId>
<version>${jxmpp-version}</version>
</dependency>
<!-- smack-core needs it, but the resolver used on CI can leave it out (smack asks for jxmpp in a version range) -->
<dependency>
<groupId>org.jxmpp</groupId>
<artifactId>jxmpp-util-cache</artifactId>
<version>${jxmpp-version}</version>
</dependency>
<dependency>
<groupId>org.igniterealtime.smack</groupId>
<artifactId>smack-core</artifactId>
Expand Down
93 changes: 83 additions & 10 deletions components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();
}
Expand Down Expand Up @@ -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() {
Expand All @@ -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());
Expand Down
61 changes: 33 additions & 28 deletions components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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));
Expand Down
Loading
Loading