| /* |
| * 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.ignite.messaging; |
| |
| import java.io.Externalizable; |
| import java.io.IOException; |
| import java.io.ObjectInput; |
| import java.io.ObjectOutput; |
| import java.io.Serializable; |
| import java.net.URL; |
| import java.net.URLClassLoader; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.UUID; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ConcurrentLinkedDeque; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import org.apache.ignite.Ignite; |
| import org.apache.ignite.IgniteMessaging; |
| import org.apache.ignite.cluster.ClusterGroup; |
| import org.apache.ignite.internal.DiscoverySpiTestListener; |
| import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi; |
| import org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessage; |
| import org.apache.ignite.internal.processors.continuous.StartRoutineDiscoveryMessageV2; |
| import org.apache.ignite.internal.processors.continuous.StopRoutineDiscoveryMessage; |
| import org.apache.ignite.internal.util.GridConcurrentHashSet; |
| import org.apache.ignite.internal.util.typedef.P2; |
| import org.apache.ignite.internal.util.typedef.PA; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteFuture; |
| import org.apache.ignite.resources.IgniteInstanceResource; |
| import org.apache.ignite.testframework.GridTestUtils; |
| import org.apache.ignite.testframework.config.GridTestProperties; |
| import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; |
| import org.junit.Assert; |
| import org.junit.Test; |
| |
| import static org.apache.ignite.testframework.GridTestUtils.assertThrows; |
| |
| /** |
| * Various tests for Messaging public API. |
| */ |
| public class GridMessagingSelfTest extends GridCommonAbstractTest implements Serializable { |
| /** */ |
| private static final String MSG_1 = "MSG-1"; |
| |
| /** */ |
| private static final String MSG_2 = "MSG-2"; |
| |
| /** */ |
| private static final String MSG_3 = "MSG-3"; |
| |
| /** */ |
| private static final String S_TOPIC_1 = "TOPIC-1"; |
| |
| /** */ |
| private static final String S_TOPIC_2 = "TOPIC-2"; |
| |
| /** */ |
| private static final Integer I_TOPIC_1 = 1; |
| |
| /** */ |
| private static final Integer I_TOPIC_2 = 2; |
| |
| /** Message count. */ |
| private static AtomicInteger MSG_CNT; |
| |
| /** */ |
| public static final String EXT_RESOURCE_CLS_NAME = "org.apache.ignite.tests.p2p.TestUserResource"; |
| |
| /** */ |
| protected static CountDownLatch rcvLatch; |
| |
| /** |
| * A test message topic. |
| */ |
| private enum TestTopic { |
| /** */ |
| TOPIC_1, |
| |
| /** */ |
| TOPIC_2 |
| } |
| |
| /** |
| * A test message with a hack for delay |
| * emulation. |
| */ |
| private static class TestMessage implements Externalizable { |
| /** */ |
| private Object body; |
| |
| /** */ |
| private long delayMs; |
| |
| /** |
| * No-arg constructor for {@link Externalizable}. |
| */ |
| public TestMessage() { |
| // No-op. |
| } |
| |
| /** |
| * @param body Message body. |
| */ |
| TestMessage(Object body) { |
| this.body = body; |
| } |
| |
| /** |
| * @param body Message body. |
| * @param delayMs Message send delay in milliseconds. |
| */ |
| TestMessage(Object body, long delayMs) { |
| this.body = body; |
| this.delayMs = delayMs; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return "TestMessage [body=" + body + "]"; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public int hashCode() { |
| return body.hashCode(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean equals(Object obj) { |
| return obj instanceof TestMessage && body.equals(((TestMessage)obj).body); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeExternal(ObjectOutput out) throws IOException { |
| if (delayMs > 0) { |
| try { |
| Thread.sleep(delayMs); |
| } |
| catch (InterruptedException ignored) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| out.writeObject(body); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { |
| body = in.readObject(); |
| } |
| } |
| |
| /** */ |
| protected Ignite ignite1; |
| |
| /** */ |
| protected Ignite ignite2; |
| |
| /** {@inheritDoc} */ |
| @Override protected void beforeTest() throws Exception { |
| MSG_CNT = new AtomicInteger(); |
| |
| ignite1 = startGrid(1); |
| ignite2 = startGrid(2); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void afterTest() throws Exception { |
| stopAllGrids(); |
| |
| ignite1 = null; |
| ignite2 = null; |
| } |
| |
| /** |
| * Tests simple message sending-receiving. |
| * |
| * @throws Exception If error occurs. |
| */ |
| @Test |
| public void testSendReceiveMessage() throws Exception { |
| final Collection<Object> rcvMsgs = new GridConcurrentHashSet<>(); |
| |
| final AtomicBoolean error = new AtomicBoolean(false); //to make it modifiable |
| |
| final CountDownLatch rcvLatch = new CountDownLatch(3); |
| |
| ignite1.message().localListen(null, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| if (!nodeId.equals(ignite2.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ClusterGroup rNode1 = ignite2.cluster().forRemotes(); |
| |
| message(rNode1).send(null, MSG_1); |
| message(rNode1).send(null, MSG_2); |
| message(rNode1).send(null, MSG_3); |
| |
| assertTrue(rcvLatch.await(3, TimeUnit.SECONDS)); |
| |
| assertFalse(error.get()); |
| |
| assertTrue(rcvMsgs.contains(MSG_1)); |
| assertTrue(rcvMsgs.contains(MSG_2)); |
| assertTrue(rcvMsgs.contains(MSG_3)); |
| } |
| |
| /** |
| * @throws Exception If error occurs. |
| */ |
| @SuppressWarnings("TooBroadScope") |
| @Test |
| public void testStopLocalListen() throws Exception { |
| final AtomicInteger msgCnt1 = new AtomicInteger(); |
| |
| final AtomicInteger msgCnt2 = new AtomicInteger(); |
| |
| final AtomicInteger msgCnt3 = new AtomicInteger(); |
| |
| P2<UUID, Object> lsnr1 = new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| log.info("Listener1 received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| msgCnt1.incrementAndGet(); |
| |
| return true; |
| } |
| }; |
| |
| P2<UUID, Object> lsnr2 = new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| log.info("Listener2 received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| msgCnt2.incrementAndGet(); |
| |
| return true; |
| } |
| }; |
| |
| P2<UUID, Object> lsnr3 = new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| log.info("Listener3 received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| msgCnt3.incrementAndGet(); |
| |
| return true; |
| } |
| }; |
| |
| final String topic1 = null; |
| final String topic2 = "top1"; |
| final String topic3 = "top3"; |
| |
| ignite1.message().localListen(topic1, lsnr1); |
| ignite1.message().localListen(topic2, lsnr2); |
| ignite1.message().localListen(topic3, lsnr3); |
| |
| ClusterGroup rNode1 = ignite2.cluster().forRemotes(); |
| |
| message(rNode1).send(topic1, "msg1-1"); |
| message(rNode1).send(topic2, "msg1-2"); |
| message(rNode1).send(topic3, "msg1-3"); |
| |
| GridTestUtils.waitForCondition(new PA() { |
| @Override public boolean apply() { |
| return msgCnt1.get() > 0 && msgCnt2.get() > 0 && msgCnt3.get() > 0; |
| } |
| }, 5000); |
| |
| assertEquals(1, msgCnt1.get()); |
| assertEquals(1, msgCnt2.get()); |
| assertEquals(1, msgCnt3.get()); |
| |
| ignite1.message().stopLocalListen(topic2, lsnr2); |
| |
| message(rNode1).send(topic1, "msg2-1"); |
| message(rNode1).send(topic2, "msg2-2"); |
| message(rNode1).send(topic3, "msg2-3"); |
| |
| GridTestUtils.waitForCondition(new PA() { |
| @Override public boolean apply() { |
| return msgCnt1.get() > 1 && msgCnt3.get() > 1; |
| } |
| }, 5000); |
| |
| assertEquals(2, msgCnt1.get()); |
| assertEquals(1, msgCnt2.get()); |
| assertEquals(2, msgCnt3.get()); |
| |
| ignite1.message().stopLocalListen(topic2, lsnr1); // Try to use wrong topic for lsnr1 removing. |
| |
| message(rNode1).send(topic1, "msg3-1"); |
| message(rNode1).send(topic2, "msg3-2"); |
| message(rNode1).send(topic3, "msg3-3"); |
| |
| GridTestUtils.waitForCondition(new PA() { |
| @Override public boolean apply() { |
| return msgCnt1.get() > 2 && msgCnt3.get() > 2; |
| } |
| }, 5000); |
| |
| assertEquals(3, msgCnt1.get()); |
| assertEquals(1, msgCnt2.get()); |
| assertEquals(3, msgCnt3.get()); |
| |
| ignite1.message().stopLocalListen(topic1, lsnr1); |
| ignite1.message().stopLocalListen(topic3, lsnr3); |
| |
| message(rNode1).send(topic1, "msg4-1"); |
| message(rNode1).send(topic2, "msg4-2"); |
| message(rNode1).send(topic3, "msg4-3"); |
| |
| U.sleep(1000); |
| |
| assertEquals(3, msgCnt1.get()); |
| assertEquals(1, msgCnt2.get()); |
| assertEquals(3, msgCnt3.get()); |
| } |
| |
| /** |
| * Tests simple message sending-receiving with string topic. |
| * |
| * @throws Exception If error occurs. |
| */ |
| @Test |
| public void testSendReceiveMessageWithStringTopic() throws Exception { |
| final Collection<Object> rcvMsgs = new GridConcurrentHashSet<>(); |
| |
| final AtomicBoolean error = new AtomicBoolean(false); //to make it modifiable |
| |
| final CountDownLatch rcvLatch = new CountDownLatch(3); |
| |
| ignite1.message().localListen(S_TOPIC_1, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=" + S_TOPIC_1 + ']'); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_1.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: " + S_TOPIC_1); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ignite1.message().localListen(S_TOPIC_2, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=" + S_TOPIC_2 + ']'); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_2.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: " + S_TOPIC_2); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ignite1.message().localListen(null, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=default]"); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_3.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: default"); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ClusterGroup rNode1 = ignite1.cluster().forLocal(); |
| |
| message(rNode1).send(S_TOPIC_1, MSG_1); |
| message(rNode1).send(S_TOPIC_2, MSG_2); |
| message(rNode1).send(null, MSG_3); |
| |
| assertTrue(rcvLatch.await(3, TimeUnit.SECONDS)); |
| |
| assertFalse(error.get()); |
| |
| assertTrue(rcvMsgs.contains(MSG_1)); |
| assertTrue(rcvMsgs.contains(MSG_2)); |
| assertTrue(rcvMsgs.contains(MSG_3)); |
| } |
| |
| /** |
| * Tests simple message sending-receiving with enumerated topic. |
| * |
| * @throws Exception If error occurs. |
| */ |
| @Test |
| public void testSendReceiveMessageWithEnumTopic() throws Exception { |
| final Collection<Object> rcvMsgs = new GridConcurrentHashSet<>(); |
| |
| final AtomicBoolean error = new AtomicBoolean(false); //to make it modifiable |
| |
| final CountDownLatch rcvLatch = new CountDownLatch(3); |
| |
| ignite1.message().localListen(TestTopic.TOPIC_1, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=" + TestTopic.TOPIC_1 + ']'); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_1.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: " + TestTopic.TOPIC_1); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ignite1.message().localListen(TestTopic.TOPIC_2, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=" + TestTopic.TOPIC_2 + ']'); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_2.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: " + TestTopic.TOPIC_2); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ignite1.message().localListen(null, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=default]"); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_3.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: default"); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ClusterGroup rNode1 = ignite1.cluster().forLocal(); |
| |
| message(rNode1).send(TestTopic.TOPIC_1, MSG_1); |
| message(rNode1).send(TestTopic.TOPIC_2, MSG_2); |
| message(rNode1).send(null, MSG_3); |
| |
| assertTrue(rcvLatch.await(3, TimeUnit.SECONDS)); |
| |
| assertFalse(error.get()); |
| |
| assertTrue(rcvMsgs.contains(MSG_1)); |
| assertTrue(rcvMsgs.contains(MSG_2)); |
| assertTrue(rcvMsgs.contains(MSG_3)); |
| } |
| |
| /** |
| * Tests simple message sending-receiving with the use of |
| * remoteListen() method. |
| * |
| * @throws Exception If error occurs. |
| */ |
| @Test |
| public void testRemoteListen() throws Exception { |
| final Collection<Object> rcvMsgs = new GridConcurrentHashSet<>(); |
| |
| rcvLatch = new CountDownLatch(4); |
| |
| ignite2.message().remoteListen(null, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ClusterGroup prj2 = ignite1.cluster().forRemotes(); // Includes node from grid2. |
| |
| message(prj2).send(null, MSG_1); |
| message(prj2).send(null, MSG_2); |
| message(ignite2.cluster().forLocal()).send(null, MSG_3); |
| |
| assertFalse(rcvLatch.await(3, TimeUnit.SECONDS)); // We should get only 3 message. |
| |
| assertTrue(rcvMsgs.contains(MSG_1)); |
| assertTrue(rcvMsgs.contains(MSG_2)); |
| assertTrue(rcvMsgs.contains(MSG_3)); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testStopRemoteListen() throws Exception { |
| final AtomicInteger msgCnt1 = new AtomicInteger(); |
| |
| final AtomicInteger msgCnt2 = new AtomicInteger(); |
| |
| final AtomicInteger msgCnt3 = new AtomicInteger(); |
| |
| final String topic1 = null; |
| final String topic2 = "top2"; |
| final String topic3 = "top3"; |
| |
| UUID id1 = ignite2.message().remoteListen(topic1, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| System.out.println( |
| Thread.currentThread().getName() + " Listener1 received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']' |
| ); |
| |
| msgCnt1.incrementAndGet(); |
| |
| return true; |
| } |
| }); |
| |
| UUID id2 = ignite2.message().remoteListen(topic2, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| System.out.println( |
| Thread.currentThread().getName() + " Listener2 received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']' |
| ); |
| |
| msgCnt2.incrementAndGet(); |
| |
| return true; |
| } |
| }); |
| |
| UUID id3 = ignite2.message().remoteListen(topic3, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| System.out.println( |
| Thread.currentThread().getName() + " Listener3 received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']' |
| ); |
| |
| msgCnt3.incrementAndGet(); |
| |
| return true; |
| } |
| }); |
| |
| message(ignite1.cluster().forRemotes()).send(topic1, "msg1-1"); |
| message(ignite1.cluster().forRemotes()).send(topic2, "msg1-2"); |
| message(ignite1.cluster().forRemotes()).send(topic3, "msg1-3"); |
| |
| GridTestUtils.waitForCondition(new PA() { |
| @Override public boolean apply() { |
| return msgCnt1.get() > 0 && msgCnt2.get() > 0 && msgCnt3.get() > 0; |
| } |
| }, 5000); |
| |
| assertEquals(1, msgCnt1.get()); |
| assertEquals(1, msgCnt2.get()); |
| assertEquals(1, msgCnt3.get()); |
| |
| ignite2.message().stopRemoteListen(id2); |
| |
| message(ignite1.cluster().forRemotes()).send(topic1, "msg2-1"); |
| message(ignite1.cluster().forRemotes()).send(topic2, "msg2-2"); |
| message(ignite1.cluster().forRemotes()).send(topic3, "msg2-3"); |
| |
| GridTestUtils.waitForCondition(new PA() { |
| @Override public boolean apply() { |
| return msgCnt1.get() > 1 && msgCnt3.get() > 1; |
| } |
| }, 5000); |
| |
| assertEquals(2, msgCnt1.get()); |
| assertEquals(1, msgCnt2.get()); |
| assertEquals(2, msgCnt3.get()); |
| |
| ignite2.message().stopRemoteListen(id2); // Try remove one more time. |
| |
| ignite2.message().stopRemoteListen(id1); |
| ignite2.message().stopRemoteListen(id3); |
| |
| message(ignite1.cluster().forRemotes()).send(topic1, "msg3-1"); |
| message(ignite1.cluster().forRemotes()).send(topic2, "msg3-2"); |
| message(ignite1.cluster().forRemotes()).send(topic3, "msg3-3"); |
| |
| U.sleep(1000); |
| |
| assertEquals(2, msgCnt1.get()); |
| assertEquals(1, msgCnt2.get()); |
| assertEquals(2, msgCnt3.get()); |
| } |
| |
| /** |
| * Tests simple message sending-receiving with the use of |
| * remoteListen() method. |
| * |
| * @throws Exception If error occurs. |
| */ |
| @Test |
| public void testRemoteListenOrderedMessages() throws Exception { |
| List<TestMessage> msgs = Arrays.asList( |
| new TestMessage(MSG_1), |
| new TestMessage(MSG_2, 3000), |
| new TestMessage(MSG_3)); |
| |
| final Collection<Object> rcvMsgs = new ConcurrentLinkedDeque<>(); |
| |
| final AtomicBoolean error = new AtomicBoolean(false); //to make it modifiable |
| |
| rcvLatch = new CountDownLatch(3); |
| |
| ignite2.message().remoteListen(S_TOPIC_1, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ClusterGroup prj2 = ignite1.cluster().forRemotes(); // Includes node from grid2. |
| |
| for (TestMessage msg : msgs) |
| message(prj2).sendOrdered(S_TOPIC_1, msg, 15000); |
| |
| assertTrue(rcvLatch.await(6, TimeUnit.SECONDS)); |
| |
| assertFalse(error.get()); |
| |
| assertEquals(msgs, Arrays.asList(rcvMsgs.toArray())); |
| } |
| |
| /** |
| * Tests simple message sending-receiving with the use of |
| * remoteListen() method and topics. |
| * |
| * @throws Exception If error occurs. |
| */ |
| @Test |
| public void testRemoteListenWithIntTopic() throws Exception { |
| final Collection<Object> rcvMsgs = new GridConcurrentHashSet<>(); |
| |
| final AtomicBoolean error = new AtomicBoolean(false); //to make it modifiable |
| |
| rcvLatch = new CountDownLatch(3); |
| |
| ignite2.message().remoteListen(I_TOPIC_1, new P2<UUID, Object>() { |
| @IgniteInstanceResource |
| private transient Ignite g; |
| |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| assertEquals(ignite2, g); |
| |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=" + I_TOPIC_1 + ']'); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_1.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: " + I_TOPIC_1); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ignite2.message().remoteListen(I_TOPIC_2, new P2<UUID, Object>() { |
| @IgniteInstanceResource |
| private transient Ignite g; |
| |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| assertEquals(ignite2, g); |
| |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=" + I_TOPIC_2 + ']'); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_2.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: " + I_TOPIC_2); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ignite2.message().remoteListen(null, new P2<UUID, Object>() { |
| @IgniteInstanceResource |
| private transient Ignite g; |
| |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| assertEquals(ignite2, g); |
| |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + |
| ", topic=default]"); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| if (!MSG_3.equals(msg)) { |
| log.error("Unexpected message " + msg + " for topic: default"); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| rcvMsgs.add(msg); |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| ClusterGroup prj2 = ignite1.cluster().forRemotes(); // Includes node from grid2. |
| |
| message(prj2).send(I_TOPIC_1, MSG_1); |
| message(prj2).send(I_TOPIC_2, MSG_2); |
| message(prj2).send(null, MSG_3); |
| |
| assertTrue(rcvLatch.await(3, TimeUnit.SECONDS)); |
| |
| assertFalse(error.get()); |
| |
| assertTrue(rcvMsgs.contains(MSG_1)); |
| assertTrue(rcvMsgs.contains(MSG_2)); |
| assertTrue(rcvMsgs.contains(MSG_3)); |
| } |
| |
| /** |
| * Checks, if it is OK to send the message, loaded with external |
| * class loader. |
| * |
| * @throws Exception If error occurs. |
| */ |
| @Test |
| public void testSendMessageWithExternalClassLoader() throws Exception { |
| URL[] urls = new URL[] {new URL(GridTestProperties.getProperty("p2p.uri.cls"))}; |
| |
| ClassLoader extLdr = new URLClassLoader(urls); |
| |
| Class rcCls = extLdr.loadClass(EXT_RESOURCE_CLS_NAME); |
| |
| final AtomicBoolean error = new AtomicBoolean(false); //to make it modifiable |
| |
| final CountDownLatch rcvLatch = new CountDownLatch(1); |
| |
| ignite2.message().remoteListen(S_TOPIC_1, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| try { |
| log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| if (!nodeId.equals(ignite1.cluster().localNode().id())) { |
| log.error("Unexpected sender node: " + nodeId); |
| |
| error.set(true); |
| |
| return false; |
| } |
| |
| return true; |
| } |
| finally { |
| rcvLatch.countDown(); |
| } |
| } |
| }); |
| |
| message(ignite1.cluster().forRemotes()).send(S_TOPIC_1, Collections.singleton(rcCls.newInstance())); |
| |
| assertTrue(rcvLatch.await(3, TimeUnit.SECONDS)); |
| |
| assertFalse(error.get()); |
| } |
| |
| /** |
| * Test case for {@code null} messages. |
| * |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testNullMessages() throws Exception { |
| assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| ignite1.message().send(null, null); |
| |
| return null; |
| } |
| }, IllegalArgumentException.class, "Ouch! Argument is invalid: msgs cannot be null or empty"); |
| |
| assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| ignite1.message().send(null, Collections.emptyList()); |
| |
| return null; |
| } |
| }, IllegalArgumentException.class, "Ouch! Argument is invalid: msgs cannot be null or empty"); |
| |
| assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| ignite1.message().send(null, (Object)null); |
| |
| return null; |
| } |
| }, NullPointerException.class, "Ouch! Argument cannot be null: msg"); |
| |
| assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| ignite1.message().send(null, Arrays.asList(null, new Object())); |
| |
| return null; |
| } |
| }, NullPointerException.class, "Ouch! Argument cannot be null: msg"); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testAsyncOld() throws Exception { |
| final AtomicInteger msgCnt = new AtomicInteger(); |
| |
| IgniteDiscoverySpi discoSpi = (IgniteDiscoverySpi)ignite2.configuration().getDiscoverySpi(); |
| |
| DiscoverySpiTestListener lsnr = new DiscoverySpiTestListener(); |
| |
| discoSpi.setInternalListener(lsnr); |
| |
| assertFalse(ignite2.message().isAsync()); |
| |
| final IgniteMessaging msg = ignite2.message().withAsync(); |
| |
| assertTrue(msg.isAsync()); |
| |
| assertFalse(ignite2.message().isAsync()); |
| |
| GridTestUtils.assertThrows(log, new Callable<Void>() { |
| @Override public Void call() throws Exception { |
| msg.future(); |
| |
| return null; |
| } |
| }, IllegalStateException.class, null); |
| |
| lsnr.blockCustomEvent(StartRoutineDiscoveryMessage.class, StartRoutineDiscoveryMessageV2.class); |
| |
| final String topic = "topic"; |
| |
| UUID id = msg.remoteListen(topic, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| System.out.println(Thread.currentThread().getName() + |
| " Listener received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| msgCnt.incrementAndGet(); |
| |
| return true; |
| } |
| }); |
| |
| Assert.assertNull(id); |
| |
| IgniteFuture<UUID> starFut = msg.future(); |
| |
| Assert.assertNotNull(starFut); |
| |
| U.sleep(500); |
| |
| Assert.assertFalse(starFut.isDone()); |
| |
| lsnr.stopBlockCustomEvents(); |
| |
| GridTestUtils.assertThrows(log, new Callable<Void>() { |
| @Override public Void call() throws Exception { |
| msg.future(); |
| |
| return null; |
| } |
| }, IllegalStateException.class, null); |
| |
| id = starFut.get(); |
| |
| Assert.assertNotNull(id); |
| |
| Assert.assertTrue(starFut.isDone()); |
| |
| lsnr.blockCustomEvent(StopRoutineDiscoveryMessage.class); |
| |
| message(ignite1.cluster().forRemotes()).send(topic, "msg1"); |
| |
| GridTestUtils.waitForCondition(new PA() { |
| @Override public boolean apply() { |
| return msgCnt.get() > 0; |
| } |
| }, 5000); |
| |
| assertEquals(1, msgCnt.get()); |
| |
| msg.stopRemoteListen(id); |
| |
| IgniteFuture<?> stopFut = msg.future(); |
| |
| Assert.assertNotNull(stopFut); |
| |
| GridTestUtils.assertThrows(log, new Callable<Void>() { |
| @Override public Void call() throws Exception { |
| msg.future(); |
| |
| return null; |
| } |
| }, IllegalStateException.class, null); |
| |
| U.sleep(500); |
| |
| Assert.assertFalse(stopFut.isDone()); |
| |
| lsnr.stopBlockCustomEvents(); |
| |
| stopFut.get(); |
| |
| Assert.assertTrue(stopFut.isDone()); |
| |
| message(ignite1.cluster().forRemotes()).send(topic, "msg2"); |
| |
| U.sleep(1000); |
| |
| assertEquals(1, msgCnt.get()); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testAsync() throws Exception { |
| final AtomicInteger msgCnt = new AtomicInteger(); |
| |
| IgniteDiscoverySpi discoSpi = (IgniteDiscoverySpi)ignite2.configuration().getDiscoverySpi(); |
| |
| DiscoverySpiTestListener lsnr = new DiscoverySpiTestListener(); |
| |
| discoSpi.setInternalListener(lsnr); |
| |
| lsnr.blockCustomEvent(StartRoutineDiscoveryMessage.class, StartRoutineDiscoveryMessageV2.class); |
| |
| final String topic = "topic"; |
| |
| IgniteFuture<UUID> starFut = ignite2.message().remoteListenAsync(topic, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| System.out.println(Thread.currentThread().getName() + |
| " Listener received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| msgCnt.incrementAndGet(); |
| |
| return true; |
| } |
| }); |
| |
| Assert.assertNotNull(starFut); |
| |
| U.sleep(500); |
| |
| Assert.assertFalse(starFut.isDone()); |
| |
| lsnr.stopBlockCustomEvents(); |
| |
| UUID id = starFut.get(); |
| |
| Assert.assertNotNull(id); |
| |
| Assert.assertTrue(starFut.isDone()); |
| |
| lsnr.blockCustomEvent(StopRoutineDiscoveryMessage.class); |
| |
| message(ignite1.cluster().forRemotes()).send(topic, "msg1"); |
| |
| GridTestUtils.waitForCondition(new PA() { |
| @Override public boolean apply() { |
| return msgCnt.get() > 0; |
| } |
| }, 5000); |
| |
| assertEquals(1, msgCnt.get()); |
| |
| IgniteFuture<?> stopFut = ignite2.message().stopRemoteListenAsync(id); |
| |
| Assert.assertNotNull(stopFut); |
| |
| U.sleep(500); |
| |
| Assert.assertFalse(stopFut.isDone()); |
| |
| lsnr.stopBlockCustomEvents(); |
| |
| stopFut.get(); |
| |
| Assert.assertTrue(stopFut.isDone()); |
| |
| message(ignite1.cluster().forRemotes()).send(topic, "msg2"); |
| |
| U.sleep(1000); |
| |
| assertEquals(1, msgCnt.get()); |
| } |
| |
| /** |
| * Tests that message listener registers only for one oldest node. |
| * |
| * @throws Exception If an error occurred. |
| */ |
| @Test |
| public void testRemoteListenForOldest() throws Exception { |
| remoteListenForOldest(ignite1); |
| |
| // Restart oldest node. |
| stopGrid(1); |
| |
| ignite1 = startGrid(1); |
| |
| MSG_CNT.set(0); |
| |
| // Ignite2 is oldest now. |
| remoteListenForOldest(ignite2); |
| } |
| |
| /** |
| * @param expOldestIgnite Expected oldest ignite. |
| * @throws InterruptedException If failed. |
| */ |
| private void remoteListenForOldest(Ignite expOldestIgnite) throws InterruptedException { |
| ClusterGroup grp = ignite1.cluster().forOldest(); |
| |
| assertEquals(1, grp.nodes().size()); |
| assertEquals(expOldestIgnite.cluster().localNode().id(), grp.node().id()); |
| |
| ignite1.message(grp).remoteListen(null, new P2<UUID, Object>() { |
| @Override public boolean apply(UUID nodeId, Object msg) { |
| System.out.println("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); |
| |
| MSG_CNT.incrementAndGet(); |
| |
| return true; |
| } |
| }); |
| |
| ignite1.message().send(null, MSG_1); |
| |
| Thread.sleep(3000); |
| |
| assertEquals(1, MSG_CNT.get()); |
| } |
| } |
