davsclaus commented on code in PR #28325:
URL: https://github.com/apache/camel/pull/28325#discussion_r4357370106
##########
components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java:
##########
@@ -227,15 +233,23 @@ private JChannel resolveChannel() throws Exception {
* Connect shared channel, called by producer and consumer.
*/
public void connect() throws Exception {
- connectCount.incrementAndGet();
LOG.trace("Connecting JGroups Channel {}", getEndpointUri());
+ if (!isOwnChannel()) {
+ ((JGroupsComponent)
getComponent()).connectSharedChannel(resolvedChannel, clusterName);
+ return;
+ }
+ connectCount.incrementAndGet();
resolvedChannel.connect(clusterName);
Review Comment:
With the new `connected` flags, a producer/consumer whose `connect` throws
no longer calls `disconnect()` on stop, so the increment here is never undone
and the endpoint's own channel is not disconnected when the route stops later.
Counting after a successful connect matches `connectSharedChannel`:
```suggestion
resolvedChannel.connect(clusterName);
connectCount.incrementAndGet();
```
##########
components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java:
##########
@@ -136,4 +150,61 @@ public void setAcceptAllObjects(boolean acceptAllObjects) {
this.acceptAllObjects = acceptAllObjects;
}
+ /**
+ * Connects a channel that several endpoints can use, for a producer or a
consumer of one of them.
+ */
+ void connectSharedChannel(JChannel sharedChannel, String clusterName)
throws Exception {
+ sharedChannelLock.lock();
+ try {
+ sharedChannel.connect(clusterName);
+ sharedChannelUsers.merge(sharedChannel, 1, Integer::sum);
+ sharedChannels.add(sharedChannel);
+ } finally {
+ sharedChannelLock.unlock();
+ }
+ }
+
+ /**
+ * Disconnects a channel that several endpoints can use, when its last
producer or consumer stops.
+ */
+ void disconnectSharedChannel(JChannel sharedChannel) {
+ sharedChannelLock.lock();
+ try {
+ Integer users = sharedChannelUsers.get(sharedChannel);
+ if (users == null) {
+ return;
+ }
+ if (users > 1) {
+ sharedChannelUsers.put(sharedChannel, users - 1);
+ } else {
+ sharedChannelUsers.remove(sharedChannel);
+ sharedChannel.disconnect();
+ }
+ } finally {
+ sharedChannelLock.unlock();
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ List<JChannel> channels;
+ sharedChannelLock.lock();
+ try {
+ channels = new ArrayList<>(sharedChannels);
+ sharedChannelUsers.clear();
+ sharedChannels.clear();
+ } finally {
+ sharedChannelLock.unlock();
+ }
+ // the channel of the component is used by all its endpoints, so it is
closed with the component, as is a
+ // channel given to an endpoint
+ if (channel != null && !channels.contains(channel)) {
+ channels.add(channel);
+ }
+ for (JChannel sharedChannel : channels) {
+ sharedChannel.close();
Review Comment:
Keeping the close at shutdown is fine for this PR. Note it still leaves a
`channel=#bean` unusable after a CamelContext restart in place, since a closed
`JChannel` cannot reconnect. Disconnecting rather than closing channels Camel
did not create could be a follow-up (it would need an upgrade-guide note).
--
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]
