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.buffer;
21  
22  import java.io.BufferedOutputStream;
23  import java.util.concurrent.ConcurrentHashMap;
24  
25  import org.apache.mina.core.buffer.IoBuffer;
26  import org.apache.mina.core.filterchain.IoFilter;
27  import org.apache.mina.core.filterchain.IoFilterAdapter;
28  import org.apache.mina.core.session.IoSession;
29  import org.apache.mina.core.write.DefaultWriteRequest;
30  import org.apache.mina.core.write.WriteRequest;
31  import org.apache.mina.filter.codec.ProtocolCodecFilter;
32  import org.apache.mina.util.LazyInitializedCacheMap;
33  import org.slf4j.Logger;
34  import org.slf4j.LoggerFactory;
35  
36  /**
37   * An {@link IoFilter} implementation used to buffer outgoing {@link WriteRequest} almost 
38   * like what {@link BufferedOutputStream} does. Using this filter allows to be less dependent 
39   * from network latency. It is also useful when a session is generating very small messages 
40   * too frequently and consequently generating unnecessary traffic overhead.
41   * 
42   * Please note that it should always be placed before the {@link ProtocolCodecFilter} 
43   * as it only handles {@link WriteRequest}'s carrying {@link IoBuffer} objects.
44   * 
45   * @author <a href="http://mina.apache.org">Apache MINA Project</a>
46   * @since MINA 2.0.0-M2
47   * @org.apache.xbean.XBean
48   */
49  public final class BufferedWriteFilter extends IoFilterAdapter {
50      private final Logger logger = LoggerFactory
51              .getLogger(BufferedWriteFilter.class);
52  
53      /**
54       * Default buffer size value in bytes.
55       */
56      public final static int DEFAULT_BUFFER_SIZE = 8192;
57  
58      /**
59       * The buffer size allocated for each new session's buffer.
60       */
61      private int bufferSize = DEFAULT_BUFFER_SIZE;
62  
63      /**
64       * The map that matches an {@link IoSession} and it's {@link IoBuffer}
65       * buffer.
66       */
67      private final LazyInitializedCacheMap<IoSession, IoBuffer> buffersMap;
68  
69      /**
70       * Default constructor. Sets buffer size to {@link #DEFAULT_BUFFER_SIZE}
71       * bytes. Uses a default instance of {@link ConcurrentHashMap}.
72       */
73      public BufferedWriteFilter() {
74          this(DEFAULT_BUFFER_SIZE, null);
75      }
76  
77      /**
78       * Constructor which sets buffer size to <code>bufferSize</code>.Uses a default 
79       * instance of {@link ConcurrentHashMap}.
80       * 
81       * @param bufferSize the new buffer size
82       */
83      public BufferedWriteFilter(int bufferSize) {
84          this(bufferSize, null);
85      }
86  
87      /**
88       * Constructor which sets buffer size to <code>bufferSize</code>. If 
89       * <code>buffersMap</code> is null then a default instance of {@link ConcurrentHashMap} 
90       * is created else the provided instance is used.
91       * 
92       * @param bufferSize the new buffer size
93       * @param buffersMap the map to use for storing each session buffer 
94       */
95      public BufferedWriteFilter(int bufferSize,
96              LazyInitializedCacheMap<IoSession, IoBuffer> buffersMap) {
97          super();
98          this.bufferSize = bufferSize;
99          if (buffersMap == null) {
100             this.buffersMap = new LazyInitializedCacheMap<IoSession, IoBuffer>();
101         } else {
102             this.buffersMap = buffersMap;
103         }
104     }
105 
106     /**
107      * Returns buffer size.
108      */
109     public int getBufferSize() {
110         return bufferSize;
111     }
112 
113     /**
114      * Sets the buffer size but only for the newly created buffers.
115      * 
116      * @param bufferSize the new buffer size
117      */
118     public void setBufferSize(int bufferSize) {
119         this.bufferSize = bufferSize;
120     }
121 
122     /**
123      * {@inheritDoc}
124      * 
125      * @throws Exception if <code>writeRequest.message</code> isn't an
126      *                   {@link IoBuffer} instance.
127      */
128     @Override
129     public void filterWrite(NextFilter nextFilter, IoSession session,
130             WriteRequest writeRequest) throws Exception {
131 
132         Object data = writeRequest.getMessage();
133 
134         if (data instanceof IoBuffer) {
135             write(session, (IoBuffer) data);
136         } else {
137             throw new IllegalArgumentException(
138                     "This filter should only buffer IoBuffer objects");
139         }
140     }
141 
142     /**
143      * Writes an {@link IoBuffer} to the session's buffer.
144      * 
145      * @param session the session to which a write is requested
146      * @param data the data to buffer
147      */
148     private void write(IoSession session, IoBuffer data) {
149         IoBuffer dest = buffersMap.putIfAbsent(session,
150                 new IoBufferLazyInitializer(bufferSize));
151 
152         write(session, data, dest);
153     }
154 
155     /**
156      * Writes <code>data</code> {@link IoBuffer} to the <code>buf</code>
157      * {@link IoBuffer} which buffers write requests for the
158      * <code>session</code> {@ link IoSession} until buffer is full 
159      * or manually flushed.
160      * 
161      * @param session the session where buffer will be written
162      * @param data the data to buffer
163      * @param buf the buffer where data will be temporarily written 
164      */
165     private void write(IoSession session, IoBuffer data, IoBuffer buf) {
166         try {
167             int len = data.remaining();
168             if (len >= buf.capacity()) {
169                 /*
170                  * If the request length exceeds the size of the output buffer,
171                  * flush the output buffer and then write the data directly.
172                  */
173                 NextFilter nextFilter = session.getFilterChain().getNextFilter(
174                         this);
175                 internalFlush(nextFilter, session, buf);
176                 nextFilter.filterWrite(session, new DefaultWriteRequest(data));
177                 return;
178             }
179             if (len > (buf.limit() - buf.position())) {
180                 internalFlush(session.getFilterChain().getNextFilter(this),
181                         session, buf);
182             }
183             synchronized (buf) {
184                 buf.put(data);
185             }
186         } catch (Throwable e) {
187             session.getFilterChain().fireExceptionCaught(e);
188         }
189     }
190 
191     /**
192      * Internal method that actually flushes the buffered data.
193      * 
194      * @param nextFilter the {@link NextFilter} of this filter
195      * @param session the session where buffer will be written
196      * @param buf the data to write
197      * @throws Exception if a write operation fails
198      */
199     private void internalFlush(NextFilter nextFilter, IoSession session,
200             IoBuffer buf) throws Exception {
201         IoBuffer tmp = null;
202         synchronized (buf) {
203             buf.flip();
204             tmp = buf.duplicate();
205             buf.clear();
206         }
207         logger.debug("Flushing buffer: {}", tmp);
208         nextFilter.filterWrite(session, new DefaultWriteRequest(tmp));
209     }
210 
211     /**
212      * Flushes the buffered data.
213      * 
214      * @param session the session where buffer will be written
215      */
216     public void flush(IoSession session) {
217         try {
218             internalFlush(session.getFilterChain().getNextFilter(this),
219                     session, buffersMap.get(session));
220         } catch (Throwable e) {
221             session.getFilterChain().fireExceptionCaught(e);
222         }
223     }
224 
225     /**
226      * Internal method that actually frees the {@link IoBuffer} that contains
227      * the buffered data that has not been flushed.
228      * 
229      * @param session the session we operate on
230      */
231     private void free(IoSession session) {
232         IoBuffer buf = buffersMap.remove(session);
233         if (buf != null) {
234             buf.free();
235         }
236     }
237 
238     /**
239      * {@inheritDoc}
240      */
241     @Override
242     public void exceptionCaught(NextFilter nextFilter, IoSession session,
243             Throwable cause) throws Exception {
244         free(session);
245         nextFilter.exceptionCaught(session, cause);
246     }
247 
248     /**
249      * {@inheritDoc}
250      */
251     @Override
252     public void sessionClosed(NextFilter nextFilter, IoSession session)
253             throws Exception {
254         free(session);
255         nextFilter.sessionClosed(session);
256     }
257 }