1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.mina.core.service;
21
22 import java.lang.reflect.Constructor;
23 import java.util.concurrent.Executor;
24 import java.util.concurrent.ExecutorService;
25 import java.util.concurrent.Executors;
26
27 import org.apache.mina.core.RuntimeIoException;
28 import org.apache.mina.core.session.AbstractIoSession;
29 import org.apache.mina.core.session.AttributeKey;
30 import org.apache.mina.core.session.IoSession;
31 import org.slf4j.Logger;
32 import org.slf4j.LoggerFactory;
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 public class SimpleIoProcessorPool<T extends AbstractIoSession> implements IoProcessor<T> {
77
78 private final static Logger LOGGER = LoggerFactory.getLogger(SimpleIoProcessorPool.class);
79
80
81 private static final int DEFAULT_SIZE = Runtime.getRuntime().availableProcessors() + 1;
82
83
84 private static final AttributeKey PROCESSOR = new AttributeKey( SimpleIoProcessorPool.class, "processor");
85
86
87 private final IoProcessor<T>[] pool;
88
89
90 private final Executor executor;
91
92
93 private final boolean createdExecutor;
94
95
96 private final Object disposalLock = new Object();
97
98
99 private volatile boolean disposing;
100
101
102 private volatile boolean disposed;
103
104
105
106
107
108
109
110 public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType) {
111 this(processorType, null, DEFAULT_SIZE);
112 }
113
114
115
116
117
118
119
120
121 public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType, int size) {
122 this(processorType, null, size);
123 }
124
125
126
127
128
129
130
131 public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType, Executor executor) {
132 this(processorType, executor, DEFAULT_SIZE);
133 }
134
135
136
137
138
139
140
141 @SuppressWarnings("unchecked")
142 public SimpleIoProcessorPool(Class<? extends IoProcessor<T>> processorType,
143 Executor executor, int size) {
144 if (processorType == null) {
145 throw new IllegalArgumentException("processorType");
146 }
147
148 if (size <= 0) {
149 throw new IllegalArgumentException("size: " + size
150 + " (expected: positive integer)");
151 }
152
153
154 createdExecutor = (executor == null);
155
156 if (createdExecutor) {
157 this.executor = Executors.newCachedThreadPool();
158 } else {
159 this.executor = executor;
160 }
161
162 pool = new IoProcessor[size];
163
164 boolean success = false;
165 Constructor<? extends IoProcessor<T>> processorConstructor = null;
166 boolean usesExecutorArg = true;
167
168 try {
169
170 try {
171 try {
172 processorConstructor = processorType.getConstructor(ExecutorService.class);
173 pool[0] = processorConstructor.newInstance(this.executor);
174 } catch (NoSuchMethodException e) {
175
176 }
177
178 try {
179 processorConstructor = processorType.getConstructor(Executor.class);
180 pool[0] = processorConstructor.newInstance(this.executor);
181 } catch (NoSuchMethodException e) {
182
183 }
184
185 try {
186 processorConstructor = processorType.getConstructor();
187 usesExecutorArg = false;
188 pool[0] = processorConstructor.newInstance();
189 } catch (NoSuchMethodException e) {
190
191 }
192 } catch (RuntimeException re) {
193 LOGGER.error("Cannot create an IoProcessor :{}", re.getMessage());
194 throw re;
195 } catch (Exception e) {
196 String msg = "Failed to create a new instance of " + processorType.getName() + ":" + e.getMessage();
197 LOGGER.error(msg, e);
198 throw new RuntimeIoException(msg , e);
199 }
200
201 if (processorConstructor == null) {
202
203 String msg = String.valueOf(processorType)
204 + " must have a public constructor with one "
205 + ExecutorService.class.getSimpleName()
206 + " parameter, a public constructor with one "
207 + Executor.class.getSimpleName()
208 + " parameter or a public default constructor.";
209 LOGGER.error(msg);
210 throw new IllegalArgumentException(msg);
211 }
212
213
214 for (int i = 1; i < pool.length; i++) {
215 try {
216 if (usesExecutorArg) {
217 pool[i] = processorConstructor.newInstance(this.executor);
218 } else {
219 pool[i] = processorConstructor.newInstance();
220 }
221 } catch (Exception e) {
222
223 }
224 }
225
226 success = true;
227 } finally {
228 if (!success) {
229 dispose();
230 }
231 }
232 }
233
234
235
236
237 public final void add(T session) {
238 getProcessor(session).add(session);
239 }
240
241
242
243
244 public final void flush(T session) {
245 getProcessor(session).flush(session);
246 }
247
248
249
250
251 public final void remove(T session) {
252 getProcessor(session).remove(session);
253 }
254
255
256
257
258 public final void updateTrafficControl(T session) {
259 getProcessor(session).updateTrafficControl(session);
260 }
261
262
263
264
265 public boolean isDisposed() {
266 return disposed;
267 }
268
269
270
271
272 public boolean isDisposing() {
273 return disposing;
274 }
275
276
277
278
279 public final void dispose() {
280 if (disposed) {
281 return;
282 }
283
284 synchronized (disposalLock) {
285 if (!disposing) {
286 disposing = true;
287
288
289 for (int i = pool.length - 1; i >= 0; i--) {
290 if ((pool[i] == null) || pool[i].isDisposing()) {
291
292 continue;
293 }
294
295 try {
296 pool[i].dispose();
297 } catch (Exception e) {
298 LOGGER.warn("Failed to dispose a "
299 + pool[i].getClass().getSimpleName()
300 + " at index " + i + ".", e);
301 } finally {
302 pool[i] = null;
303 }
304 }
305
306 if (createdExecutor) {
307 ((ExecutorService) executor).shutdown();
308 }
309 }
310 }
311
312 disposed = true;
313 }
314
315
316
317
318
319 @SuppressWarnings("unchecked")
320 private IoProcessor<T> getProcessor(T session) {
321 IoProcessor<T> processor = (IoProcessor<T>) session.getAttribute(PROCESSOR);
322
323 if (processor == null) {
324 processor = nextProcessor(session);
325 session.setAttributeIfAbsent(PROCESSOR, processor);
326 }
327
328 return processor;
329 }
330
331
332
333
334 private IoProcessor<T> nextProcessor(T session) {
335 if (disposed) {
336 throw new IllegalStateException(
337 "A disposed processor cannot be accessed.");
338 }
339
340 return pool[Math.abs((int)session.getId()) % pool.length];
341 }
342 }