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;
    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 04:54:05 +0200 (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 &lt; 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 }