1 /*
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing,
13 * software distributed under the License is distributed on an
14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 * KIND, either express or implied. See the License for the
16 * specific language governing permissions and limitations
17 * under the License.
18 *
19 */
20 package org.apache.mina.common.support;
21
22 import java.util.ArrayList;
23 import java.util.HashMap;
24 import java.util.Iterator;
25 import java.util.List;
26 import java.util.Map;
27
28 import org.apache.mina.common.ConnectFuture;
29 import org.apache.mina.common.IdleStatus;
30 import org.apache.mina.common.IoFilter;
31 import org.apache.mina.common.IoFilterAdapter;
32 import org.apache.mina.common.IoFilterChain;
33 import org.apache.mina.common.IoFilterLifeCycleException;
34 import org.apache.mina.common.IoSession;
35 import org.apache.mina.common.IoFilter.NextFilter;
36 import org.apache.mina.common.IoFilter.WriteRequest;
37 import org.apache.mina.util.ByteBufferUtil;
38 import org.apache.mina.util.SessionLog;
39
40 /**
41 * An abstract implementation of {@link IoFilterChain} that provides
42 * common operations for developers to implement their own transport layer.
43 * <p>
44 * The only method a developer should implement is
45 * {@link #doWrite(IoSession, IoFilter.WriteRequest)}. This method is invoked
46 * when filter chain is evaluated for
47 * {@link IoFilter#filterWrite(NextFilter, IoSession, IoFilter.WriteRequest)} and
48 * finally to be written out.
49 *
50 * @author The Apache Directory Project (mina-dev@directory.apache.org)
51 * @version $Rev: 561238 $, $Date: 2007-07-31 15:12:43 +0900 (Tue, 31 Jul 2007) $
52 */
53 public abstract class AbstractIoFilterChain implements IoFilterChain {
54 /**
55 * A session attribute that stores a {@link ConnectFuture} related with
56 * the {@link IoSession}. {@link AbstractIoFilterChain} clears this
57 * attribute and notifies the future when {@link #fireSessionOpened(IoSession)}
58 * or {@link #fireExceptionCaught(IoSession, Throwable)} is invoked
59 */
60 public static final String CONNECT_FUTURE = AbstractIoFilterChain.class
61 .getName()
62 + ".connectFuture";
63
64 private final IoSession session;
65
66 private final Map<String, Entry> name2entry = new HashMap<String, Entry>();
67
68 private final EntryImpl head;
69
70 private final EntryImpl tail;
71
72 protected AbstractIoFilterChain(IoSession session) {
73 if (session == null) {
74 throw new NullPointerException("session");
75 }
76
77 this.session = session;
78 head = new EntryImpl(null, null, "head", new HeadFilter());
79 tail = new EntryImpl(head, null, "tail", new TailFilter());
80 head.nextEntry = tail;
81 }
82
83 public IoSession getSession() {
84 return session;
85 }
86
87 public Entry getEntry(String name) {
88 Entry e = name2entry.get(name);
89 if (e == null) {
90 return null;
91 }
92 return e;
93 }
94
95 public IoFilter get(String name) {
96 Entry e = getEntry(name);
97 if (e == null) {
98 return null;
99 }
100
101 return e.getFilter();
102 }
103
104 public NextFilter getNextFilter(String name) {
105 Entry e = getEntry(name);
106 if (e == null) {
107 return null;
108 }
109
110 return e.getNextFilter();
111 }
112
113 public synchronized void addFirst(String name, IoFilter filter) {
114 checkAddable(name);
115 register(head, name, filter);
116 }
117
118 public synchronized void addLast(String name, IoFilter filter) {
119 checkAddable(name);
120 register(tail.prevEntry, name, filter);
121 }
122
123 public synchronized void addBefore(String baseName, String name,
124 IoFilter filter) {
125 EntryImpl baseEntry = checkOldName(baseName);
126 checkAddable(name);
127 register(baseEntry.prevEntry, name, filter);
128 }
129
130 public synchronized void addAfter(String baseName, String name,
131 IoFilter filter) {
132 EntryImpl baseEntry = checkOldName(baseName);
133 checkAddable(name);
134 register(baseEntry, name, filter);
135 }
136
137 public synchronized IoFilter remove(String name) {
138 EntryImpl entry = checkOldName(name);
139 deregister(entry);
140 return entry.getFilter();
141 }
142
143 public synchronized void clear() throws Exception {
144 Iterator<String> it = new ArrayList<String>(name2entry.keySet())
145 .iterator();
146 while (it.hasNext()) {
147 this.remove(it.next());
148 }
149 }
150
151 private void register(EntryImpl prevEntry, String name, IoFilter filter) {
152 EntryImpl newEntry = new EntryImpl(prevEntry, prevEntry.nextEntry,
153 name, filter);
154
155 try {
156 filter.onPreAdd(this, name, newEntry.getNextFilter());
157 } catch (Exception e) {
158 throw new IoFilterLifeCycleException("onPreAdd(): " + name + ':'
159 + filter + " in " + getSession(), e);
160 }
161
162 prevEntry.nextEntry.prevEntry = newEntry;
163 prevEntry.nextEntry = newEntry;
164 name2entry.put(name, newEntry);
165
166 try {
167 filter.onPostAdd(this, name, newEntry.getNextFilter());
168 } catch (Exception e) {
169 deregister0(newEntry);
170 throw new IoFilterLifeCycleException("onPostAdd(): " + name + ':'
171 + filter + " in " + getSession(), e);
172 }
173 }
174
175 private void deregister(EntryImpl entry) {
176 IoFilter filter = entry.getFilter();
177
178 try {
179 filter.onPreRemove(this, entry.getName(), entry.getNextFilter());
180 } catch (Exception e) {
181 throw new IoFilterLifeCycleException("onPreRemove(): "
182 + entry.getName() + ':' + filter + " in " + getSession(), e);
183 }
184
185 deregister0(entry);
186
187 try {
188 filter.onPostRemove(this, entry.getName(), entry.getNextFilter());
189 } catch (Exception e) {
190 throw new IoFilterLifeCycleException("onPostRemove(): "
191 + entry.getName() + ':' + filter + " in " + getSession(), e);
192 }
193 }
194
195 private void deregister0(EntryImpl entry) {
196 EntryImpl prevEntry = entry.prevEntry;
197 EntryImpl nextEntry = entry.nextEntry;
198 prevEntry.nextEntry = nextEntry;
199 nextEntry.prevEntry = prevEntry;
200
201 name2entry.remove(entry.name);
202 }
203
204 /**
205 * Throws an exception when the specified filter name is not registered in this chain.
206 *
207 * @return An filter entry with the specified name.
208 */
209 private EntryImpl checkOldName(String baseName) {
210 EntryImpl e = (EntryImpl) name2entry.get(baseName);
211 if (e == null) {
212 throw new IllegalArgumentException("Unknown filter name:"
213 + baseName);
214 }
215 return e;
216 }
217
218 /**
219 * Checks the specified filter name is already taken and throws an exception if already taken.
220 */
221 private void checkAddable(String name) {
222 if (name2entry.containsKey(name)) {
223 throw new IllegalArgumentException(
224 "Other filter is using the same name '" + name + "'");
225 }
226 }
227
228 public void fireSessionCreated(IoSession session) {
229 Entry head = this.head;
230 callNextSessionCreated(head, session);
231 }
232
233 private void callNextSessionCreated(Entry entry, IoSession session) {
234 try {
235 entry.getFilter().sessionCreated(entry.getNextFilter(), session);
236 } catch (Throwable e) {
237 fireExceptionCaught(session, e);
238 }
239 }
240
241 public void fireSessionOpened(IoSession session) {
242 Entry head = this.head;
243 callNextSessionOpened(head, session);
244 }
245
246 private void callNextSessionOpened(Entry entry, IoSession session) {
247 try {
248 entry.getFilter().sessionOpened(entry.getNextFilter(), session);
249 } catch (Throwable e) {
250 fireExceptionCaught(session, e);
251 }
252 }
253
254 public void fireSessionClosed(IoSession session) {
255 // Update future.
256 try {
257 session.getCloseFuture().setClosed();
258 } catch (Throwable t) {
259 fireExceptionCaught(session, t);
260 }
261
262 // And start the chain.
263 Entry head = this.head;
264 callNextSessionClosed(head, session);
265 }
266
267 private void callNextSessionClosed(Entry entry, IoSession session) {
268 try {
269 entry.getFilter().sessionClosed(entry.getNextFilter(), session);
270
271 } catch (Throwable e) {
272 fireExceptionCaught(session, e);
273 }
274 }
275
276 public void fireSessionIdle(IoSession session, IdleStatus status) {
277 Entry head = this.head;
278 callNextSessionIdle(head, session, status);
279 }
280
281 private void callNextSessionIdle(Entry entry, IoSession session,
282 IdleStatus status) {
283 try {
284 entry.getFilter().sessionIdle(entry.getNextFilter(), session,
285 status);
286 } catch (Throwable e) {
287 fireExceptionCaught(session, e);
288 }
289 }
290
291 public void fireMessageReceived(IoSession session, Object message) {
292 Entry head = this.head;
293 callNextMessageReceived(head, session, message);
294 }
295
296 private void callNextMessageReceived(Entry entry, IoSession session,
297 Object message) {
298 try {
299 entry.getFilter().messageReceived(entry.getNextFilter(), session,
300 message);
301 } catch (Throwable e) {
302 fireExceptionCaught(session, e);
303 }
304 }
305
306 public void fireMessageSent(IoSession session, WriteRequest request) {
307 try {
308 request.getFuture().setWritten(true);
309 } catch (Throwable t) {
310 fireExceptionCaught(session, t);
311 }
312
313 Entry head = this.head;
314 callNextMessageSent(head, session, request.getMessage());
315 }
316
317 private void callNextMessageSent(Entry entry, IoSession session,
318 Object message) {
319 try {
320 entry.getFilter().messageSent(entry.getNextFilter(), session,
321 message);
322 } catch (Throwable e) {
323 fireExceptionCaught(session, e);
324 }
325 }
326
327 public void fireExceptionCaught(IoSession session, Throwable cause) {
328 // Notify the related ConnectFuture
329 // if the session is created from SocketConnector.
330 ConnectFuture future = (ConnectFuture) session
331 .removeAttribute(CONNECT_FUTURE);
332 if (future == null) {
333 Entry head = this.head;
334 callNextExceptionCaught(head, session, cause);
335 } else {
336 // Please note that this place is not the only place that
337 // calls ConnectFuture.setException().
338 future.setException(cause);
339 }
340 }
341
342 private void callNextExceptionCaught(Entry entry, IoSession session,
343 Throwable cause) {
344 try {
345 entry.getFilter().exceptionCaught(entry.getNextFilter(), session,
346 cause);
347 } catch (Throwable e) {
348 SessionLog.warn(session,
349 "Unexpected exception from exceptionCaught handler.", e);
350 }
351 }
352
353 public void fireFilterWrite(IoSession session, WriteRequest writeRequest) {
354 Entry tail = this.tail;
355 callPreviousFilterWrite(tail, session, writeRequest);
356 }
357
358 private void callPreviousFilterWrite(Entry entry, IoSession session,
359 WriteRequest writeRequest) {
360 try {
361 entry.getFilter().filterWrite(entry.getNextFilter(), session,
362 writeRequest);
363 } catch (Throwable e) {
364 writeRequest.getFuture().setWritten(false);
365 fireExceptionCaught(session, e);
366 }
367 }
368
369 public void fireFilterClose(IoSession session) {
370 Entry tail = this.tail;
371 callPreviousFilterClose(tail, session);
372 }
373
374 private void callPreviousFilterClose(Entry entry, IoSession session) {
375 try {
376 entry.getFilter().filterClose(entry.getNextFilter(), session);
377 } catch (Throwable e) {
378 fireExceptionCaught(session, e);
379 }
380 }
381
382 public List<Entry> getAll() {
383 List<Entry> list = new ArrayList<Entry>();
384 EntryImpl e = head.nextEntry;
385 while (e != tail) {
386 list.add(e);
387 e = e.nextEntry;
388 }
389
390 return list;
391 }
392
393 public List<Entry> getAllReversed() {
394 List<Entry> list = new ArrayList<Entry>();
395 EntryImpl e = tail.prevEntry;
396 while (e != head) {
397 list.add(e);
398 e = e.prevEntry;
399 }
400 return list;
401 }
402
403 public boolean contains(String name) {
404 return getEntry(name) != null;
405 }
406
407 public boolean contains(IoFilter filter) {
408 EntryImpl e = head.nextEntry;
409 while (e != tail) {
410 if (e.getFilter() == filter) {
411 return true;
412 }
413 e = e.nextEntry;
414 }
415 return false;
416 }
417
418 public boolean contains(Class<? extends IoFilter> filterType) {
419 EntryImpl e = head.nextEntry;
420 while (e != tail) {
421 if (filterType.isAssignableFrom(e.getFilter().getClass())) {
422 return true;
423 }
424 e = e.nextEntry;
425 }
426 return false;
427 }
428
429 public String toString() {
430 StringBuffer buf = new StringBuffer();
431 buf.append("{ ");
432
433 boolean empty = true;
434
435 EntryImpl e = head.nextEntry;
436 while (e != tail) {
437 if (!empty) {
438 buf.append(", ");
439 } else {
440 empty = false;
441 }
442
443 buf.append('(');
444 buf.append(e.getName());
445 buf.append(':');
446 buf.append(e.getFilter());
447 buf.append(')');
448
449 e = e.nextEntry;
450 }
451
452 if (empty) {
453 buf.append("empty");
454 }
455
456 buf.append(" }");
457
458 return buf.toString();
459 }
460
461 protected void finalize() throws Throwable {
462 try {
463 this.clear();
464 } finally {
465 super.finalize();
466 }
467 }
468
469 protected abstract void doWrite(IoSession session, WriteRequest writeRequest)
470 throws Exception;
471
472 protected abstract void doClose(IoSession session) throws Exception;
473
474 private class HeadFilter extends IoFilterAdapter {
475 public void sessionCreated(NextFilter nextFilter, IoSession session) {
476 nextFilter.sessionCreated(session);
477 }
478
479 public void sessionOpened(NextFilter nextFilter, IoSession session) {
480 nextFilter.sessionOpened(session);
481 }
482
483 public void sessionClosed(NextFilter nextFilter, IoSession session) {
484 nextFilter.sessionClosed(session);
485 }
486
487 public void sessionIdle(NextFilter nextFilter, IoSession session,
488 IdleStatus status) {
489 nextFilter.sessionIdle(session, status);
490 }
491
492 public void exceptionCaught(NextFilter nextFilter, IoSession session,
493 Throwable cause) {
494 nextFilter.exceptionCaught(session, cause);
495 }
496
497 public void messageReceived(NextFilter nextFilter, IoSession session,
498 Object message) {
499 nextFilter.messageReceived(session, message);
500 }
501
502 public void messageSent(NextFilter nextFilter, IoSession session,
503 Object message) {
504 nextFilter.messageSent(session, message);
505 }
506
507 public void filterWrite(NextFilter nextFilter, IoSession session,
508 WriteRequest writeRequest) throws Exception {
509 if (session.getTransportType().getEnvelopeType().isAssignableFrom(
510 writeRequest.getMessage().getClass())) {
511 doWrite(session, writeRequest);
512 } else {
513 throw new IllegalStateException(
514 "Write requests must be transformed to "
515 + session.getTransportType().getEnvelopeType()
516 + ": " + writeRequest);
517 }
518 }
519
520 public void filterClose(NextFilter nextFilter, IoSession session)
521 throws Exception {
522 doClose(session);
523 }
524 }
525
526 private static class TailFilter extends IoFilterAdapter {
527 public void sessionCreated(NextFilter nextFilter, IoSession session)
528 throws Exception {
529 session.getHandler().sessionCreated(session);
530 }
531
532 public void sessionOpened(NextFilter nextFilter, IoSession session)
533 throws Exception {
534 try {
535 session.getHandler().sessionOpened(session);
536 } finally {
537 // Notify the related ConnectFuture
538 // if the session is created from SocketConnector.
539 ConnectFuture future = (ConnectFuture) session
540 .removeAttribute(CONNECT_FUTURE);
541 if (future != null) {
542 future.setSession(session);
543 }
544 }
545 }
546
547 public void sessionClosed(NextFilter nextFilter, IoSession session)
548 throws Exception {
549 try {
550 session.getHandler().sessionClosed(session);
551 } finally {
552 // Remove all filters.
553 session.getFilterChain().clear();
554 }
555 }
556
557 public void sessionIdle(NextFilter nextFilter, IoSession session,
558 IdleStatus status) throws Exception {
559 session.getHandler().sessionIdle(session, status);
560 }
561
562 public void exceptionCaught(NextFilter nextFilter, IoSession session,
563 Throwable cause) throws Exception {
564 session.getHandler().exceptionCaught(session, cause);
565 }
566
567 public void messageReceived(NextFilter nextFilter, IoSession session,
568 Object message) throws Exception {
569 try {
570 session.getHandler().messageReceived(session, message);
571 } finally {
572 ByteBufferUtil.releaseIfPossible(message);
573 }
574 }
575
576 public void messageSent(NextFilter nextFilter, IoSession session,
577 Object message) throws Exception {
578 try {
579 session.getHandler().messageSent(session, message);
580 } finally {
581 ByteBufferUtil.releaseIfPossible(message);
582 }
583 }
584
585 public void filterWrite(NextFilter nextFilter, IoSession session,
586 WriteRequest writeRequest) throws Exception {
587 nextFilter.filterWrite(session, writeRequest);
588 }
589
590 public void filterClose(NextFilter nextFilter, IoSession session)
591 throws Exception {
592 nextFilter.filterClose(session);
593 }
594 }
595
596 private class EntryImpl implements Entry {
597 private EntryImpl prevEntry;
598
599 private EntryImpl nextEntry;
600
601 private final String name;
602
603 private final IoFilter filter;
604
605 private final NextFilter nextFilter;
606
607 private EntryImpl(EntryImpl prevEntry, EntryImpl nextEntry,
608 String name, IoFilter filter) {
609 if (filter == null) {
610 throw new NullPointerException("filter");
611 }
612 if (name == null) {
613 throw new NullPointerException("name");
614 }
615
616 this.prevEntry = prevEntry;
617 this.nextEntry = nextEntry;
618 this.name = name;
619 this.filter = filter;
620 this.nextFilter = new NextFilter() {
621 public void sessionCreated(IoSession session) {
622 Entry nextEntry = EntryImpl.this.nextEntry;
623 callNextSessionCreated(nextEntry, session);
624 }
625
626 public void sessionOpened(IoSession session) {
627 Entry nextEntry = EntryImpl.this.nextEntry;
628 callNextSessionOpened(nextEntry, session);
629 }
630
631 public void sessionClosed(IoSession session) {
632 Entry nextEntry = EntryImpl.this.nextEntry;
633 callNextSessionClosed(nextEntry, session);
634 }
635
636 public void sessionIdle(IoSession session, IdleStatus status) {
637 Entry nextEntry = EntryImpl.this.nextEntry;
638 callNextSessionIdle(nextEntry, session, status);
639 }
640
641 public void exceptionCaught(IoSession session, Throwable cause) {
642 Entry nextEntry = EntryImpl.this.nextEntry;
643 callNextExceptionCaught(nextEntry, session, cause);
644 }
645
646 public void messageReceived(IoSession session, Object message) {
647 Entry nextEntry = EntryImpl.this.nextEntry;
648 callNextMessageReceived(nextEntry, session, message);
649 }
650
651 public void messageSent(IoSession session, Object message) {
652 Entry nextEntry = EntryImpl.this.nextEntry;
653 callNextMessageSent(nextEntry, session, message);
654 }
655
656 public void filterWrite(IoSession session,
657 WriteRequest writeRequest) {
658 Entry nextEntry = EntryImpl.this.prevEntry;
659 callPreviousFilterWrite(nextEntry, session, writeRequest);
660 }
661
662 public void filterClose(IoSession session) {
663 Entry nextEntry = EntryImpl.this.prevEntry;
664 callPreviousFilterClose(nextEntry, session);
665 }
666 };
667 }
668
669 public String getName() {
670 return name;
671 }
672
673 public IoFilter getFilter() {
674 return filter;
675 }
676
677 public NextFilter getNextFilter() {
678 return nextFilter;
679 }
680
681 public String toString() {
682 return "(" + getName() + ':' + filter + ')';
683 }
684 }
685 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！