1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.mina.filter.executor;
21
22 import static org.junit.Assert.assertEquals;
23
24 import java.lang.reflect.Field;
25 import java.util.concurrent.TimeUnit;
26 import java.util.concurrent.atomic.AtomicInteger;
27
28 import org.apache.mina.core.filterchain.IoFilter;
29 import org.apache.mina.core.session.DummySession;
30 import org.apache.mina.core.session.IdleStatus;
31 import org.apache.mina.core.session.IoSession;
32 import org.apache.mina.core.write.WriteRequest;
33 import org.junit.Test;
34
35
36
37
38
39
40
41 public class UnorderedThreadPoolExecutorTest
42 {
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58 @Test
59 public void testRuntimeExceptionInWorkerRun() throws Throwable
60 {
61
62 int corePoolSize = 1;
63 UnorderedThreadPoolExecutor executor = new UnorderedThreadPoolExecutor(corePoolSize,1);
64 IoFilter.NextFilter nextFilter = new NextFilterAdapter() {
65 @Override
66 public void messageReceived(IoSession session, Object message) {
67 throw new RuntimeException("A RuntimeException thrown during unit testing.");
68 }
69 };
70 DummySession session = new DummySession();
71 ExecutorFilter filter = new ExecutorFilter(executor);
72
73 try {
74
75 filter.messageReceived(nextFilter, session, null);
76
77
78 executor.shutdown();
79 if (!executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS)) {
80 throw new IllegalStateException("Bug in test implementation.");
81 }
82
83
84 final Field idleWorkersField = UnorderedThreadPoolExecutor.class.getDeclaredField("idleWorkers");
85 idleWorkersField.setAccessible(true);
86 final AtomicInteger idleWorkers = (AtomicInteger) idleWorkersField.get(executor);
87 assertEquals("After all tasks have finished, the amount of workers that are idle should equal the amount of workers, but did not.", executor.getPoolSize(), idleWorkers.get());
88 } finally {
89
90 if (!executor.isShutdown()) {
91 executor.shutdownNow();
92 }
93 }
94 }
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111 @Test
112 public void testErrorInWorkerRun() throws Throwable
113 {
114
115 int corePoolSize = 1;
116 UnorderedThreadPoolExecutor executor = new UnorderedThreadPoolExecutor(corePoolSize,1);
117 IoFilter.NextFilter nextFilter = new NextFilterAdapter() {
118 @Override
119 public void messageReceived(IoSession session, Object message) {
120 throw new Error("An Error thrown during unit testing.");
121 }
122 };
123 DummySession session = new DummySession();
124 ExecutorFilter filter = new ExecutorFilter(executor);
125
126 try {
127
128 filter.messageReceived(nextFilter, session, null);
129
130
131 executor.shutdown();
132 if (!executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS)) {
133 throw new IllegalStateException("Bug in test implementation.");
134 }
135
136
137 final Field idleWorkersField = UnorderedThreadPoolExecutor.class.getDeclaredField("idleWorkers");
138 idleWorkersField.setAccessible(true);
139 final AtomicInteger idleWorkers = (AtomicInteger) idleWorkersField.get(executor);
140 assertEquals("After all tasks have finished, the amount of workers that are idle should equal the amount of workers, but did not.", executor.getPoolSize(), idleWorkers.get());
141 } finally {
142
143 if (!executor.isShutdown()) {
144 executor.shutdownNow();
145 }
146 }
147 }
148
149
150
151
152 private abstract static class NextFilterAdapter implements IoFilter.NextFilter {
153 public void sessionOpened(IoSession session) {}
154 public void sessionClosed(IoSession session) {}
155 public void sessionIdle(IoSession session, IdleStatus status) {}
156 public void exceptionCaught(IoSession session, Throwable cause) {}
157 public void inputClosed(IoSession session) {}
158 public void messageReceived(IoSession session, Object message) {}
159 public void messageSent(IoSession session, WriteRequest writeRequest) {}
160 public void filterWrite(IoSession session, WriteRequest writeRequest) {}
161 public void filterClose(IoSession session) {}
162 public void sessionCreated(IoSession session) {}
163 }
164 }