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 }