This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new d761ff850947 CAMEL-25852: camel-jgroups - an endpoint no longer
disconnects or closes the channel of the component while other endpoints use it
(#28325)
d761ff850947 is described below
commit d761ff850947da3d63c7549a6ba1fa9016ebd892
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 22:04:25 2026 +0530
CAMEL-25852: camel-jgroups - an endpoint no longer disconnects or closes
the channel of the component while other endpoints use it (#28325)
* CAMEL-25852: camel-jgroups - an endpoint no longer disconnects or closes
the channel of the component while other endpoints use it
The channel option of the JGroups component is given to every endpoint of
the
component, but each endpoint counted only its own producers and consumers:
when
those of one endpoint stopped, it disconnected the shared channel, and when
the
endpoint stopped (for example when its route was removed) it closed it. A
route
consuming from another endpoint of the same component (another URI of the
same
cluster, such as one with consumer options) then stayed started and received
nothing more; after the close, for good.
The component now counts the producers and consumers of a channel the
endpoints
did not create, and disconnects it when the last of them stops. An endpoint
only
closes a channel it created (from channelProperties or the default); the
channel
of the component, and a channel given to an endpoint (channel=#bean), are
closed
when the component stops.
A producer or consumer disconnects the channel only when it connected it:
Camel stops a service whose start failed (a consumer without
acceptAllObjects
or a filter refuses to start), and that stop took away a use it never added,
so the channel was disconnected under the other endpoints. A stopping
consumer
removes the receiver of the channel only when it is its own.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25852: camel-jgroups - count a connect of the endpoint's own
channel only when it succeeded
Review of #28325: with the connected flags of this PR, a producer or
consumer whose connect
failed no longer calls disconnect() on stop, so counting before
resolvedChannel.connect() left
the count of the endpoint's own channel one too high: the channel was not
disconnected when
the last producer or consumer stopped. JGroupsEndpoint.connect() now counts
after a successful
connect, as JGroupsComponent.connectSharedChannel does.
The own-channel path also connects and disconnects under a lock of the
endpoint (as the shared
path does under the component's lock), so that a producer that stops while
a consumer of the
same endpoint starts no longer leaves the consumer on a channel that the
producer disconnects.
resolveChannel() is package-private so that the tests can give the endpoint
a channel.
Tests: JGroupsEndpointOwnChannelTest
- aProducerWhoseConnectFailedDoesNotKeepTheChannelConnected fails when the
connect is counted
before it runs;
- aConsumerThatStartsWhileTheProducerStopsKeepsTheChannelConnected (a
channel whose disconnect
waits for the test) fails without the lock.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25852: camel-jgroups - make the count of the endpoint's own channel
a plain int
Review of #28325: connectCount of JGroupsEndpoint is only read and written
under connectLock
since the previous commit, so it no longer needs to be an AtomicInteger.
The Javadoc of
connect()/disconnect() said "shared channel"; it now says that a channel
the endpoint did not
create is counted by the component and a channel the endpoint created by
the endpoint.
No change in behaviour; the camel-jgroups unit tests (33) pass.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/component/jgroups/JGroupsComponent.java | 71 +++++++++
.../camel/component/jgroups/JGroupsConsumer.java | 17 ++-
.../camel/component/jgroups/JGroupsEndpoint.java | 58 ++++++--
.../camel/component/jgroups/JGroupsProducer.java | 9 +-
.../JGroupsComponentChannelEndpointsTest.java | 151 +++++++++++++++++++
.../jgroups/JGroupsEndpointOwnChannelTest.java | 161 +++++++++++++++++++++
6 files changed, 453 insertions(+), 14 deletions(-)
diff --git
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java
index 043dd32d7016..264a3738c861 100644
---
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java
+++
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsComponent.java
@@ -16,7 +16,14 @@
*/
package org.apache.camel.component.jgroups;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.Endpoint;
import org.apache.camel.spi.Metadata;
@@ -56,6 +63,13 @@ public class JGroupsComponent extends DefaultComponent {
+ " insecure setting.")
private boolean acceptAllObjects;
+ // the connected producers and consumers of each channel that the
endpoints did not create (such as the channel of
+ // this component), as several endpoints use such a channel
+ private final Map<JChannel, Integer> sharedChannelUsers = new
IdentityHashMap<>();
+ // the channels that the endpoints did not create and connected, closed
when this component stops
+ private final Set<JChannel> sharedChannels = Collections.newSetFromMap(new
IdentityHashMap<>());
+ private final Lock sharedChannelLock = new ReentrantLock();
+
public JGroupsComponent() {
}
@@ -136,4 +150,61 @@ public class JGroupsComponent extends DefaultComponent {
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();
+ }
+ }
+
}
diff --git
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsConsumer.java
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsConsumer.java
index c2b6237d079a..488c07e8ff61 100644
---
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsConsumer.java
+++
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsConsumer.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.jgroups;
import org.apache.camel.Processor;
import org.apache.camel.support.DefaultConsumer;
+import org.jgroups.JChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,6 +34,10 @@ public class JGroupsConsumer extends DefaultConsumer {
private final CamelJGroupsReceiver receiver;
private final JGroupsEndpoint endpoint;
+ // whether this consumer connected the channel, so that only a consumer
that connected it disconnects it (a consumer
+ // that failed to start is stopped as well, while the channel can be used
by the producers and consumers of other
+ // endpoints)
+ private volatile boolean connected;
public JGroupsConsumer(JGroupsEndpoint endpoint, Processor processor,
String clusterName) {
super(endpoint, processor);
@@ -50,13 +55,21 @@ public class JGroupsConsumer extends DefaultConsumer {
LOG.debug("Connecting receiver: {} to the cluster: {}.", receiver,
clusterName);
endpoint.getResolvedChannel().setReceiver(receiver);
endpoint.connect();
+ connected = true;
}
@Override
protected void doStop() throws Exception {
LOG.debug("Closing connection to cluster: {} from receiver: {}.",
clusterName, receiver);
- endpoint.getResolvedChannel().setReceiver(null);
- endpoint.disconnect();
+ JChannel channel = endpoint.getResolvedChannel();
+ if (channel != null && channel.getReceiver() == receiver) {
+ // only remove the receiver of this consumer, not the one of
another consumer of the channel
+ channel.setReceiver(null);
+ }
+ if (connected) {
+ connected = false;
+ endpoint.disconnect();
+ }
super.doStop();
}
}
diff --git
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
index b1f87d63ced8..9a88d432c277 100644
---
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
+++
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
@@ -17,7 +17,8 @@
package org.apache.camel.component.jgroups;
import java.io.ObjectInputFilter;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.Category;
import org.apache.camel.Component;
@@ -49,7 +50,11 @@ import org.slf4j.LoggerFactory;
public class JGroupsEndpoint extends DefaultEndpoint {
private static final Logger LOG =
LoggerFactory.getLogger(JGroupsEndpoint.class);
- private AtomicInteger connectCount = new AtomicInteger();
+ // the producers and consumers connected to the channel this endpoint
created, guarded by connectLock
+ private int connectCount;
+ // connects and disconnects the channel this endpoint created together
with its count, so that a producer that
+ // stops and a consumer that starts at the same time leave the channel
connected for the consumer
+ private final Lock connectLock = new ReentrantLock();
private JChannel channel;
private JChannel resolvedChannel;
@@ -155,13 +160,19 @@ public class JGroupsEndpoint extends DefaultEndpoint {
@Override
protected void doStop() throws Exception {
- if (resolvedChannel != null) {
+ // a channel this endpoint did not create (such as the channel of the
component) is used by other endpoints,
+ // and is closed by the component
+ if (resolvedChannel != null && isOwnChannel()) {
LOG.trace("Closing JGroups Channel {}", getEndpointUri());
resolvedChannel.close();
}
super.doStop();
}
+ private boolean isOwnChannel() {
+ return channel == null;
+ }
+
private ObjectInputFilter resolveDeserializationFilter() {
return DeserializationFilterHelper.resolveDeserializationFilter(
deserializationFilter,
DeserializationFilterHelper.DEFAULT_CLASS_DESERIALIZATION_FILTER);
@@ -213,7 +224,10 @@ public class JGroupsEndpoint extends DefaultEndpoint {
return stack != null && stack.findProtocol(AUTH.class,
SYM_ENCRYPT.class, ASYM_ENCRYPT.class) != null;
}
- private JChannel resolveChannel() throws Exception {
+ /**
+ * The channel of the endpoint: the channel option, or a channel it
creates. Package-private for tests.
+ */
+ JChannel resolveChannel() throws Exception {
if (channel != null) {
return channel;
}
@@ -224,21 +238,43 @@ public class JGroupsEndpoint extends DefaultEndpoint {
}
/**
- * Connect shared channel, called by producer and consumer.
+ * Connects the channel of this endpoint, called by producer and consumer.
A channel this endpoint did not create
+ * (such as the channel of the component) is counted by the component, as
other endpoints can use it; a channel this
+ * endpoint created is counted by the endpoint.
*/
public void connect() throws Exception {
- connectCount.incrementAndGet();
LOG.trace("Connecting JGroups Channel {}", getEndpointUri());
- resolvedChannel.connect(clusterName);
+ if (!isOwnChannel()) {
+ ((JGroupsComponent)
getComponent()).connectSharedChannel(resolvedChannel, clusterName);
+ return;
+ }
+ connectLock.lock();
+ try {
+ resolvedChannel.connect(clusterName);
+ // counted after a successful connect, as a producer or consumer
whose connect failed does not disconnect
+ connectCount++;
+ } finally {
+ connectLock.unlock();
+ }
}
/**
- * Disconnect shared channel, called by producer and consumer.
+ * Disconnects the channel of this endpoint when its last producer or
consumer disconnects, called by producer and
+ * consumer.
*/
public void disconnect() {
- if (connectCount.decrementAndGet() == 0) {
- LOG.trace("Disconnecting JGroups Channel {}", getEndpointUri());
- resolvedChannel.disconnect();
+ if (!isOwnChannel()) {
+ ((JGroupsComponent)
getComponent()).disconnectSharedChannel(resolvedChannel);
+ return;
+ }
+ connectLock.lock();
+ try {
+ if (--connectCount == 0) {
+ LOG.trace("Disconnecting JGroups Channel {}",
getEndpointUri());
+ resolvedChannel.disconnect();
+ }
+ } finally {
+ connectLock.unlock();
}
}
diff --git
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsProducer.java
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsProducer.java
index 1d041f6a10f6..e635c33d4efb 100644
---
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsProducer.java
+++
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsProducer.java
@@ -37,6 +37,9 @@ public class JGroupsProducer extends DefaultProducer {
private final String clusterName;
+ // whether this producer connected the channel, so that a producer that
failed to start does not disconnect it
+ private volatile boolean connected;
+
// Constructor
public JGroupsProducer(JGroupsEndpoint endpoint, String clusterName) {
@@ -52,11 +55,15 @@ public class JGroupsProducer extends DefaultProducer {
protected void doStart() throws Exception {
super.doStart();
endpoint.connect();
+ connected = true;
}
@Override
protected void doStop() throws Exception {
- endpoint.disconnect();
+ if (connected) {
+ connected = false;
+ endpoint.disconnect();
+ }
super.doStop();
}
diff --git
a/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsComponentChannelEndpointsTest.java
b/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsComponentChannelEndpointsTest.java
new file mode 100644
index 000000000000..74498c81132c
--- /dev/null
+++
b/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsComponentChannelEndpointsTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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.jgroups;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.jgroups.JChannel;
+import org.jgroups.ObjectMessage;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The channel of the component is used by every endpoint of the component: an
endpoint whose producer or consumer
+ * stops, or that is stopped itself, must not disconnect or close the channel
while the other endpoints use it. The
+ * channel is closed when the CamelContext stops, as is a channel given to an
endpoint.
+ */
+class JGroupsComponentChannelEndpointsTest extends CamelTestSupport {
+
+ private static final String CLUSTER = "componentChannelEndpoints";
+
+ private JChannel componentChannel;
+ private JChannel endpointChannel;
+ private JChannel sender;
+
+ @Override
+ protected void doPreSetup() throws Exception {
+ super.doPreSetup();
+ componentChannel = new JChannel();
+ sender = new JChannel();
+ sender.connect(CLUSTER);
+ }
+
+ @Override
+ protected void bindToRegistry(Registry registry) throws Exception {
+ // a channel given to one endpoint (channel=#endpointChannel) is not
created by the endpoint either
+ endpointChannel = new JChannel();
+ registry.bind("endpointChannel", endpointChannel);
+ }
+
+ @Override
+ public void doPostTearDown() {
+ sender.close();
+ componentChannel.close();
+ endpointChannel.close();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ JGroupsComponent component = new JGroupsComponent();
+ component.setChannel(componentChannel);
+ context.addComponent("jgroups", component);
+
+ // two endpoints of the same cluster: the consumer has an
option the producer has not
+ from("direct:send").routeId("producer").to("jgroups:" +
CLUSTER);
+ from("jgroups:" + CLUSTER +
"?acceptAllObjects=true").routeId("consumer").to("mock:received");
+
+ from("jgroups:" + CLUSTER +
"B?channel=#endpointChannel&acceptAllObjects=true").to("mock:endpointChannel");
+ }
+ };
+ }
+
+ @Test
+ void stoppingTheProducerKeepsTheConsumerReceiving() throws Exception {
+ context.getRouteController().stopRoute("producer");
+
+ MockEndpoint mock = getMockEndpoint("mock:received");
+ mock.expectedBodiesReceived("after the producer stopped");
+
+ sender.send(new ObjectMessage(null, "after the producer stopped"));
+
+ mock.assertIsSatisfied();
+ assertTrue(componentChannel.isConnected());
+ }
+
+ @Test
+ void removingTheProducerRouteKeepsTheConsumerReceiving() throws Exception {
+ context.getRouteController().stopRoute("producer");
+ context.removeRoute("producer");
+
+ MockEndpoint mock = getMockEndpoint("mock:received");
+ mock.expectedBodiesReceived("after the producer route was removed");
+
+ sender.send(new ObjectMessage(null, "after the producer route was
removed"));
+
+ mock.assertIsSatisfied();
+ assertTrue(componentChannel.isConnected());
+ }
+
+ @Test
+ void aConsumerThatFailsToStartKeepsTheChannelOfTheProducerConnected()
throws Exception {
+ context.getRouteController().stopRoute("consumer");
+ assertTrue(componentChannel.isConnected(), "connected for the
producer");
+
+ // no acceptAllObjects and no filter: the consumer refuses to start,
and is stopped
+ addRouteWithAConsumerThatFailsToStart();
+
+ assertTrue(componentChannel.isConnected(), "the channel of the
producer is still connected");
+ }
+
+ @Test
+ void aConsumerThatFailsToStartKeepsTheOtherConsumerReceiving() throws
Exception {
+ addRouteWithAConsumerThatFailsToStart();
+
+ MockEndpoint mock = getMockEndpoint("mock:received");
+ mock.expectedBodiesReceived("after a consumer failed to start");
+
+ sender.send(new ObjectMessage(null, "after a consumer failed to
start"));
+
+ mock.assertIsSatisfied();
+ }
+
+ private void addRouteWithAConsumerThatFailsToStart() {
+ assertThrows(Exception.class, () -> context.addRoutes(new
RouteBuilder() {
+ @Override
+ public void configure() {
+ from("jgroups:" + CLUSTER +
"?enableViewMessages=true").routeId("failing").to("mock:failing");
+ }
+ }));
+ }
+
+ @Test
+ void stoppingTheCamelContextClosesTheChannels() {
+ assertTrue(endpointChannel.isConnected());
+
+ context.stop();
+
+ assertTrue(componentChannel.isClosed());
+ assertTrue(endpointChannel.isClosed());
+ }
+}
diff --git
a/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsEndpointOwnChannelTest.java
b/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsEndpointOwnChannelTest.java
new file mode 100644
index 000000000000..e8b2f748a966
--- /dev/null
+++
b/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsEndpointOwnChannelTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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.jgroups;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.jgroups.JChannel;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The producers and consumers of an endpoint share the channel the endpoint
created: it is disconnected when the last
+ * of them that connected it stops.
+ */
+class JGroupsEndpointOwnChannelTest {
+
+ private static final String CLUSTER = "endpointOwnChannel";
+
+ @Test
+ void aProducerWhoseConnectFailedDoesNotKeepTheChannelConnected() throws
Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ JChannel channel = new FailingFirstConnectChannel();
+ JGroupsEndpoint endpoint = endpointWithChannel(context, channel);
+
+ Producer producer = endpoint.createProducer();
+ assertThrows(Exception.class, producer::start);
+ // Camel stops a service whose start failed
+ producer.stop();
+
+ Consumer consumer = endpoint.createConsumer(exchange -> {
+ });
+ consumer.start();
+ assertTrue(channel.isConnected());
+
+ // the only producer or consumer that connected the channel stops
+ consumer.stop();
+ assertFalse(channel.isConnected(), "the failed connect of the
producer is not counted");
+
+ endpoint.stop();
+ }
+ }
+
+ @Test
+ void aConsumerThatStartsWhileTheProducerStopsKeepsTheChannelConnected()
throws Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ BlockingDisconnectChannel channel = new
BlockingDisconnectChannel();
+ JGroupsEndpoint endpoint = endpointWithChannel(context, channel);
+
+ Producer producer = endpoint.createProducer();
+ producer.start();
+ Consumer consumer = endpoint.createConsumer(exchange -> {
+ });
+
+ // the producer stops: it is the last user of the channel, so it
disconnects it
+ Thread stopping = new Thread(producer::stop, "stopping-producer");
+ stopping.start();
+ assertTrue(channel.disconnecting.await(20, TimeUnit.SECONDS));
+
+ // the consumer starts while the channel is being disconnected
+ Thread starting = new Thread(consumer::start, "starting-consumer");
+ starting.start();
+ // it waits until the disconnect is done (or, without the lock,
finds the channel still connected and returns)
+ await().atMost(20, TimeUnit.SECONDS).until(
+ () -> starting.getState() == Thread.State.WAITING ||
starting.getState() == Thread.State.TERMINATED);
+ channel.release.countDown();
+ stopping.join(20000);
+ starting.join(20000);
+ assertFalse(stopping.isAlive());
+ assertFalse(starting.isAlive());
+
+ assertTrue(channel.isConnected(), "connected for the consumer");
+
+ consumer.stop();
+ assertFalse(channel.isConnected());
+ endpoint.stop();
+ }
+ }
+
+ private static JGroupsEndpoint endpointWithChannel(CamelContext context,
JChannel channel) throws Exception {
+ JGroupsComponent component = context.getComponent("jgroups",
JGroupsComponent.class);
+ JGroupsEndpoint endpoint = new JGroupsEndpoint("jgroups:" + CLUSTER,
component, null, CLUSTER, null, false) {
+ @Override
+ JChannel resolveChannel() {
+ // the channel that the endpoint creates (no channel option)
+ return channel;
+ }
+ };
+ endpoint.setAcceptAllObjects(true);
+ endpoint.setCamelContext(context);
+ endpoint.start();
+ return endpoint;
+ }
+
+ /**
+ * A channel whose first connect fails, as when its transport cannot bind.
+ */
+ private static final class FailingFirstConnectChannel extends JChannel {
+
+ private final AtomicInteger connects = new AtomicInteger();
+
+ FailingFirstConnectChannel() throws Exception {
+ }
+
+ @Override
+ public JChannel connect(String clusterName) throws Exception {
+ if (connects.incrementAndGet() == 1) {
+ throw new IllegalStateException("first connect fails");
+ }
+ return super.connect(clusterName);
+ }
+ }
+
+ /**
+ * A channel whose disconnect waits until the test releases it.
+ */
+ private static final class BlockingDisconnectChannel extends JChannel {
+
+ private final CountDownLatch disconnecting = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ BlockingDisconnectChannel() throws Exception {
+ }
+
+ @Override
+ public JChannel disconnect() {
+ disconnecting.countDown();
+ try {
+ release.await(20, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return super.disconnect();
+ }
+ }
+}
