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.stream;
21  
22  import java.io.IOException;
23  import java.util.Queue;
24  import java.util.concurrent.ConcurrentLinkedQueue;
25  
26  import org.apache.mina.core.buffer.IoBuffer;
27  import org.apache.mina.core.filterchain.IoFilterAdapter;
28  import org.apache.mina.core.filterchain.IoFilterChain;
29  import org.apache.mina.core.session.AttributeKey;
30  import org.apache.mina.core.session.IoSession;
31  import org.apache.mina.core.write.DefaultWriteRequest;
32  import org.apache.mina.core.write.WriteRequest;
33  
34  /**
35   * TODO Add documentation
36   * 
37   * @author <a href="http://mina.apache.org">Apache MINA Project</a>
38   */
39  public abstract class AbstractStreamWriteFilter<T> extends IoFilterAdapter {
40      /**
41       * The default buffer size this filter uses for writing.
42       */
43      public static final int DEFAULT_STREAM_BUFFER_SIZE = 4096;
44  
45      /**
46       * The attribute name used when binding the streaming object to the session.
47       */
48      protected final AttributeKey CURRENT_STREAM = new AttributeKey(getClass(), "stream");
49  
50      protected final AttributeKey WRITE_REQUEST_QUEUE = new AttributeKey(getClass(), "queue");
51  
52      protected final AttributeKey CURRENT_WRITE_REQUEST = new AttributeKey(getClass(), "writeRequest");
53  
54      private int writeBufferSize = DEFAULT_STREAM_BUFFER_SIZE;
55  
56      @Override
57      public void onPreAdd(IoFilterChain parent, String name, NextFilter nextFilter) throws Exception {
58          Class<? extends IoFilterAdapter> clazz = getClass();
59          if (parent.contains(clazz)) {
60              throw new IllegalStateException("Only one " + clazz.getName() + " is permitted.");
61          }
62      }
63  
64      @Override
65      public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
66          // If we're already processing a stream we need to queue the WriteRequest.
67          if (session.getAttribute(CURRENT_STREAM) != null) {
68              Queue<WriteRequest> queue = getWriteRequestQueue(session);
69              if (queue == null) {
70                  queue = new ConcurrentLinkedQueue<WriteRequest>();
71                  session.setAttribute(WRITE_REQUEST_QUEUE, queue);
72              }
73              queue.add(writeRequest);
74              return;
75          }
76  
77          Object message = writeRequest.getMessage();
78  
79          if (getMessageClass().isInstance(message)) {
80  
81              T stream = getMessageClass().cast(message);
82  
83              IoBuffer buffer = getNextBuffer(stream);
84              if (buffer == null) {
85                  // End of stream reached.
86                  writeRequest.getFuture().setWritten();
87                  nextFilter.messageSent(session, writeRequest);
88              } else {
89                  session.setAttribute(CURRENT_STREAM, message);
90                  session.setAttribute(CURRENT_WRITE_REQUEST, writeRequest);
91  
92                  nextFilter.filterWrite(session, new DefaultWriteRequest(buffer));
93              }
94  
95          } else {
96              nextFilter.filterWrite(session, writeRequest);
97          }
98      }
99  
100     abstract protected Class<T> getMessageClass();
101 
102     @SuppressWarnings("unchecked")
103     private Queue<WriteRequest> getWriteRequestQueue(IoSession session) {
104         return (Queue<WriteRequest>) session.getAttribute(WRITE_REQUEST_QUEUE);
105     }
106 
107     @SuppressWarnings("unchecked")
108     private Queue<WriteRequest> removeWriteRequestQueue(IoSession session) {
109         return (Queue<WriteRequest>) session.removeAttribute(WRITE_REQUEST_QUEUE);
110     }
111 
112     @Override
113     public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
114         T stream = getMessageClass().cast(session.getAttribute(CURRENT_STREAM));
115 
116         if (stream == null) {
117             nextFilter.messageSent(session, writeRequest);
118         } else {
119             IoBuffer buffer = getNextBuffer(stream);
120 
121             if (buffer == null) {
122                 // End of stream reached.
123                 session.removeAttribute(CURRENT_STREAM);
124                 WriteRequest currentWriteRequest = (WriteRequest) session.removeAttribute(CURRENT_WRITE_REQUEST);
125 
126                 // Write queued WriteRequests.
127                 Queue<WriteRequest> queue = removeWriteRequestQueue(session);
128                 if (queue != null) {
129                     WriteRequest wr = queue.poll();
130                     while (wr != null) {
131                         filterWrite(nextFilter, session, wr);
132                         wr = queue.poll();
133                     }
134                 }
135 
136                 currentWriteRequest.getFuture().setWritten();
137                 nextFilter.messageSent(session, currentWriteRequest);
138             } else {
139                 nextFilter.filterWrite(session, new DefaultWriteRequest(buffer));
140             }
141         }
142     }
143 
144     /**
145      * Returns the size of the write buffer in bytes. Data will be read from the
146      * stream in chunks of this size and then written to the next filter.
147      *
148      * @return the write buffer size.
149      */
150     public int getWriteBufferSize() {
151         return writeBufferSize;
152     }
153 
154     /**
155      * Sets the size of the write buffer in bytes. Data will be read from the
156      * stream in chunks of this size and then written to the next filter.
157      *
158      * @throws IllegalArgumentException if the specified size is &lt; 1.
159      */
160     public void setWriteBufferSize(int writeBufferSize) {
161         if (writeBufferSize < 1) {
162             throw new IllegalArgumentException("writeBufferSize must be at least 1");
163         }
164         this.writeBufferSize = writeBufferSize;
165     }
166 
167     abstract protected IoBuffer getNextBuffer(T message) throws IOException;
168 }