View Javadoc

    1   /*
    2    * Copyright 2009 Red Hat, Inc.
    3    *
    4    * Red Hat licenses this file to you under the Apache License, version 2.0
    5    * (the "License"); you may not use this file except in compliance with the
    6    * License.  You may obtain a copy of the License at:
    7    *
    8    *    http://www.apache.org/licenses/LICENSE-2.0
    9    *
    10   * Unless required by applicable law or agreed to in writing, software
    11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
    13   * License for the specific language governing permissions and limitations
    14   * under the License.
    15   */
    16  package org.jboss.netty.handler.stream;
    17
    18  import static org.jboss.netty.buffer.ChannelBuffers.*;
    19
    20  import java.nio.ByteBuffer;
    21  import java.nio.channels.ReadableByteChannel;
    22
    23  import org.jboss.netty.buffer.ChannelBuffer;
    24
    25  /**
    26   * A {@link ChunkedInput} that fetches data from a {@link ReadableByteChannel}
    27   * chunk by chunk.  Please note that the {@link ReadableByteChannel} must
    28   * operate in blocking mode.  Non-blocking mode channels are not supported.
    29   *
    30   * @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
    31   * @author <a href="http://gleamynode.net/">Trustin Lee</a>
    32   * @author Frederic Bregier
    33   * @version $Rev: 2236 $, $Date: 2010-04-12 19:22:51 +0900 (Mon, 12 Apr 2010) $
    34   */
    35  public class ChunkedNioStream implements ChunkedInput {
    36
    37      private final ReadableByteChannel in;
    38
    39      private final int chunkSize;
    40      private volatile long offset;
    41
    42      /**
    43       * Associated ByteBuffer
    44       */
    45      private final ByteBuffer byteBuffer;
    46
    47      /**
    48       * Creates a new instance that fetches data from the specified channel.
    49       */
    50      public ChunkedNioStream(ReadableByteChannel in) {
    51          this(in, ChunkedStream.DEFAULT_CHUNK_SIZE);
    52      }
    53
    54      /**
    55       * Creates a new instance that fetches data from the specified channel.
    56       *
    57       * @param chunkSize the number of bytes to fetch on each
    58       *                  {@link #nextChunk()} call
    59       */
    60      public ChunkedNioStream(ReadableByteChannel in, int chunkSize) {
    61          if (in == null) {
    62              throw new NullPointerException("in");
    63          }
    64          if (chunkSize <= 0) {
    65              throw new IllegalArgumentException("chunkSize: " + chunkSize +
    66                      " (expected: a positive integer)");
    67          }
    68          this.in = in;
    69          offset = 0;
    70          this.chunkSize = chunkSize;
    71          byteBuffer = ByteBuffer.allocate(chunkSize);
    72      }
    73
    74      /**
    75       * Returns the number of transferred bytes.
    76       */
    77      public long getTransferredBytes() {
    78          return offset;
    79      }
    80
    81      public boolean hasNextChunk() throws Exception {
    82          if (byteBuffer.position() > 0) {
    83              // A previous read was not over, so there is a next chunk in the buffer at least
    84              return true;
    85          }
    86          if (in.isOpen()) {
    87              // Try to read a new part, and keep this part (no rewind)
    88              int b = in.read(byteBuffer);
    89              if (b < 0) {
    90                  return false;
    91              } else {
    92                  offset += b;
    93                  return true;
    94              }
    95          }
    96          return false;
    97      }
    98
    99      public boolean isEndOfInput() throws Exception {
    100         return !hasNextChunk();
    101     }
    102
    103     public void close() throws Exception {
    104         in.close();
    105     }
    106
    107     public Object nextChunk() throws Exception {
    108         if (!hasNextChunk()) {
    109             return null;
    110         }
    111         // buffer cannot be not be empty from there
    112         int readBytes = byteBuffer.position();
    113         for (;;) {
    114             int localReadBytes = in.read(byteBuffer);
    115             if (localReadBytes < 0) {
    116                 break;
    117             }
    118             readBytes += localReadBytes;
    119             offset += localReadBytes;
    120
    121             if (readBytes == chunkSize) {
    122                 break;
    123             }
    124         }
    125         byteBuffer.flip();
    126         // copy since buffer is keeped for next usage
    127         ChannelBuffer buffer = copiedBuffer(byteBuffer);
    128         byteBuffer.clear();
    129         return buffer;
    130     }
    131 }