NettyMessagingTest.java
4.75 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
package org.onlab.netty;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.onlab.packet.IpAddress;
import org.onosproject.store.cluster.messaging.Endpoint;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.Uninterruptibles;
import static org.junit.Assert.*;
import static org.onlab.junit.TestTools.findAvailablePort;
/**
* Unit tests for NettyMessaging.
*/
public class NettyMessagingTest {
NettyMessaging netty1;
NettyMessaging netty2;
Endpoint ep1 = new Endpoint(IpAddress.valueOf("127.0.0.1"), 5001);
Endpoint ep2 = new Endpoint(IpAddress.valueOf("127.0.0.1"), 5002);
Endpoint invalidEndPoint = new Endpoint(IpAddress.valueOf("127.0.0.1"), 5003);
@Before
public void setUp() throws Exception {
ep1 = new Endpoint(IpAddress.valueOf("127.0.0.1"), findAvailablePort(5001));
netty1 = new NettyMessaging();
netty1.start(12, ep1);
ep2 = new Endpoint(IpAddress.valueOf("127.0.0.1"), findAvailablePort(5003));
netty2 = new NettyMessaging();
netty2.start(12, ep2);
}
@After
public void tearDown() throws Exception {
if (netty1 != null) {
netty1.stop();
}
if (netty2 != null) {
netty2.stop();
}
}
@Test
public void testSendAsync() {
CountDownLatch latch1 = new CountDownLatch(1);
CompletableFuture<Void> response = netty1.sendAsync(ep2, "test-subject", "hello world".getBytes());
response.whenComplete((r, e) -> {
assertNull(e);
latch1.countDown();
});
Uninterruptibles.awaitUninterruptibly(latch1);
CountDownLatch latch2 = new CountDownLatch(1);
response = netty1.sendAsync(invalidEndPoint, "test-subject", "hello world".getBytes());
response.whenComplete((r, e) -> {
assertNotNull(e);
latch2.countDown();
});
Uninterruptibles.awaitUninterruptibly(latch2);
}
@Test
public void testSendAndReceive() {
AtomicBoolean handlerInvoked = new AtomicBoolean(false);
AtomicReference<byte[]> request = new AtomicReference<>();
AtomicReference<Endpoint> sender = new AtomicReference<>();
BiFunction<Endpoint, byte[], byte[]> handler = (ep, data) -> {
handlerInvoked.set(true);
sender.set(ep);
request.set(data);
return "hello there".getBytes();
};
netty2.registerHandler("test-subject", handler, MoreExecutors.directExecutor());
CompletableFuture<byte[]> response = netty1.sendAndReceive(ep2, "test-subject", "hello world".getBytes());
assertTrue(Arrays.equals("hello there".getBytes(), response.join()));
assertTrue(handlerInvoked.get());
assertTrue(Arrays.equals(request.get(), "hello world".getBytes()));
assertEquals(ep1, sender.get());
}
/*
* Supplies executors when registering a handler and calling sendAndReceive and verifies the request handling
* and response completion occurs on the expected thread.
*/
@Test
public void testSendAndReceiveWithExecutor() {
ExecutorService completionExecutor = Executors.newSingleThreadExecutor(r -> new Thread(r, "completion-thread"));
ExecutorService handlerExecutor = Executors.newSingleThreadExecutor(r -> new Thread(r, "handler-thread"));
AtomicReference<String> handlerThreadName = new AtomicReference<>();
AtomicReference<String> completionThreadName = new AtomicReference<>();
BiFunction<Endpoint, byte[], byte[]> handler = (ep, data) -> {
handlerThreadName.set(Thread.currentThread().getName());
return "hello there".getBytes();
};
netty2.registerHandler("test-subject", handler, handlerExecutor);
CompletableFuture<byte[]> response = netty1.sendAndReceive(ep2,
"test-subject",
"hello world".getBytes(),
completionExecutor);
response.whenComplete((r, e) -> {
completionThreadName.set(Thread.currentThread().getName());
});
// Verify that the message was request handling and response completion happens on the correct thread.
assertTrue(Arrays.equals("hello there".getBytes(), response.join()));
assertEquals("completion-thread", completionThreadName.get());
assertEquals("handler-thread", handlerThreadName.get());
}
}