Mercurial Hosting > luan

comparison src/org/eclipse/jetty/io/nio/ChannelEndPoint.java @ 1062:4a50422596b6

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
fix JBuffer.getByteBuffer()
author Franklin Schmidt <fschmidt@gmail.com>
date Wed, 09 Nov 2016 02:04:02 -0700
parents 4afdf0f0c5bc
children ebb0f1343ef6
comparison
equal deleted inserted replaced
1061:c880589715c7 1062:4a50422596b6
41 public class ChannelEndPoint implements EndPoint 41 public class ChannelEndPoint implements EndPoint
42 { 42 {
43 private static final Logger LOG = LoggerFactory.getLogger(ChannelEndPoint.class); 43 private static final Logger LOG = LoggerFactory.getLogger(ChannelEndPoint.class);
44 44
45 private final SocketChannel _channel; 45 private final SocketChannel _channel;
46 private final ByteBuffer[] _gather2 = new ByteBuffer[2];
47 protected final Socket _socket; 46 protected final Socket _socket;
48 private final InetSocketAddress _local; 47 private final InetSocketAddress _local;
49 private final InetSocketAddress _remote; 48 private final InetSocketAddress _remote;
50 protected volatile int _maxIdleTime; 49 protected volatile int _maxIdleTime;
51 private volatile boolean _ishut = false; 50 private volatile boolean _ishut = false;
167 public int fill(JBuffer buffer) throws IOException 166 public int fill(JBuffer buffer) throws IOException
168 { 167 {
169 if (_ishut) 168 if (_ishut)
170 return -1; 169 return -1;
171 int len = 0; 170 int len = 0;
172 final ByteBuffer bbuf = buffer.getByteBuffer(); 171 final ByteBuffer bbuf = buffer.getByteBuffer().duplicate();
173 172
174 //noinspection SynchronizationOnLocalVariableOrMethodParameter 173 //noinspection SynchronizationOnLocalVariableOrMethodParameter
175 try 174 try
176 { 175 {
177 synchronized(bbuf) 176 bbuf.limit(bbuf.capacity());
178 { 177 bbuf.position(buffer.limit());
179 try 178 len = _channel.read(bbuf);
180 { 179 buffer.limit(bbuf.position());
181 bbuf.position(buffer.putIndex());
182 len = _channel.read(bbuf);
183 }
184 finally
185 {
186 buffer.setPutIndex(bbuf.position());
187 bbuf.position(0);
188 }
189 }
190 180
191 if (len<0 && isOpen()) 181 if (len<0 && isOpen())
192 { 182 {
193 if (!isInputShutdown()) 183 if (!isInputShutdown())
194 shutdownInput(); 184 shutdownInput();
218 } 208 }
219 209
220 @Override 210 @Override
221 public int flush(JBuffer buffer) throws IOException 211 public int flush(JBuffer buffer) throws IOException
222 { 212 {
223 int len=0; 213 final ByteBuffer bbuf = buffer.getByteBuffer();
224 final ByteBuffer bbuf = buffer.getByteBuffer().asReadOnlyBuffer(); 214 return _channel.write(bbuf);
225 try
226 {
227 bbuf.position(buffer.getIndex());
228 bbuf.limit(buffer.putIndex());
229 len=_channel.write(bbuf);
230 }
231 finally
232 {
233 if (len>0)
234 buffer.skip(len);
235 }
236 return len;
237 } 215 }
238 216
239 @Override 217 @Override
240 public int flush(JBuffer header, JBuffer buffer, JBuffer trailer) throws IOException 218 public int flush(JBuffer header, JBuffer buffer, JBuffer trailer) throws IOException
241 { 219 {
243 221
244 if (_channel instanceof GatheringByteChannel && 222 if (_channel instanceof GatheringByteChannel &&
245 header!=null && header.remaining()!=0 && 223 header!=null && header.remaining()!=0 &&
246 buffer!=null && buffer.remaining()!=0) 224 buffer!=null && buffer.remaining()!=0)
247 { 225 {
248 length = gatheringFlush(header,header.getByteBuffer(),buffer,buffer.getByteBuffer()); 226 length = gatheringFlush(header,buffer);
249 } 227 }
250 else 228 else
251 { 229 {
252 // flush header 230 // flush header
253 if (header!=null && header.remaining()>0) 231 if (header!=null && header.remaining()>0)
266 } 244 }
267 245
268 return length; 246 return length;
269 } 247 }
270 248
271 private int gatheringFlush(JBuffer header, ByteBuffer bbuf0, JBuffer buffer, ByteBuffer bbuf1) throws IOException 249 private synchronized int gatheringFlush(JBuffer header, JBuffer buffer) throws IOException
272 { 250 {
273 int length; 251 ByteBuffer bbuf0 = header.getByteBuffer();
274 252 ByteBuffer bbuf1 = buffer.getByteBuffer();
275 synchronized(this) 253 return (int)_channel.write(new ByteBuffer[]{bbuf0,bbuf1});
276 {
277 // Adjust position indexs of buf0 and buf1
278 bbuf0=bbuf0.asReadOnlyBuffer();
279 bbuf0.position(header.getIndex());
280 bbuf0.limit(header.putIndex());
281 bbuf1=bbuf1.asReadOnlyBuffer();
282 bbuf1.position(buffer.getIndex());
283 bbuf1.limit(buffer.putIndex());
284
285 _gather2[0]=bbuf0;
286 _gather2[1]=bbuf1;
287
288 // do the gathering write.
289 length=(int)((GatheringByteChannel)_channel).write(_gather2);
290
291 int hl=header.remaining();
292 if (length>hl)
293 {
294 header.clear();
295 buffer.skip(length-hl);
296 }
297 else if (length>0)
298 {
299 header.skip(length);
300 }
301 }
302 return length;
303 } 254 }
304 255
305 public final SocketChannel getChannel() 256 public final SocketChannel getChannel()
306 { 257 {
307 return _channel; 258 return _channel;