1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.mortbay.io.nio;
17
18 import java.io.IOException;
19 import java.net.InetSocketAddress;
20 import java.net.Socket;
21 import java.nio.ByteBuffer;
22 import java.nio.channels.ByteChannel;
23 import java.nio.channels.GatheringByteChannel;
24 import java.nio.channels.SelectableChannel;
25 import java.nio.channels.SocketChannel;
26
27 import org.mortbay.io.Buffer;
28 import org.mortbay.io.EndPoint;
29 import org.mortbay.io.Portable;
30 import org.mortbay.log.Log;
31
32
33
34
35
36
37
38
39 public class ChannelEndPoint implements EndPoint
40 {
41 protected ByteChannel _channel;
42 protected ByteBuffer[] _gather2=new ByteBuffer[2];
43 protected Socket _socket;
44 protected InetSocketAddress _local;
45 protected InetSocketAddress _remote;
46
47
48
49
50 public ChannelEndPoint(ByteChannel channel)
51 {
52 super();
53 this._channel = channel;
54 if (channel instanceof SocketChannel)
55 _socket=((SocketChannel)channel).socket();
56 }
57
58 public boolean isBlocking()
59 {
60 if (_channel instanceof SelectableChannel)
61 return ((SelectableChannel)_channel).isBlocking();
62 return true;
63 }
64
65 public boolean blockReadable(long millisecs) throws IOException
66 {
67 return true;
68 }
69
70 public boolean blockWritable(long millisecs) throws IOException
71 {
72 return true;
73 }
74
75
76
77
78 public boolean isOpen()
79 {
80 return _channel.isOpen();
81 }
82
83
84
85
86 public void close() throws IOException
87 {
88 if (_channel.isOpen())
89 {
90 try
91 {
92 if (_channel instanceof SocketChannel)
93 {
94
95 Socket socket= ((SocketChannel)_channel).socket();
96 if (!socket.isClosed() && !socket.isOutputShutdown())
97 socket.shutdownOutput();
98 }
99 }
100 catch(IOException e)
101 {
102 Log.ignore(e);
103 }
104 catch(UnsupportedOperationException e)
105 {
106 Log.ignore(e);
107 }
108 finally
109 {
110 _channel.close();
111 }
112 }
113 }
114
115
116
117
118 public int fill(Buffer buffer) throws IOException
119 {
120 Buffer buf = buffer.buffer();
121 int len=0;
122 if (buf instanceof NIOBuffer)
123 {
124 NIOBuffer nbuf = (NIOBuffer)buf;
125 ByteBuffer bbuf=nbuf.getByteBuffer();
126 synchronized(nbuf)
127 {
128 try
129 {
130 bbuf.position(buffer.putIndex());
131 len=_channel.read(bbuf);
132 if (len<0)
133 _channel.close();
134 }
135 finally
136 {
137 buffer.setPutIndex(bbuf.position());
138 bbuf.position(0);
139 }
140 }
141 }
142 else
143 {
144 throw new IOException("Not Implemented");
145 }
146
147 return len;
148 }
149
150
151
152
153 public int flush(Buffer buffer) throws IOException
154 {
155 Buffer buf = buffer.buffer();
156 int len=0;
157 if (buf instanceof NIOBuffer)
158 {
159 NIOBuffer nbuf = (NIOBuffer)buf;
160 ByteBuffer bbuf=nbuf.getByteBuffer();
161
162
163 synchronized(bbuf)
164 {
165 try
166 {
167 bbuf.position(buffer.getIndex());
168 bbuf.limit(buffer.putIndex());
169 len=_channel.write(bbuf);
170 }
171 finally
172 {
173 if (len>0)
174 buffer.skip(len);
175 bbuf.position(0);
176 bbuf.limit(bbuf.capacity());
177 }
178 }
179 }
180 else if (buffer.array()!=null)
181 {
182 ByteBuffer b = ByteBuffer.wrap(buffer.array(), buffer.getIndex(), buffer.length());
183 len=_channel.write(b);
184 if (len>0)
185 buffer.skip(len);
186 }
187 else
188 {
189 throw new IOException("Not Implemented");
190 }
191 return len;
192 }
193
194
195
196
197 public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
198 {
199 int length=0;
200
201 Buffer buf0 = header==null?null:header.buffer();
202 Buffer buf1 = buffer==null?null:buffer.buffer();
203
204 if (_channel instanceof GatheringByteChannel &&
205 header!=null && header.length()!=0 && header instanceof NIOBuffer &&
206 buffer!=null && buffer.length()!=0 && buffer instanceof NIOBuffer)
207 {
208 NIOBuffer nbuf0 = (NIOBuffer)buf0;
209 ByteBuffer bbuf0=nbuf0.getByteBuffer();
210 NIOBuffer nbuf1 = (NIOBuffer)buf1;
211 ByteBuffer bbuf1=nbuf1.getByteBuffer();
212
213 synchronized(this)
214 {
215
216 synchronized(bbuf0)
217 {
218 synchronized(bbuf1)
219 {
220 try
221 {
222
223 bbuf0.position(header.getIndex());
224 bbuf0.limit(header.putIndex());
225 bbuf1.position(buffer.getIndex());
226 bbuf1.limit(buffer.putIndex());
227
228 _gather2[0]=bbuf0;
229 _gather2[1]=bbuf1;
230
231
232 length=(int)((GatheringByteChannel)_channel).write(_gather2);
233
234 int hl=header.length();
235 if (length>hl)
236 {
237 header.clear();
238 buffer.skip(length-hl);
239 }
240 else if (length>0)
241 {
242 header.skip(length);
243 }
244
245 }
246 finally
247 {
248
249 if (!header.isImmutable())
250 header.setGetIndex(bbuf0.position());
251 if (!buffer.isImmutable())
252 buffer.setGetIndex(bbuf1.position());
253
254 bbuf0.position(0);
255 bbuf1.position(0);
256 bbuf0.limit(bbuf0.capacity());
257 bbuf1.limit(bbuf1.capacity());
258 }
259 }
260 }
261 }
262 }
263 else
264 {
265 if (header!=null)
266 {
267 if (buffer!=null && buffer.length()>0 && header.space()>buffer.length())
268 {
269 header.put(buffer);
270 buffer.clear();
271 }
272 if (trailer!=null && trailer.length()>0 && header.space()>trailer.length())
273 {
274 header.put(trailer);
275 trailer.clear();
276 }
277 }
278
279
280 if (header!=null && header.length()>0)
281 length=flush(header);
282
283
284 if ((header==null || header.length()==0) &&
285 buffer!=null && buffer.length()>0)
286 length+=flush(buffer);
287
288
289 if ((header==null || header.length()==0) &&
290 (buffer==null || buffer.length()==0) &&
291 trailer!=null && trailer.length()>0)
292 length+=flush(trailer);
293 }
294
295 return length;
296 }
297
298
299
300
301 public ByteChannel getChannel()
302 {
303 return _channel;
304 }
305
306
307
308
309
310
311 public String getLocalAddr()
312 {
313 if (_socket==null)
314 return null;
315
316 if (_local==null)
317 _local=(InetSocketAddress)_socket.getLocalSocketAddress();
318
319 if (_local==null || _local.getAddress()==null || _local.getAddress().isAnyLocalAddress())
320 return Portable.ALL_INTERFACES;
321
322 return _local.getAddress().getHostAddress();
323 }
324
325
326
327
328
329 public String getLocalHost()
330 {
331 if (_socket==null)
332 return null;
333
334 if (_local==null)
335 _local=(InetSocketAddress)_socket.getLocalSocketAddress();
336
337 if (_local==null || _local.getAddress()==null || _local.getAddress().isAnyLocalAddress())
338 return Portable.ALL_INTERFACES;
339
340 return _local.getAddress().getCanonicalHostName();
341 }
342
343
344
345
346
347 public int getLocalPort()
348 {
349 if (_socket==null)
350 return 0;
351
352 if (_local==null)
353 _local=(InetSocketAddress)_socket.getLocalSocketAddress();
354 if (_local==null)
355 return -1;
356 return _local.getPort();
357 }
358
359
360
361
362
363 public String getRemoteAddr()
364 {
365 if (_socket==null)
366 return null;
367
368 if (_remote==null)
369 _remote=(InetSocketAddress)_socket.getRemoteSocketAddress();
370
371 if (_remote==null)
372 return null;
373 return _remote.getAddress().getHostAddress();
374 }
375
376
377
378
379
380 public String getRemoteHost()
381 {
382 if (_socket==null)
383 return null;
384
385 if (_remote==null)
386 _remote=(InetSocketAddress)_socket.getRemoteSocketAddress();
387
388 if (_remote==null)
389 return null;
390 return _remote.getAddress().getCanonicalHostName();
391 }
392
393
394
395
396
397 public int getRemotePort()
398 {
399 if (_socket==null)
400 return 0;
401
402 if (_remote==null)
403 _remote=(InetSocketAddress)_socket.getRemoteSocketAddress();
404
405 if (_remote==null)
406 return -1;
407 return _remote==null?-1:_remote.getPort();
408 }
409
410
411
412
413
414 public Object getTransport()
415 {
416 return _channel;
417 }
418
419
420 public void flush()
421 throws IOException
422 {
423 }
424
425
426 public boolean isBufferingInput()
427 {
428 return false;
429 }
430
431
432 public boolean isBufferingOutput()
433 {
434 return false;
435 }
436
437
438 public boolean isBufferred()
439 {
440 return false;
441 }
442 }