| /* |
| * 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.activemq.artemis.tests.integration.jms.cluster; |
| |
| import javax.jms.Connection; |
| import javax.jms.DeliveryMode; |
| import javax.jms.MessageConsumer; |
| import javax.jms.MessageProducer; |
| import javax.jms.Session; |
| import javax.jms.TextMessage; |
| import javax.jms.Topic; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.apache.activemq.artemis.api.core.ActiveMQException; |
| import org.apache.activemq.artemis.api.core.ActiveMQNotConnectedException; |
| import org.apache.activemq.artemis.api.core.TransportConfiguration; |
| import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient; |
| import org.apache.activemq.artemis.api.jms.JMSFactoryType; |
| import org.apache.activemq.artemis.core.remoting.FailureListener; |
| import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnectorFactory; |
| import org.apache.activemq.artemis.core.server.cluster.Bridge; |
| import org.apache.activemq.artemis.core.server.cluster.impl.BridgeImpl; |
| import org.apache.activemq.artemis.core.server.cluster.impl.ClusterConnectionImpl; |
| import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; |
| import org.apache.activemq.artemis.tests.util.JMSClusteredTestBase; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.junit.runner.RunWith; |
| import org.junit.runners.Parameterized; |
| |
| @RunWith(value = Parameterized.class) |
| public class BindingsClusterTest extends JMSClusteredTestBase { |
| |
| // TODO: find a solution to this |
| // the "jms." prefix is needed because the cluster connection is matching on this |
| public static final String TOPIC = "jms.t1"; |
| |
| private final boolean crash; |
| |
| public BindingsClusterTest(boolean crash) { |
| this.crash = crash; |
| } |
| |
| @Parameterized.Parameters(name = "crash={0}") |
| public static Collection getParameters() { |
| return Arrays.asList(new Object[][]{{true}, {false}}); |
| } |
| |
| @Override |
| @Before |
| public void setUp() throws Exception { |
| //todo fix if needed |
| super.setUp(); |
| jmsServer1.getActiveMQServer().setIdentity("Server 1"); |
| jmsServer2.getActiveMQServer().setIdentity("Server 2"); |
| } |
| |
| @Override |
| protected boolean enablePersistence() { |
| return true; |
| } |
| |
| @Test |
| public void testSendToSingleDisconnectedBinding() throws Exception { |
| Connection conn1 = cf1.createConnection(); |
| |
| conn1.setClientID("someClient1"); |
| |
| Connection conn2 = cf2.createConnection(); |
| |
| conn2.setClientID("someClient2"); |
| |
| conn1.start(); |
| |
| conn2.start(); |
| |
| try { |
| |
| Topic topic1 = createTopic(TOPIC, true); |
| |
| Topic topic2 = (Topic) context1.lookup("topic/" + TOPIC); |
| |
| Session session1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| Session session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| MessageConsumer cons2 = session2.createDurableSubscriber(topic2, "sub2"); |
| |
| cons2.close(); |
| |
| session2.close(); |
| |
| conn2.close(); |
| |
| Thread.sleep(500); |
| |
| MessageProducer prod1 = session1.createProducer(topic1); |
| |
| prod1.setDeliveryMode(DeliveryMode.PERSISTENT); |
| |
| prod1.send(session1.createTextMessage("m1")); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| printBindings(jmsServer2.getActiveMQServer(), TOPIC); |
| |
| crash(); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| |
| prod1.send(session1.createTextMessage("m2")); |
| |
| restart(); |
| |
| Thread.sleep(2000); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| printBindings(jmsServer2.getActiveMQServer(), TOPIC); |
| |
| prod1.send(session1.createTextMessage("m3")); |
| |
| cf2 = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName(), generateInVMParams(2))); |
| |
| conn2 = cf2.createConnection(); |
| |
| conn2.setClientID("someClient2"); |
| |
| session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| cons2 = session2.createDurableSubscriber(topic2, "sub2"); |
| |
| conn2.start(); |
| |
| TextMessage received = (TextMessage) cons2.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m1", received.getText()); |
| |
| received = (TextMessage) cons2.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m2", received.getText()); |
| |
| received = (TextMessage) cons2.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m3", received.getText()); |
| |
| cons2.close(); |
| } finally { |
| conn1.close(); |
| conn2.close(); |
| } |
| |
| jmsServer1.destroyTopic(TOPIC); |
| jmsServer2.destroyTopic(TOPIC); |
| } |
| |
| @Test |
| public void testSendToSingleDisconnectedBindingWhenLocalAvailable() throws Exception { |
| Connection conn1 = cf1.createConnection(); |
| |
| conn1.setClientID("someClient2"); |
| |
| Connection conn2 = cf2.createConnection(); |
| |
| conn2.setClientID("someClient2"); |
| |
| conn1.start(); |
| |
| conn2.start(); |
| |
| try { |
| |
| Topic topic1 = createTopic(TOPIC, true); |
| |
| Topic topic2 = (Topic) context1.lookup("topic/" + TOPIC); |
| |
| Session session1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| Session session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| MessageConsumer cons1 = session1.createDurableSubscriber(topic1, "sub2"); |
| |
| MessageConsumer cons2 = session2.createDurableSubscriber(topic2, "sub2"); |
| |
| cons2.close(); |
| |
| session2.close(); |
| |
| conn2.close(); |
| |
| Thread.sleep(500); |
| |
| MessageProducer prod1 = session1.createProducer(topic1); |
| |
| prod1.setDeliveryMode(DeliveryMode.PERSISTENT); |
| |
| prod1.send(session1.createTextMessage("m1")); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| printBindings(jmsServer2.getActiveMQServer(), TOPIC); |
| |
| crash(); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| |
| //send a few messages while the binding is disconnected |
| prod1.send(session1.createTextMessage("m2")); |
| prod1.send(session1.createTextMessage("m3")); |
| prod1.send(session1.createTextMessage("m4")); |
| |
| restart(); |
| |
| Thread.sleep(2000); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| printBindings(jmsServer2.getActiveMQServer(), TOPIC); |
| |
| prod1.send(session1.createTextMessage("m5")); |
| prod1.send(session1.createTextMessage("m6")); |
| |
| cf2 = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName(), generateInVMParams(2))); |
| |
| conn2 = cf2.createConnection(); |
| |
| conn2.setClientID("someClient2"); |
| |
| session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| cons2 = session2.createDurableSubscriber(topic2, "sub2"); |
| |
| conn2.start(); |
| |
| TextMessage received = (TextMessage) cons2.receiveNoWait(); |
| |
| assertNull(received); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m1", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m2", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m3", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m4", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m5", received.getText()); |
| |
| cons2.close(); |
| } finally { |
| conn1.close(); |
| conn2.close(); |
| } |
| |
| jmsServer1.destroyTopic(TOPIC); |
| jmsServer2.destroyTopic(TOPIC); |
| } |
| |
| @Test |
| public void testRemoteBindingRemovedOnReconnectLocalAvailable() throws Exception { |
| Connection conn1 = cf1.createConnection(); |
| |
| conn1.setClientID("someClient2"); |
| |
| Connection conn2 = cf2.createConnection(); |
| |
| conn2.setClientID("someClient2"); |
| |
| conn1.start(); |
| |
| conn2.start(); |
| |
| try { |
| |
| Topic topic1 = createTopic(TOPIC, true); |
| |
| Topic topic2 = (Topic) context1.lookup("topic/" + TOPIC); |
| |
| Session session1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| Session session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| MessageConsumer cons1 = session1.createSharedConsumer(topic1, "sub2"); |
| |
| MessageConsumer cons2 = session2.createSharedConsumer(topic2, "sub2"); |
| |
| Thread.sleep(500); |
| |
| MessageProducer prod1 = session1.createProducer(topic1); |
| |
| prod1.setDeliveryMode(DeliveryMode.PERSISTENT); |
| |
| prod1.send(session1.createTextMessage("m1")); |
| prod1.send(session1.createTextMessage("m2")); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| printBindings(jmsServer2.getActiveMQServer(), TOPIC); |
| |
| // verify receipt by remote binding before crash |
| TextMessage received = (TextMessage) cons2.receive(5000); |
| assertNotNull(received); |
| assertEquals("m2", received.getText()); |
| |
| crash(); |
| |
| //this may or may not be closed, if the server was crashed then it would have been closed on failure. |
| cons2.close(); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| |
| //send a few messages while the binding is disconnected |
| prod1.send(session1.createTextMessage("m3")); |
| prod1.send(session1.createTextMessage("m4")); |
| prod1.send(session1.createTextMessage("m5")); |
| |
| restart(); |
| |
| Thread.sleep(2000); |
| |
| printBindings(jmsServer1.getActiveMQServer(), TOPIC); |
| printBindings(jmsServer2.getActiveMQServer(), TOPIC); |
| |
| prod1.send(session1.createTextMessage("m6")); |
| prod1.send(session1.createTextMessage("m7")); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m1", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m3", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m4", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m5", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m6", received.getText()); |
| |
| received = (TextMessage) cons1.receive(5000); |
| |
| assertNotNull(received); |
| |
| assertEquals("m7", received.getText()); |
| |
| cons2.close(); |
| } finally { |
| conn1.close(); |
| conn2.close(); |
| } |
| |
| jmsServer1.destroyTopic(TOPIC); |
| jmsServer2.destroyTopic(TOPIC); |
| } |
| |
| private void crash() throws Exception { |
| /* |
| * Rather than just calling stop() on the server here we want to simulate an actual node crash or bridge failure |
| * so the bridge's failure listener needs to get something other than a DISCONNECTED message. In this case we |
| * simulate a NOT_CONNECTED exception. |
| */ |
| final CountDownLatch latch = new CountDownLatch(1); |
| ClusterConnectionImpl next = (ClusterConnectionImpl) server1.getClusterManager().getClusterConnections().iterator().next(); |
| BridgeImpl bridge = (BridgeImpl) next.getRecords().values().iterator().next().getBridge(); |
| RemotingConnection forwardingConnection = getForwardingConnection(bridge); |
| forwardingConnection.addFailureListener(new FailureListener() { |
| @Override |
| public void connectionFailed(ActiveMQException exception, boolean failedOver) { |
| latch.countDown(); |
| } |
| |
| @Override |
| public void connectionFailed(final ActiveMQException me, boolean failedOver, String scaleDownTargetNodeID) { |
| connectionFailed(me, failedOver); |
| } |
| }); |
| forwardingConnection.fail(new ActiveMQNotConnectedException()); |
| assertTrue(latch.await(5000, TimeUnit.MILLISECONDS)); |
| |
| if (crash) { |
| jmsServer2.stop(); |
| } |
| } |
| |
| private void restart() throws Exception { |
| if (crash) { |
| jmsServer2.start(); |
| } |
| } |
| |
| private RemotingConnection getForwardingConnection(final Bridge bridge) throws Exception { |
| long start = System.currentTimeMillis(); |
| |
| do { |
| RemotingConnection forwardingConnection = ((BridgeImpl) bridge).getForwardingConnection(); |
| |
| if (forwardingConnection != null) { |
| return forwardingConnection; |
| } |
| |
| Thread.sleep(10); |
| } |
| while (System.currentTimeMillis() - start < 50000); |
| |
| throw new IllegalStateException("Failed to get forwarding connection"); |
| } |
| } |
