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.jgroups.tests;
import org.jgroups.Address;
import org.jgroups.Global;
import org.jgroups.util.ResponseCollector;
import org.jgroups.util.Util;
import org.testng.annotations.Test;
import java.util.Map;
/**
* @author Bela Ban
*/
@Test(groups=Global.FUNCTIONAL,sequential=false)
public class ResponseCollectorTest {
static final Address a=Util.createRandomAddress(), b=Util.createRandomAddress(), c=Util.createRandomAddress();
public static void testAdd() {
ResponseCollector<Integer> coll=new ResponseCollector<Integer>(a, b, c);
coll.add(a, 1);
System.out.println("coll = " + coll);
assert coll.size() == 3;
assert !coll.hasAllResponses();
coll.add(c, 3);
coll.add(b, 2);
System.out.println("coll = " + coll);
assert coll.size() == 3;
assert coll.hasAllResponses();
}
public static void testAddNonExistentKeys() {
ResponseCollector<Integer> coll=new ResponseCollector<Integer>(a, b);
coll.add(a, 1);
System.out.println("coll = " + coll);
assert coll.size() == 2;
assert !coll.hasAllResponses();
coll.add(c, 3); // will get dropped
coll.add(b, 2);
System.out.println("coll = " + coll);
assert coll.size() == 2;
assert coll.hasAllResponses();
}
public static void testWaitForAllResponses() {
final ResponseCollector<Integer> coll=new ResponseCollector<Integer>(a, b, c);
boolean rc=coll.waitForAllResponses(500);
assert !rc;
new Thread() {
public void run() {
coll.add(a, 1);
Util.sleep(500);
coll.add(b, 2);
coll.add(c, 3);
}
}.start();
rc=coll.waitForAllResponses(5000);
System.out.println("coll = " + coll);
assert rc;
assert coll.hasAllResponses();
}
public static void testWaitForAllResponsesAndTimeout() {
final ResponseCollector<Integer> coll=new ResponseCollector<Integer>(a, b, c);
new Thread() {
public void run() {
coll.add(a, 1);
Util.sleep(1000);
coll.add(b, 2);
Util.sleep(1000);
coll.add(c, 3);
}
}.start();
boolean rc=coll.waitForAllResponses(400);
System.out.println("coll = " + coll);
assert !rc;
assert !coll.hasAllResponses() : "collector had all responses (not expected)";
}
public static void testWaitForAllResponsesAndReset() {
final ResponseCollector<Integer> coll=new ResponseCollector<Integer>(a, b, c);
new Thread() {
public void run() {
Util.sleep(1000);
coll.add(a, 1);
coll.reset();
}
}.start();
boolean rc=coll.waitForAllResponses(5000);
System.out.println("coll = " + coll);
assert rc;
assert coll.hasAllResponses();
}
public static void testWaitForAllResponsesAndGetResults() throws InterruptedException {
final ResponseCollector<Integer> coll=new ResponseCollector<Integer>(a, b, c);
coll.add(a, 1); coll.add(b, 2); coll.add(c, 3);
Map<Address, Integer> results=coll.getResults();
System.out.println("results = " + results);
Thread thread=new Thread() {
public void run() {
coll.reset();
}
}; thread.start();
thread.join();
System.out.println("results = " + results);
assert coll.size() == 0;
}
}
|
