diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java index 1876b723a4c7..85a9093d5fb5 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java @@ -150,6 +150,8 @@ public class TransportConstants { public static final String TRUST_MANAGER_FACTORY_PLUGIN_PROP_NAME = "trustManagerFactoryPlugin"; + public static final String CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME = "clientFailoverAdvertisingEnabled"; + public static final String NETTY_VERSION; /** diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java index 21d9373f5598..3cd841585814 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java @@ -27,10 +27,12 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffers; import org.apache.activemq.artemis.api.core.ActiveMQException; import org.apache.activemq.artemis.api.core.Message; +import org.apache.activemq.artemis.api.core.TransportConfiguration; import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper; import org.apache.activemq.artemis.core.client.impl.TopologyMemberImpl; import org.apache.activemq.artemis.core.remoting.CloseListener; import org.apache.activemq.artemis.core.remoting.FailureListener; +import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants; import org.apache.activemq.artemis.core.server.ActiveMQServer; import org.apache.activemq.artemis.core.server.cluster.ClusterConnection; import org.apache.activemq.artemis.core.server.cluster.ClusterManager; @@ -49,12 +51,14 @@ import org.apache.activemq.artemis.protocol.amqp.sasl.ServerSASLFactory; import org.apache.activemq.artemis.spi.core.remoting.Connection; import org.apache.activemq.artemis.spi.core.remoting.ReadyListener; +import org.apache.activemq.artemis.utils.ConfigurationHelper; import org.apache.activemq.artemis.utils.UUIDGenerator; import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.Symbol; import org.apache.qpid.proton.amqp.transport.AmqpError; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.lang.invoke.MethodHandles; import io.netty.buffer.ByteBuf; @@ -191,7 +195,7 @@ public AMQPSessionCallback createSessionCallback(AMQPConnectionContext connectio } public void sendSASLSupported() { - connection.write(ActiveMQBuffers.wrappedBuffer(new byte[]{'A', 'M', 'Q', 'P', 3, 1, 0, 0})); + connection.write(ActiveMQBuffers.wrappedBuffer(new byte[] {'A', 'M', 'Q', 'P', 3, 1, 0, 0})); } public boolean validateConnection(org.apache.qpid.proton.engine.Connection connection, SASLResult saslResult) { @@ -266,7 +270,14 @@ public URI getFailoverList() { if (clusterConnection != null) { TopologyMemberImpl member = clusterConnection.getTopology().getMember(server.getNodeID().toString()); if (member != null) { - return member.toBackupURI(); + TransportConfiguration backupConnector = member.getBackup(); + if (backupConnector == null) { + return null; + } + boolean clientFailoverAdvertisingEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME, true, backupConnector.getCombinedParams()); + if (clientFailoverAdvertisingEnabled) { + return member.toBackupURI(); + } } } return null; diff --git a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallbackTest.java b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallbackTest.java index 0cf0dcb49c49..38c1d8ecf601 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallbackTest.java +++ b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallbackTest.java @@ -23,10 +23,18 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import org.apache.activemq.artemis.api.core.ActiveMQSecurityException; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.api.core.TransportConfiguration; +import org.apache.activemq.artemis.core.client.impl.Topology; +import org.apache.activemq.artemis.core.client.impl.TopologyMemberImpl; import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnection; import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnection; +import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory; +import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants; import org.apache.activemq.artemis.core.security.SecurityStore; import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.cluster.ClusterConnection; +import org.apache.activemq.artemis.core.server.cluster.ClusterManager; import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl; import org.apache.activemq.artemis.protocol.amqp.sasl.AnonymousServerSASL; import org.apache.activemq.artemis.protocol.amqp.sasl.GSSAPIServerSASL; @@ -36,6 +44,10 @@ import org.junit.jupiter.api.Test; import org.mockito.Mockito; +import java.net.URI; +import java.util.HashMap; +import java.util.Map; + public class AMQPConnectionCallbackTest { @Test @@ -93,4 +105,122 @@ public void testAnonymousSupportCheck() throws Exception { assertFalse(callback.isSupportsAnonymous()); Mockito.verify(securityStore).authenticate(Mockito.any(), Mockito.any(), Mockito.same(connectionDelegate)); } + + @Test + public void testGetFailoverListFailoverEnabled() throws Exception { + ActiveMQServer server = Mockito.mock(ActiveMQServer.class); + + TransportConfiguration backup = createBackupTransportConfiguration(); + backup.getParams().put(TransportConstants.CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME, true); + + AMQPConnectionCallback callback = createCallback(server, backup); + + URI failoverList = callback.getFailoverList(); + + assertNotNull(failoverList); + assertEquals("tcp", failoverList.getScheme()); + assertEquals("backup", failoverList.getHost()); + assertEquals(61617, failoverList.getPort()); + assertEquals("sslEnabled=false", failoverList.getQuery()); + } + + @Test + public void testGetFailoverListFailoverDisabled() throws Exception { + ActiveMQServer server = Mockito.mock(ActiveMQServer.class); + + TransportConfiguration backup = createBackupTransportConfiguration(); + backup.getParams().put(TransportConstants.CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME, false); + + AMQPConnectionCallback callback = createCallback(server, backup); + + assertNull(callback.getFailoverList()); + } + + @Test + public void testGetFailoverListFailoverEnabledByDefault() throws Exception { + ActiveMQServer server = Mockito.mock(ActiveMQServer.class); + + TransportConfiguration backup = createBackupTransportConfiguration(); + + // CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME is intentionally not set. + // The default value is true. + AMQPConnectionCallback callback = createCallback(server, backup); + + URI failoverList = callback.getFailoverList(); + + assertNotNull(failoverList); + assertEquals("tcp", failoverList.getScheme()); + assertEquals("backup", failoverList.getHost()); + assertEquals(61617, failoverList.getPort()); + assertEquals("sslEnabled=false", failoverList.getQuery()); + } + + @Test + public void testGetFailoverListIncludesSslEnabled() throws Exception { + ActiveMQServer server = Mockito.mock(ActiveMQServer.class); + + TransportConfiguration backup = createBackupTransportConfiguration(); + backup.getParams().put( + TransportConstants.SSL_ENABLED_PROP_NAME, + true); + backup.getParams().put( + TransportConstants.CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME, + true); + + AMQPConnectionCallback callback = createCallback(server, backup); + + URI failoverList = callback.getFailoverList(); + + assertNotNull(failoverList); + assertEquals("tcp", failoverList.getScheme()); + assertEquals("backup", failoverList.getHost()); + assertEquals(61617, failoverList.getPort()); + assertEquals("sslEnabled=true", failoverList.getQuery()); + } + + private TransportConfiguration createBackupTransportConfiguration() { + Map params = new HashMap<>(); + params.put(TransportConstants.HOST_PROP_NAME, "backup"); + params.put(TransportConstants.PORT_PROP_NAME, 61617); + + return new TransportConfiguration( + NettyConnectorFactory.class.getName(), + params); + } + + + private AMQPConnectionCallback createCallback( + ActiveMQServer server, + TransportConfiguration backup) throws Exception { + + String nodeId = "test-node"; + + Mockito.when(server.getNodeID()).thenReturn(SimpleString.of(nodeId)); + + ClusterManager clusterManager = Mockito.mock(ClusterManager.class); + ClusterConnection clusterConnection = Mockito.mock(ClusterConnection.class); + Topology topology = Mockito.mock(Topology.class); + + Mockito.when(server.getClusterManager()).thenReturn(clusterManager); + Mockito.when(clusterManager.getDefaultConnection(null)).thenReturn(clusterConnection); + Mockito.when(clusterConnection.getTopology()).thenReturn(topology); + + TopologyMemberImpl member = new TopologyMemberImpl( + nodeId, + null, + null, + null, + backup); + + Mockito.when(topology.getMember(nodeId)).thenReturn(member); + + ProtonProtocolManager protocolManager = Mockito.mock(ProtonProtocolManager.class); + Mockito.when(protocolManager.getServer()).thenReturn(server); + + return new AMQPConnectionCallback( + protocolManager, + null, + null, + server); + } } \ No newline at end of file diff --git a/docs/user-manual/configuring-transports.adoc b/docs/user-manual/configuring-transports.adoc index e2aeac6005d6..5e1fb98c9f2d 100644 --- a/docs/user-manual/configuring-transports.adoc +++ b/docs/user-manual/configuring-transports.adoc @@ -491,6 +491,11 @@ This value takes precedence of all other SSL parameters which apply to the trust + Any plugin specified will need to be placed on the xref:using-server.adoc#adding-runtime-dependencies[broker's classpath]. +clientFailoverAdvertisingEnabled:: +When used on a `connector` determines whether the connector is advertised to clients as a failover connector. +Valid values are `true` or `false`. +Default is `true`. + ==== Configuring an SSLContextFactory If you use `JDK` as SSL provider (the default), you can configure which SSLContextFactory to use.