View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2012 Mort Bay Consulting Pty. Ltd.
4   //  ------------------------------------------------------------------------
5   //  All rights reserved. This program and the accompanying materials
6   //  are made available under the terms of the Eclipse Public License v1.0
7   //  and Apache License v2.0 which accompanies this distribution.
8   //
9   //      The Eclipse Public License is available at
10  //      http://www.eclipse.org/legal/epl-v10.html
11  //
12  //      The Apache License v2.0 is available at
13  //      http://www.opensource.org/licenses/apache2.0.php
14  //
15  //  You may elect to redistribute this code under either of these licenses.
16  //  ========================================================================
17  //
18  
19  package org.eclipse.jetty.io;
20  
21  import java.io.EOFException;
22  import java.io.IOException;
23  import java.net.InetSocketAddress;
24  import java.net.Socket;
25  import java.net.SocketException;
26  import java.nio.ByteBuffer;
27  import java.nio.channels.ByteChannel;
28  import java.nio.channels.ClosedChannelException;
29  import java.nio.channels.GatheringByteChannel;
30  import java.nio.channels.SocketChannel;
31  
32  import org.eclipse.jetty.util.BufferUtil;
33  import org.eclipse.jetty.util.log.Log;
34  import org.eclipse.jetty.util.log.Logger;
35  import org.eclipse.jetty.util.thread.Scheduler;
36  
37  /**
38   * Channel End Point.
39   * <p>Holds the channel and socket for an NIO endpoint.
40   */
41  public class ChannelEndPoint extends AbstractEndPoint implements SocketBased
42  {
43      private static final Logger LOG = Log.getLogger(ChannelEndPoint.class);
44  
45      private final ByteChannel _channel;
46      private final Socket _socket;
47      private volatile boolean _ishut;
48      private volatile boolean _oshut;
49  
50      public ChannelEndPoint(Scheduler scheduler,SocketChannel channel)
51      {
52          super(scheduler,
53              (InetSocketAddress)channel.socket().getLocalSocketAddress(),
54              (InetSocketAddress)channel.socket().getRemoteSocketAddress());
55          _channel = channel;
56          _socket=channel.socket();
57      }
58  
59      @Override
60      public boolean isOpen()
61      {
62          return _channel.isOpen();
63      }
64  
65      protected void shutdownInput()
66      {
67          LOG.debug("ishut {}", this);
68          _ishut=true;
69          if (_oshut)
70              close();
71      }
72  
73      @Override
74      public void shutdownOutput()
75      {
76          LOG.debug("oshut {}", this);
77          _oshut = true;
78          if (_channel.isOpen())
79          {
80              try
81              {
82                  if (!_socket.isOutputShutdown())
83                      _socket.shutdownOutput();
84              }
85              catch (IOException e)
86              {
87                  LOG.debug(e);
88              }
89              finally
90              {
91                  if (_ishut)
92                  {
93                      close();
94                  }
95              }
96          }
97      }
98  
99      @Override
100     public boolean isOutputShutdown()
101     {
102         return _oshut || !_channel.isOpen() || _socket.isOutputShutdown();
103     }
104 
105     @Override
106     public boolean isInputShutdown()
107     {
108         return _ishut || !_channel.isOpen() || _socket.isInputShutdown();
109     }
110 
111     @Override
112     public void close()
113     {
114         LOG.debug("close {}", this);
115         try
116         {
117             _channel.close();
118         }
119         catch (IOException e)
120         {
121             LOG.debug(e);
122         }
123         finally
124         {
125             _ishut=true;
126             _oshut=true;
127         }
128     }
129 
130     @Override
131     public int fill(ByteBuffer buffer) throws IOException
132     {
133         if (_ishut)
134             return -1;
135 
136         int pos=BufferUtil.flipToFill(buffer);
137         try
138         {
139             int filled = _channel.read(buffer);
140             LOG.debug("filled {} {}", filled, this);
141 
142             if (filled>0)
143                 notIdle();
144             else if (filled==-1)
145                 shutdownInput();
146 
147             return filled;
148         }
149         catch(IOException e)
150         {
151             LOG.debug(e);
152             shutdownInput();
153             return -1;
154         }
155         finally
156         {
157             BufferUtil.flipToFlush(buffer,pos);
158         }
159     }
160 
161     @Override
162     public boolean flush(ByteBuffer... buffers) throws IOException
163     {
164         int flushed=0;
165         try
166         {
167             if (buffers.length==1)
168                 flushed=_channel.write(buffers[0]);
169             else if (buffers.length>1 && _channel instanceof GatheringByteChannel)
170                 flushed= (int)((GatheringByteChannel)_channel).write(buffers,0,buffers.length);
171             else
172             {
173                 for (ByteBuffer b : buffers)
174                 {
175                     if (b.hasRemaining())
176                     {
177                         int l=_channel.write(b);
178                         if (l>0)
179                             flushed+=l;
180                         if (b.hasRemaining())
181                             break;
182                     }
183                 }
184             }
185             LOG.debug("flushed {} {}", flushed, this);
186         }
187         catch (IOException e)
188         {
189             throw new EofException(e);
190         }
191         
192         boolean all_flushed=true;
193         if (flushed>0)
194         {
195             notIdle();
196 
197             // clear empty buffers to prevent position creeping up the buffer
198             for (ByteBuffer b : buffers)
199             {
200                 if (BufferUtil.isEmpty(b))
201                     BufferUtil.clear(b);
202                 else
203                     all_flushed=false;
204             }
205         }
206         
207         return all_flushed;
208     }
209 
210     public ByteChannel getChannel()
211     {
212         return _channel;
213     }
214 
215     @Override
216     public Object getTransport()
217     {
218         return _channel;
219     }
220 
221     @Override
222     public Socket getSocket()
223     {
224         return _socket;
225     }
226 
227     @Override
228     protected void onIncompleteFlush()
229     {
230         throw new UnsupportedOperationException();
231     }
232 
233     @Override
234     protected boolean needsFill() throws IOException
235     {
236         throw new UnsupportedOperationException();
237     }
238 }