You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cs...@apache.org on 2026/06/15 17:18:59 UTC
(activemq) branch activemq-6.2.x updated: Refactor network connections to process after ConnectionInfo (#2112) (#2116)
This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch activemq-6.2.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-6.2.x by this push:
new 711675892c Refactor network connections to process after ConnectionInfo (#2112) (#2116)
711675892c is described below
commit 711675892c29db0313ef35f5820d4f531e8b891b
Author: Christopher L. Shannon <cs...@apache.org>
AuthorDate: Mon Jun 15 13:18:54 2026 -0400
Refactor network connections to process after ConnectionInfo (#2112) (#2116)
This refactors the setup for network connections to be processed
after receiving ConnectionInfo. Network bridges are established by
each broker sending a BrokerInfo command to the other broker to provide
remote broker information, but this is done before ConnectionInfo.
This commit reworks the connection to capture the BrokerInfo information,
but delay processing until after the connection information has been
received and processed by broker.addConnection(). The future that was
previously added for the connection id is no longer needed and removed.
This commit also simplifies durable sync for bridges and prevents a
race condition on startup by making sure the initial bridge and the
duplex side only send back the BrokerSubscriptionInfo command after
fully initialized.
(cherry picked from commit 230db1fec0b4bfdb8dcf06cdceeb00585ed8f82c)
---
.../activemq/broker/TransportConnection.java | 137 ++++++++++-----------
.../activemq/broker/region/RegionBroker.java | 37 ++++++
.../network/DemandForwardingBridgeSupport.java | 14 ++-
.../network/DurableSyncNetworkBridgeAuthTest.java | 115 +++++++++++++++--
4 files changed, 217 insertions(+), 86 deletions(-)
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
index 9e77101744..b6fe548857 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
@@ -165,7 +165,6 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
private final ReentrantReadWriteLock serviceLock = new ReentrantReadWriteLock();
private String duplexNetworkConnectorId;
private final long connectedTimestamp;
- private final CompletableFuture<ConnectionId> initialConnectionId = new CompletableFuture<>();
/**
* @param taskRunnerFactory - can be null if you want direct dispatch to the transport
@@ -854,16 +853,14 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
try {
broker.addConnection(context, info);
- // Complete the future with the connectionId if we completed
- // the broker.addConnection() chain successfully
- initialConnectionId.complete(info.getConnectionId());
+ // If we completed broker.addConnection() successfully we can now
+ // continue the required extra setup for any network connections
+ addNetworkConnection();
} catch (Exception e) {
synchronized (brokerConnectionStates) {
brokerConnectionStates.remove(info.getConnectionId());
}
unregisterConnectionState(info.getConnectionId());
- // complete with the exception
- initialConnectionId.completeExceptionally(e);
LOG.warn("Failed to add Connection id={}, clientId={}, clientIP={} due to {}",
info.getConnectionId(), clientId, info.getClientIp(), e.getLocalizedMessage());
//AMQ-6561 - stop for all exceptions on addConnection
@@ -1401,44 +1398,75 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
throw new IOException("Unexpected extra broker info command received from: " + info.getBrokerId());
}
if (info.isSlaveBroker()) {
- LOG.error(" Slave Brokers are no longer supported - slave trying to attach is: {}", info.getBrokerName());
- } else if (info.isNetworkConnection() && !info.isDuplexConnection()) {
+ LOG.error("Slave Brokers are no longer supported - slave trying to attach is: {}", info.getBrokerName());
+ throw new IOException("Slave Brokers are no longer supported - slave trying to attach is: " + info.getBrokerName());
+ }
+
+ // The only thing this method now does is capture the BrokerInfo object and mark as a network connection.
+ // Actual processing for starting up duplex bridges and for durable sync has been moved until
+ // after ConnectionInfo has been received.
+
+ // If this is duplex we need to get the ID configured so we can use it
+ // to close existing connections later that match the same ID
+ // This will be done inside the RegionBroker
+ if (info.isNetworkConnection() && info.isDuplexConnection()) {
+ NetworkBridgeConfiguration config = getNetworkConfiguration(info);
+ config.setBrokerName(broker.getBrokerName());
+ String duplexNetworkConnectorId = config.getName() + "@" + info.getBrokerId();
+ setDuplexNetworkConnectorId(duplexNetworkConnectorId);
+ }
+
+ this.brokerInfo = info;
+ networkConnection = true;
+ List<TransportConnectionState> connectionStates = listConnectionStates();
+ for (TransportConnectionState cs : connectionStates) {
+ cs.getContext().setNetworkConnection(true);
+ }
+ return null;
+ }
+
+ // Process the network connection set up
+ private void addNetworkConnection() throws Exception {
+ final BrokerInfo info = this.brokerInfo;
+ if (info == null || !info.isNetworkConnection()){
+ return;
+ }
+
+ // For a one way bridge we need to respond on bridge creation by sending back the durable
+ // subs if durable sync is enabled via BrokerSubscriptionInfo command. The bridge is only
+ // initialized on one broker, so if this is the passive side we know it's initialized and
+ // we can respond.
+ //
+ // For a duplex bridge, we do NOT send back the durable subs. To simplify and ensure
+ // the bridge is fully initialized, the bridge startup will now handle sending
+ // BrokerSubscriptionInfo to the remote broker once fully started.
+ if (!info.isDuplexConnection()) {
try {
- // register durable sync to be sent after ConnectionInfo has been handled
- registerDurableSync(getNetworkConfiguration(info), info);
+ NetworkBridgeConfiguration config = getNetworkConfiguration(info);
+ if (config.isSyncDurableSubs() && protocolVersion.get() >= CommandTypes.PROTOCOL_VERSION_DURABLE_SYNC) {
+ LOG.debug("SyncDurableSubs is enabled, Sending BrokerSubscriptionInfo");
+ // Send back the durable subs as this is a one way bridge
+ dispatchSync(NetworkBridgeUtils.getBrokerSubscriptionInfo(this.broker.getBrokerService(), config));
+ }
} catch (Exception e) {
- LOG.error("Failed to register durable sync for network bridge creation from broker {}", info.getBrokerId(), e);
- return null;
+ LOG.error("Failed to respond to network bridge creation from broker {}", info.getBrokerId(), e);
+ throw e;
}
- } else if (info.isNetworkConnection() && info.isDuplexConnection()) {
+ } else {
+ // duplex
// so this TransportConnection is the rear end of a network bridge
// We have been requested to create a two way pipe ...
try {
NetworkBridgeConfiguration config = getNetworkConfiguration(info);
config.setBrokerName(broker.getBrokerName());
- // register durable sync to be sent after ConnectionInfo has been handled
- registerDurableSync(config, info);
-
- // check for existing duplex connection hanging about
-
- // We first look if existing network connection already exists for the same broker Id and network connector name
- // It's possible in case of brief network fault to have this transport connector side of the connection always active
- // and the duplex network connector side wanting to open a new one
- // In this case, the old connection must be broken
- String duplexNetworkConnectorId = config.getName() + "@" + info.getBrokerId();
- CopyOnWriteArrayList<TransportConnection> connections = this.connector.getConnections();
- synchronized (connections) {
- for (TransportConnection c : connections) {
- if ((c != this) && (duplexNetworkConnectorId.equals(c.getDuplexNetworkConnectorId()))) {
- LOG.warn("Stopping an existing active duplex connection [{}] for network connector ({}).", c, duplexNetworkConnectorId);
- c.stopAsync();
- // better to wait for a bit rather than get connection id already in use and failure to start new bridge
- c.getStopped().await(1, TimeUnit.SECONDS);
- }
- }
- setDuplexNetworkConnectorId(duplexNetworkConnectorId);
- }
+ // Note: Durable sync used to be here and was moved to DemandForwardingBridgeSupport
+ // inside doStartLocalAndRemoteBridges()
+
+ //The logic to clean up existing network connections for the same ID
+ // has been moved to the RegionBroker where it will check if the broker
+ // needs to close the connection before trying to create a duplicate connection
+
Transport localTransport = NetworkBridgeFactory.createLocalTransport(config, broker.getVmConnectorURI());
Transport remoteBridgeTransport = transport;
if (! (remoteBridgeTransport instanceof ResponseCorrelator)) {
@@ -1462,46 +1490,13 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
duplexBridge.setCreatedByDuplex(true);
duplexBridge.duplexStart(this, brokerInfo, info);
LOG.info("Started responder end of duplex bridge {}", duplexNetworkConnectorId);
- return null;
} catch (TransportDisposedIOException e) {
LOG.warn("Duplex bridge {} was stopped before it was correctly started.", duplexNetworkConnectorId);
- return null;
} catch (Exception e) {
LOG.error("Failed to create responder end of duplex network bridge {}", duplexNetworkConnectorId, e);
- return null;
+ throw e;
}
}
- this.brokerInfo = info;
- networkConnection = true;
- List<TransportConnectionState> connectionStates = listConnectionStates();
- for (TransportConnectionState cs : connectionStates) {
- cs.getContext().setNetworkConnection(true);
- }
- return null;
- }
-
- private void registerDurableSync(final NetworkBridgeConfiguration config, final BrokerInfo info) {
- if (config.isSyncDurableSubs() && protocolVersion.get() >= CommandTypes.PROTOCOL_VERSION_DURABLE_SYNC) {
- // this will complete when the connection id has been set, or immediately if already set
- initialConnectionId.whenComplete((connectionId, t) -> {
- try {
- if (t != null) {
- LOG.warn("SyncDurableSubs will be skipped due to error {}",
- t.getMessage());
- return;
- }
- // check connection still registered
- if (lookupConnectionState(connectionId) != null) {
- LOG.debug("SyncDurableSubs is enabled, Sending BrokerSubscriptionInfo");
- dispatchSync(NetworkBridgeUtils.getBrokerSubscriptionInfo(
- this.broker.getBrokerService(), config));
- }
- } catch (Exception e) {
- LOG.error("Failed to respond to network bridge creation from broker {}",
- info.getBrokerId(), e);
- }
- });
- }
}
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -1707,7 +1702,7 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
this.duplexNetworkConnectorId = duplexNetworkConnectorId;
}
- protected synchronized String getDuplexNetworkConnectorId() {
+ public synchronized String getDuplexNetworkConnectorId() {
return this.duplexNetworkConnectorId;
}
@@ -1715,7 +1710,7 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
return stopping.get();
}
- protected CountDownLatch getStopped() {
+ public CountDownLatch getStopped() {
return stopped;
}
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/RegionBroker.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
index 2e6ee20497..7d5de4c184 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
@@ -29,6 +29,7 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -255,6 +256,10 @@ public class RegionBroker extends EmptyBroker {
throw new InvalidClientIDException("No clientID specified for connection request");
}
+ // Clean up existing duplex network connection if this is a reconnect attempt
+ // This was moved from TransportConnection
+ cleanupExistingDuplexNetworkConnection(context);
+
ConnectionContext oldContext = null;
synchronized (clientIdSet) {
@@ -289,6 +294,38 @@ public class RegionBroker extends EmptyBroker {
connections.add(context.getConnection());
}
+ // We first look if existing network connection already exists for the same broker Id and network connector name
+ // It's possible in case of brief network fault to have this transport connector side of the connection always active
+ // and the duplex network connector side wanting to open a new one
+ // In this case, the old connection must be broken
+ private void cleanupExistingDuplexNetworkConnection(ConnectionContext context) {
+ try {
+ if (context.isNetworkConnection()
+ && context.getConnection() instanceof TransportConnection) {
+ final TransportConnection newConn = (TransportConnection) context.getConnection();
+ if (newConn.getDuplexNetworkConnectorId() != null) {
+ for (Connection c : connections) {
+ if (c instanceof TransportConnection) {
+ final TransportConnection existingConn = (TransportConnection) c;
+ if (newConn.getDuplexNetworkConnectorId()
+ .equals(existingConn.getDuplexNetworkConnectorId())) {
+ LOG.warn("Stopping an existing active duplex connection [{}] for network connector ({}).",
+ c, existingConn.getDuplexNetworkConnectorId());
+ existingConn.stopAsync();
+ // better to wait for a bit rather than get connection id already in use and failure to start new bridge
+ existingConn.getStopped().await(2, TimeUnit.SECONDS);
+ break;
+ }
+ }
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("Error cleaning up Duplex connection: {}" , e.getMessage());
+ LOG.debug(e.getMessage(), e);
+ }
+ }
+
@Override
public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {
String clientId = info.getClientId();
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
index 26a5769ed0..17eb9c5529 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
@@ -477,6 +477,15 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
if (safeWaitUntilStarted()) {
setupStaticDestinations();
staticDestinationsLatch.countDown();
+
+ // Send to the remote broker the durable subs if sync is enabled after statup.
+ // This is done by the initiating side of a bridge as well as by duplex bridges to
+ // ensure everything is fully initialized before sending.
+ if (configuration.isSyncDurableSubs() &&
+ remoteBroker.getWireFormat().getVersion() >= CommandTypes.PROTOCOL_VERSION_DURABLE_SYNC) {
+ remoteBroker.oneway(NetworkBridgeUtils.getBrokerSubscriptionInfo(brokerService,
+ configuration));
+ }
}
} catch (Throwable e) {
serviceLocalException(e);
@@ -599,11 +608,6 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
brokerInfo.setNetworkProperties(str);
brokerInfo.setBrokerId(this.localBrokerId);
remoteBroker.oneway(brokerInfo);
- if (configuration.isSyncDurableSubs() &&
- remoteBroker.getWireFormat().getVersion() >= CommandTypes.PROTOCOL_VERSION_DURABLE_SYNC) {
- remoteBroker.oneway(NetworkBridgeUtils.getBrokerSubscriptionInfo(brokerService,
- configuration));
- }
}
if (remoteConnectionInfo != null) {
remoteBroker.oneway(remoteConnectionInfo.createRemoveCommand());
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeAuthTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeAuthTest.java
index 19c95da29b..3a99029617 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeAuthTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeAuthTest.java
@@ -16,10 +16,12 @@
*/
package org.apache.activemq.network;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
+import java.io.IOException;
import java.net.URI;
import java.util.Arrays;
import java.util.Collection;
@@ -39,8 +41,11 @@ import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.apache.activemq.store.kahadb.disk.journal.Journal.JournalDiskSyncStrategy;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFilter;
+import org.apache.activemq.transport.discovery.DiscoveryAgent;
+import org.apache.activemq.transport.discovery.DiscoveryListener;
import org.apache.activemq.util.Wait;
import org.junit.After;
+import org.junit.Assume;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -84,6 +89,7 @@ public class DurableSyncNetworkBridgeAuthTest extends AbstractDurableSyncNetwork
private static final String USER_PASSWORD = "password";
private final boolean duplex;
private final AtomicReference<BrokerSubscriptionInfo> brokerSubInfo = new AtomicReference<>();
+ private final AtomicReference<DiscoveryEvent> serviceFailed = new AtomicReference<>();
private String ncPassword = USER_PASSWORD;
public DurableSyncNetworkBridgeAuthTest(boolean duplex) {
@@ -94,6 +100,7 @@ public class DurableSyncNetworkBridgeAuthTest extends AbstractDurableSyncNetwork
public void setUp() throws Exception {
this.ncPassword = USER_PASSWORD;
this.brokerSubInfo.set(null);
+ this.serviceFailed.set(null);
}
@After
@@ -110,31 +117,77 @@ public class DurableSyncNetworkBridgeAuthTest extends AbstractDurableSyncNetwork
// automatically on connect so the remote broker will always receive it. However, the
// remote broker should only send back its list after the connection is properly authenticated.
assertTrue(Wait.waitFor(() -> brokerSubInfo.get() != null,5000,10));
+ assertNull(serviceFailed.get());
+ DemandForwardingBridge bridge = getActiveBridge();
// Simulate a connection exception and reconnect, we should receive again
brokerSubInfo.set(null);
localBroker.getNetworkConnectors().get(0).activeBridges().stream()
.findFirst().orElseThrow().serviceRemoteException(new Exception());
+ // wait for failure
+ assertTrue(Wait.waitFor(() -> serviceFailed.get() != null,5000,10));
+ assertTrue(Wait.waitFor(bridge.localBroker::isDisposed,5000,10));
+
+ // should reconnect again and get updated info
+ assertTrue(Wait.waitFor(() -> localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1,
+ TimeUnit.SECONDS.toMillis(5), 10));
assertTrue(Wait.waitFor(() -> brokerSubInfo.get() != null,5000,10));
}
@Test
public void testAuthFailure() throws Exception {
this.ncPassword = "badpassword";
- try {
- // set a shorter wait time, it won't connect with bad password
- doSetUp(true, true, tempFolder.newFolder(), tempFolder.newFolder(),
- TimeUnit.SECONDS.toMillis(5));
- throw new IllegalStateException("Should have received assertion error with bad password");
- } catch (AssertionError e) {
- // expected
- }
+ doSetUpRemoteBroker(true, tempFolder.newFolder(), 0);
+ doSetUpLocalBroker(true, true, tempFolder.newFolder());
+ // Wait for the failure due to authentication
+ assertTrue(Wait.waitFor(() -> serviceFailed.get() != null,5000,10));
+ assertTrue(Wait.waitFor(() -> localBroker.getNetworkConnectors().get(0).activeBridges().isEmpty(),
+ TimeUnit.SECONDS.toMillis(5), 10));
// Because the local broker was not authenticated by the remote broker, the local broker
// should not have received back the BrokerSubscriptionInfo
assertNull(brokerSubInfo.get());
}
+ @Test
+ public void testDuplicateDuplexBridgeFailedAuthIgnored() throws Exception {
+ Assume.assumeTrue(duplex);
+ doSetUp(true, true, tempFolder.newFolder(), tempFolder.newFolder(),
+ TimeUnit.SECONDS.toMillis(15));
+
+ // everything is good, no error and we got the sync command
+ assertTrue(Wait.waitFor(() -> brokerSubInfo.get() != null,5000,10));
+ assertNull(serviceFailed.get());
+
+ // Start a duplicate bridge with the same configuration but bad password
+ // so authentication fails. This should not cause a failure with the existing
+ // bridge because this connection won't be authenticated
+ DemandForwardingBridge bridge = getActiveBridge();
+ this.ncPassword = "badpassword";
+ NetworkConnector nc = localBroker.addNetworkConnector(configureLocalNetworkConnector());
+ nc.start();
+ try {
+ Thread.sleep(2000);
+ // Verify bridge is not disposed and still connected
+ assertFalse(bridge.disposed.get());
+ } finally {
+ nc.stop();
+ }
+
+ // try again, this will connect successfully and the broker will detect it's a duplex bridge
+ // matching the same config and close the other
+ this.ncPassword = USER_PASSWORD;
+ nc = localBroker.addNetworkConnector(configureLocalNetworkConnector());
+ nc.start();
+ try {
+ // authentication is now correct so the RegionBroker should terminate the other duplex
+ // bridge as it matches
+ assertTrue(Wait.waitFor(bridge.disposed::get,5000,10));
+ } finally {
+ nc.stop();
+ }
+ }
+
@Test
public void testRestartSync() throws Exception {
doSetUp(true, true, tempFolder.newFolder(), tempFolder.newFolder(),
@@ -144,12 +197,17 @@ public class DurableSyncNetworkBridgeAuthTest extends AbstractDurableSyncNetwork
// automatically on connect so the remote broker will always receive it. However, the
// remote broker should only send back its list after the connection is properly authenticated.
assertTrue(Wait.waitFor(() -> brokerSubInfo.get() != null,5000,10));
+ assertNull(serviceFailed.get());
// Restart, should receive again with new connection
brokerSubInfo.set(null);
restartRemoteBroker();
+ // should fail from restart
+ assertTrue(Wait.waitFor(() -> serviceFailed.get() != null,5000,10));
// Wait for the reconnect and receive of BrokerSubInfo
+ assertTrue(Wait.waitFor(() -> localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1,
+ TimeUnit.SECONDS.toMillis(5), 10));
assertTrue(Wait.waitFor(() -> brokerSubInfo.get() != null,5000,10));
}
@@ -159,10 +217,10 @@ public class DurableSyncNetworkBridgeAuthTest extends AbstractDurableSyncNetwork
doSetUp(true, true, tempFolder.newFolder(), tempFolder.newFolder(),
TimeUnit.SECONDS.toMillis(15));
assertTrue(Wait.waitFor(() -> brokerSubInfo.get() != null,5000,10));
+ assertNull(serviceFailed.get());
// find the established bridge
- DemandForwardingBridge bridge = (DemandForwardingBridge) localBroker.getNetworkConnectors().get(0).activeBridges().stream()
- .findFirst().orElseThrow();
+ DemandForwardingBridge bridge = getActiveBridge();
// send to one of the brokers (networked brokers will have already received a BrokerInfo)
// the duplicate will trigger the bridge connection to close
@@ -232,6 +290,9 @@ public class DurableSyncNetworkBridgeAuthTest extends AbstractDurableSyncNetwork
URI remoteURI = transportConnectors.get(0).getConnectUri();
String uri = "static:(" + remoteURI + ")";
NetworkConnector connector = new DiscoveryNetworkConnector(new URI(uri)) {
+ {
+ this.setDiscoveryAgent(new DiscoveryAgentFilter(getDiscoveryAgent()));
+ }
@Override
protected NetworkBridge createBridge(Transport localTransport,
Transport remoteTransport, DiscoveryEvent event) {
@@ -248,6 +309,7 @@ public class DurableSyncNetworkBridgeAuthTest extends AbstractDurableSyncNetwork
}
super.onCommand(command);
}
+
};
return super.createBridge(localTransport, remoteFilter, event);
}
@@ -285,4 +347,37 @@ public class DurableSyncNetworkBridgeAuthTest extends AbstractDurableSyncNetwork
return brokerService;
}
+ private DemandForwardingBridge getActiveBridge() {
+ return(DemandForwardingBridge) localBroker.getNetworkConnectors().get(0).activeBridges().stream()
+ .findFirst().orElseThrow();
+ }
+
+ private class DiscoveryAgentFilter implements DiscoveryAgent {
+ private final DiscoveryAgent agent;
+
+ public DiscoveryAgentFilter(DiscoveryAgent agent) {
+ this.agent = agent;
+ }
+
+ public void setDiscoveryListener(DiscoveryListener listener) {
+ agent.setDiscoveryListener(listener);
+ }
+
+ public void start() throws Exception {
+ agent.start();
+ }
+
+ public void stop() throws Exception {
+ agent.stop();
+ }
+
+ public void registerService(String name) throws IOException {
+ agent.registerService(name);
+ }
+
+ public void serviceFailed(DiscoveryEvent event) throws IOException {
+ serviceFailed.set(event);
+ agent.serviceFailed(event);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact