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;
21
22 import java.io.IOException;
23 import java.io.InputStream;
24 import java.util.Queue;
25 import java.util.concurrent.ConcurrentLinkedQueue;
26
27 import org.apache.mina.common.ByteBuffer;
28 import org.apache.mina.common.IoFilterAdapter;
29 import org.apache.mina.common.IoSession;
30 import org.apache.mina.common.WriteFuture;
31
32 /**
33 * Filter implementation which makes it possible to write {@link InputStream}
34 * objects directly using {@link IoSession#write(Object)}. When an
35 * {@link InputStream} is written to a session this filter will read the bytes
36 * from the stream into {@link ByteBuffer} objects and write those buffers
37 * to the next filter. When end of stream has been reached this filter will
38 * call {@link NextFilter#messageSent(IoSession, Object)} using the original
39 * {@link InputStream} written to the session and notifies
40 * {@link org.apache.mina.common.WriteFuture} on the
41 * original {@link org.apache.mina.common.IoFilter.WriteRequest}.
42 * <p/>
43 * This filter will ignore written messages which aren't {@link InputStream}
44 * instances. Such messages will be passed to the next filter directly.
45 * </p>
46 * <p/>
47 * NOTE: this filter does not close the stream after all data from stream
48 * has been written. The {@link org.apache.mina.common.IoHandler} should take
49 * care of that in its
50 * {@link org.apache.mina.common.IoHandler#messageSent(IoSession, Object)}
51 * callback.
52 * </p>
53 *
54 * @author The Apache Directory Project (mina-dev@directory.apache.org)
55 * @version $Rev: 587373 $, $Date: 2007-10-23 11:54:05 +0900 (Tue, 23 Oct 2007) $
56 */
57 public class StreamWriteFilter extends IoFilterAdapter {
58 /**
59 * The default buffer size this filter uses for writing.
60 */
61 public static final int DEFAULT_STREAM_BUFFER_SIZE = 4096;
62
63 /**
64 * The attribute name used when binding the {@link InputStream} to the session.
65 */
66 public static final String CURRENT_STREAM = StreamWriteFilter.class
67 .getName()
68 + ".stream";
69
70 protected static final String WRITE_REQUEST_QUEUE = StreamWriteFilter.class
71 .getName()
72 + ".queue";
73
74 protected static final String INITIAL_WRITE_FUTURE = StreamWriteFilter.class
75 .getName()
76 + ".future";
77
78 private int writeBufferSize = DEFAULT_STREAM_BUFFER_SIZE;
79
80 @Override
81 public void filterWrite(NextFilter nextFilter, IoSession session,
82 WriteRequest writeRequest) throws Exception {
83 // If we're already processing a stream we need to queue the WriteRequest.
84 if (session.getAttribute(CURRENT_STREAM) != null) {
85 Queue<WriteRequest> queue = getWriteRequestQueue(session);
86 if (queue == null) {
87 queue = new ConcurrentLinkedQueue<WriteRequest>();
88 session.setAttribute(WRITE_REQUEST_QUEUE, queue);
89 }
90 queue.add(writeRequest);
91 return;
92 }
93
94 Object message = writeRequest.getMessage();
95
96 if (message instanceof InputStream) {
97
98 InputStream inputStream = (InputStream) message;
99
100 ByteBuffer byteBuffer = getNextByteBuffer(inputStream);
101 if (byteBuffer == null) {
102 // End of stream reached.
103 writeRequest.getFuture().setWritten(true);
104 nextFilter.messageSent(session, message);
105 } else {
106 session.setAttribute(CURRENT_STREAM, inputStream);
107 session.setAttribute(INITIAL_WRITE_FUTURE, writeRequest
108 .getFuture());
109
110 nextFilter.filterWrite(session, new WriteRequest(byteBuffer));
111 }
112
113 } else {
114 nextFilter.filterWrite(session, writeRequest);
115 }
116 }
117
118 @SuppressWarnings("unchecked")
119 private Queue<WriteRequest> getWriteRequestQueue(IoSession session) {
120 return (Queue<WriteRequest>) session.getAttribute(WRITE_REQUEST_QUEUE);
121 }
122
123 @Override
124 public void messageSent(NextFilter nextFilter, IoSession session,
125 Object message) throws Exception {
126 InputStream inputStream = (InputStream) session
127 .getAttribute(CURRENT_STREAM);
128
129 if (inputStream == null) {
130 nextFilter.messageSent(session, message);
131 } else {
132 ByteBuffer byteBuffer = getNextByteBuffer(inputStream);
133
134 if (byteBuffer == null) {
135 // End of stream reached.
136 session.removeAttribute(CURRENT_STREAM);
137 WriteFuture writeFuture = (WriteFuture) session
138 .removeAttribute(INITIAL_WRITE_FUTURE);
139
140 // Write queued WriteRequests.
141 Queue<? extends WriteRequest> queue = (Queue<? extends WriteRequest>) session
142 .removeAttribute(WRITE_REQUEST_QUEUE);
143 if (queue != null) {
144 WriteRequest wr = queue.poll();
145 while (wr != null) {
146 filterWrite(nextFilter, session, wr);
147 wr = queue.poll();
148 }
149 }
150
151 writeFuture.setWritten(true);
152 nextFilter.messageSent(session, inputStream);
153 } else {
154 nextFilter.filterWrite(session, new WriteRequest(byteBuffer));
155 }
156 }
157 }
158
159 private ByteBuffer getNextByteBuffer(InputStream is) throws IOException {
160 byte[] bytes = new byte[writeBufferSize];
161
162 int off = 0;
163 int n = 0;
164 while (off < bytes.length
165 && (n = is.read(bytes, off, bytes.length - off)) != -1) {
166 off += n;
167 }
168
169 if (n == -1 && off == 0) {
170 return null;
171 }
172
173 return ByteBuffer.wrap(bytes, 0, off);
174 }
175
176 /**
177 * Returns the size of the write buffer in bytes. Data will be read from the
178 * stream in chunks of this size and then written to the next filter.
179 *
180 * @return the write buffer size.
181 */
182 public int getWriteBufferSize() {
183 return writeBufferSize;
184 }
185
186 /**
187 * Sets the size of the write buffer in bytes. Data will be read from the
188 * stream in chunks of this size and then written to the next filter.
189 *
190 * @throws IllegalArgumentException if the specified size is < 1.
191 */
192 public void setWriteBufferSize(int writeBufferSize) {
193 if (writeBufferSize < 1) {
194 throw new IllegalArgumentException(
195 "writeBufferSize must be at least 1");
196 }
197 this.writeBufferSize = writeBufferSize;
198 }
199
200 }