1
2
3
4
5
6
7
8
9
10
11
12
13
14 package org.eclipse.jetty.io.nio;
15
16 import java.io.File;
17 import java.io.FileInputStream;
18 import java.io.IOException;
19 import java.io.InputStream;
20 import java.io.OutputStream;
21 import java.nio.ByteBuffer;
22 import java.nio.channels.Channels;
23 import java.nio.channels.FileChannel;
24 import java.nio.channels.ReadableByteChannel;
25 import java.nio.channels.WritableByteChannel;
26
27 import org.eclipse.jetty.io.AbstractBuffer;
28 import org.eclipse.jetty.io.Buffer;
29
30
31
32
33
34
35 public class DirectNIOBuffer extends AbstractBuffer implements NIOBuffer
36 {
37 protected ByteBuffer _buf;
38 private ReadableByteChannel _in;
39 private InputStream _inStream;
40 private WritableByteChannel _out;
41 private OutputStream _outStream;
42
43 public DirectNIOBuffer(int size)
44 {
45 super(READWRITE,NON_VOLATILE);
46 _buf = ByteBuffer.allocateDirect(size);
47 _buf.position(0);
48 _buf.limit(_buf.capacity());
49 }
50
51 public DirectNIOBuffer(ByteBuffer buffer,boolean immutable)
52 {
53 super(immutable?IMMUTABLE:READWRITE,NON_VOLATILE);
54 if (!buffer.isDirect())
55 throw new IllegalArgumentException();
56 _buf = buffer;
57 setGetIndex(buffer.position());
58 setPutIndex(buffer.limit());
59 }
60
61
62
63
64 public DirectNIOBuffer(File file) throws IOException
65 {
66 super(READONLY,NON_VOLATILE);
67 FileInputStream fis = new FileInputStream(file);
68 FileChannel fc = fis.getChannel();
69 _buf = fc.map(FileChannel.MapMode.READ_ONLY, 0, file.length());
70 setGetIndex(0);
71 setPutIndex((int)file.length());
72 _access=IMMUTABLE;
73 }
74
75
76 public boolean isDirect()
77 {
78 return true;
79 }
80
81
82 public byte[] array()
83 {
84 return null;
85 }
86
87
88 public int capacity()
89 {
90 return _buf.capacity();
91 }
92
93
94 public byte peek(int position)
95 {
96 return _buf.get(position);
97 }
98
99 public int peek(int index, byte[] b, int offset, int length)
100 {
101 int l = length;
102 if (index+l > capacity())
103 {
104 l=capacity()-index;
105 if (l==0)
106 return -1;
107 }
108
109 if (l < 0)
110 return -1;
111 try
112 {
113 _buf.position(index);
114 _buf.get(b,offset,l);
115 }
116 finally
117 {
118 _buf.position(0);
119 }
120
121 return l;
122 }
123
124 public void poke(int index, byte b)
125 {
126 if (isReadOnly()) throw new IllegalStateException(__READONLY);
127 if (index < 0) throw new IllegalArgumentException("index<0: " + index + "<0");
128 if (index > capacity())
129 throw new IllegalArgumentException("index>capacity(): " + index + ">" + capacity());
130 _buf.put(index,b);
131 }
132
133 public int poke(int index, Buffer src)
134 {
135 if (isReadOnly()) throw new IllegalStateException(__READONLY);
136
137 byte[] array=src.array();
138 if (array!=null)
139 {
140 int length = poke(index,array,src.getIndex(),src.length());
141 return length;
142 }
143 else
144 {
145 Buffer src_buf=src.buffer();
146 if (src_buf instanceof DirectNIOBuffer)
147 {
148 ByteBuffer src_bytebuf = ((DirectNIOBuffer)src_buf)._buf;
149 if (src_bytebuf==_buf)
150 src_bytebuf=_buf.duplicate();
151 try
152 {
153 _buf.position(index);
154 int space = _buf.remaining();
155
156 int length=src.length();
157 if (length>space)
158 length=space;
159
160 src_bytebuf.position(src.getIndex());
161 src_bytebuf.limit(src.getIndex()+length);
162
163 _buf.put(src_bytebuf);
164 return length;
165 }
166 finally
167 {
168 _buf.position(0);
169 src_bytebuf.limit(src_bytebuf.capacity());
170 src_bytebuf.position(0);
171 }
172 }
173 else
174 return super.poke(index,src);
175 }
176 }
177
178 public int poke(int index, byte[] b, int offset, int length)
179 {
180 if (isReadOnly()) throw new IllegalStateException(__READONLY);
181
182 if (index < 0) throw new IllegalArgumentException("index<0: " + index + "<0");
183
184 if (index + length > capacity())
185 {
186 length=capacity()-index;
187 if (length<0)
188 throw new IllegalArgumentException("index>capacity(): " + index + ">" + capacity());
189 }
190
191 try
192 {
193 _buf.position(index);
194
195 int space=_buf.remaining();
196
197 if (length>space)
198 length=space;
199 if (length>0)
200 _buf.put(b,offset,length);
201 return length;
202 }
203 finally
204 {
205 _buf.position(0);
206 }
207 }
208
209
210 public ByteBuffer getByteBuffer()
211 {
212 return _buf;
213 }
214
215
216 public int readFrom(InputStream in, int max) throws IOException
217 {
218 if (_in==null || !_in.isOpen() || in!=_inStream)
219 {
220 _in=Channels.newChannel(in);
221 _inStream=in;
222 }
223
224 if (max<0 || max>space())
225 max=space();
226 int p = putIndex();
227
228 try
229 {
230 int len=0, total=0, available=max;
231 int loop=0;
232 while (total<max)
233 {
234 _buf.position(p);
235 _buf.limit(p+available);
236 len=_in.read(_buf);
237 if (len<0)
238 {
239 _in=null;
240 _inStream=in;
241 break;
242 }
243 else if (len>0)
244 {
245 p += len;
246 total += len;
247 available -= len;
248 setPutIndex(p);
249 loop=0;
250 }
251 else if (loop++>1)
252 break;
253 if (in.available()<=0)
254 break;
255 }
256 if (len<0 && total==0)
257 return -1;
258 return total;
259
260 }
261 catch(IOException e)
262 {
263 _in=null;
264 _inStream=in;
265 throw e;
266 }
267 finally
268 {
269 if (_in!=null && !_in.isOpen())
270 {
271 _in=null;
272 _inStream=in;
273 }
274 _buf.position(0);
275 _buf.limit(_buf.capacity());
276 }
277 }
278
279
280 public void writeTo(OutputStream out) throws IOException
281 {
282 if (_out==null || !_out.isOpen() || _out!=_outStream)
283 {
284 _out=Channels.newChannel(out);
285 _outStream=out;
286 }
287
288 synchronized (_buf)
289 {
290 try
291 {
292 int loop=0;
293 while(hasContent() && _out.isOpen())
294 {
295 _buf.position(getIndex());
296 _buf.limit(putIndex());
297 int len=_out.write(_buf);
298 if (len<0)
299 break;
300 else if (len>0)
301 {
302 skip(len);
303 loop=0;
304 }
305 else if (loop++>1)
306 break;
307 }
308
309 }
310 catch(IOException e)
311 {
312 _out=null;
313 _outStream=null;
314 throw e;
315 }
316 finally
317 {
318 if (_out!=null && !_out.isOpen())
319 {
320 _out=null;
321 _outStream=null;
322 }
323 _buf.position(0);
324 _buf.limit(_buf.capacity());
325 }
326 }
327 }
328
329
330
331 }