davsclaus commented on code in PR #28262:
URL: https://github.com/apache/camel/pull/28262#discussion_r4344924582
##########
components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java:
##########
@@ -187,8 +187,16 @@ protected void doStop() throws Exception {
muc.leave();
muc = null;
}
- if (connection != null && connection.isConnected()) {
- connection.disconnect();
+ 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);
Review Comment:
`muc.leave()` above throws `NotConnectedException` when the connection is
down, or `MucNotJoinedException` when Smack no longer has us in the room. When
that happens this release (and the listener removal) is skipped, so the
account's connection is never closed. The same applies to `chat.leave()` in
`XmppGroupChatProducer.doStop`. Please move the listener removal and
`releaseConnection` into a `finally`.
##########
components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java:
##########
@@ -92,14 +93,111 @@ protected void doStart() throws Exception {
@Override
protected void doStop() throws Exception {
- ServiceHelper.stopService(endpointCache.values());
- endpointCache.clear();
+ List<SharedConnection> remaining;
+ synchronized (connections) {
+ remaining = List.copyOf(connections.values());
+ connections.clear();
+ }
+ for (SharedConnection shared : remaining) {
+ shared.disconnect();
+ }
super.doStop();
}
- private String extractCacheKeyFromUri(String uri) throws
URISyntaxException {
- URI u = new URI(uri);
- return u.getScheme() + "://" + u.getHost() + u.getPort() +
u.getQuery();
+ /**
+ * Gets the connection of the endpoint's account for the given consumer or
producer, and creates it when the account
+ * has no connection yet or its connection is no longer connected. The
connection is shared with the other consumers
+ * and producers of the account, also of other endpoints; release it with
+ * {@link #releaseConnection(XmppEndpoint, Object)}.
+ */
+ XMPPTCPConnection acquireConnection(XmppEndpoint endpoint, Object user)
+ throws InterruptedException, IOException, SmackException,
XMPPException {
+ List<Object> key = endpoint.getConnectionKey();
+ SharedConnection shared;
+ synchronized (connections) {
+ shared = connections.computeIfAbsent(key, k -> new
SharedConnection());
+ shared.users.add(user);
+ }
+ boolean acquired = false;
+ shared.lock.lock();
+ try {
+ if (shared.connection == null || !shared.connection.isConnected())
{
+ shared.connection = endpoint.createConnection();
Review Comment:
When `shared.connection` exists but is not connected, this creates a second
`XMPPTCPConnection` for the same JID and resource. The consumers and producers
that hold the old one keep using it, and Smack's `ReconnectionManager` (enabled
by the component) connects it and logs it in again. Then the server closes one
of the two with `conflict`, and the other one is never reconnected. The old
object is also never disconnected, since `releaseConnection` only disconnects
`shared.connection`.
Create a connection only when there is none, and otherwise connect and log
in the existing object, as `reconnect` already does with the object it was
given (connect and log in, tolerating `AlreadyConnectedException` /
`AlreadyLoggedInException`). The `login` helper from #28263 would fit here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
