View Javadoc
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.filter.executor;
21  
22  import org.apache.mina.core.filterchain.IoFilterAdapter;
23  import org.apache.mina.core.future.IoFutureListener;
24  import org.apache.mina.core.future.WriteFuture;
25  import org.apache.mina.core.service.IoProcessor;
26  import org.apache.mina.core.session.IoEvent;
27  import org.apache.mina.core.session.IoEventType;
28  import org.apache.mina.core.session.IoSession;
29  import org.apache.mina.core.write.WriteRequest;
30  
31  /**
32   * Attaches an {@link IoEventQueueHandler} to an {@link IoSession}'s
33   * {@link WriteRequest} queue to provide accurate write queue status tracking.
34   * <p>
35   * The biggest difference from {@link OrderedThreadPoolExecutor} and
36   * {@link UnorderedThreadPoolExecutor} is that {@link IoEventQueueHandler#polled(Object, IoEvent)}
37   * is invoked when the write operation is completed by an {@link IoProcessor},
38   * consequently providing the accurate tracking of the write request queue
39   * status to the {@link IoEventQueueHandler}.
40   * <p>
41   * Most common usage of this filter could be detecting an {@link IoSession}
42   * which writes too fast which will cause {@link OutOfMemoryError} soon:
43   * <pre>
44   *     session.getFilterChain().addLast(
45   *             "writeThrottle",
46   *             new WriteRequestFilter(new IoEventQueueThrottle()));
47   * </pre>
48   *
49   * <h3>Known issues</h3>
50   *
51   * You can run into a dead lock if you run this filter with the blocking
52   * {@link IoEventQueueHandler} implementation such as {@link IoEventQueueThrottle}
53   * in the {@link IoProcessor} thread.  It's because an {@link IoProcessor}
54   * thread is what processes the {@link WriteRequest}s and notifies related
55   * {@link WriteFuture}s; the {@link IoEventQueueHandler} implementation that
56   * waits for the size of the write request queue to decrease will never wake
57   * up.  To use such an handler, you have to insert an {@link ExecutorFilter}
58   * before this filter or call {@link IoSession#write(Object)} method always
59   * from a different thread.
60   *
61   * @author <a href="http://mina.apache.org">Apache MINA Project</a>
62   */
63  public class WriteRequestFilter extends IoFilterAdapter {
64  
65      private final IoEventQueueHandler queueHandler;
66  
67      /**
68       * Creates a new instance with a new default {@link IoEventQueueThrottle}.
69       */
70      public WriteRequestFilter() {
71          this(new IoEventQueueThrottle());
72      }
73  
74      /**
75       * Creates a new instance with the specified {@link IoEventQueueHandler}.
76       * 
77       * @param queueHandler The {@link IoEventQueueHandler} instance to use
78       */
79      public WriteRequestFilter(IoEventQueueHandler queueHandler) {
80          if (queueHandler == null) {
81              throw new IllegalArgumentException("queueHandler");
82          }
83          this.queueHandler = queueHandler;
84      }
85  
86      /**
87       * @return the {@link IoEventQueueHandler} which is attached to this
88       * filter.
89       */
90      public IoEventQueueHandler getQueueHandler() {
91          return queueHandler;
92      }
93  
94      /**
95       * {@inheritDoc}
96       */
97      @Override
98      public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
99  
100         final IoEvent e = new IoEvent(IoEventType.WRITE, session, writeRequest);
101 
102         if (queueHandler.accept(this, e)) {
103             nextFilter.filterWrite(session, writeRequest);
104             WriteFuture writeFuture = writeRequest.getFuture();
105             if (writeFuture == null) {
106                 return;
107             }
108 
109             // We can track the write request only when it has a future.
110             queueHandler.offered(this, e);
111             writeFuture.addListener(new IoFutureListener<WriteFuture>() {
112                 /**
113                  * @inheritedDoc
114                  */
115                 @Override
116                 public void operationComplete(WriteFuture future) {
117                     queueHandler.polled(WriteRequestFilter.this, e);
118                 }
119             });
120         }
121     }
122 }