View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2016 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.IOException;
22  import java.net.InetSocketAddress;
23  import java.net.Socket;
24  import java.nio.ByteBuffer;
25  import java.nio.channels.ByteChannel;
26  import java.nio.channels.SocketChannel;
27  import java.util.Arrays;
28  import java.util.stream.Collectors;
29  
30  import org.eclipse.jetty.util.BufferUtil;
31  import org.eclipse.jetty.util.log.Log;
32  import org.eclipse.jetty.util.log.Logger;
33  import org.eclipse.jetty.util.thread.Scheduler;
34  
35  /**
36   * Channel End Point.
37   * <p>Holds the channel and socket for an NIO endpoint.
38   */
39  public class ChannelEndPoint extends AbstractEndPoint
40  {
41      private static final Logger LOG = Log.getLogger(ChannelEndPoint.class);
42  
43      private final SocketChannel _channel;
44      private final Socket _socket;
45      private volatile boolean _ishut;
46      private volatile boolean _oshut;
47  
48      public ChannelEndPoint(Scheduler scheduler,SocketChannel channel)
49      {
50          super(scheduler,
51              (InetSocketAddress)channel.socket().getLocalSocketAddress(),
52              (InetSocketAddress)channel.socket().getRemoteSocketAddress());
53          _channel=channel;
54          _socket=channel.socket();
55      }
56  
57      @Override
58      public boolean isOptimizedForDirectBuffers()
59      {
60          return true;
61      }
62  
63      @Override
64      public boolean isOpen()
65      {
66          return _channel.isOpen();
67      }
68  
69      protected void shutdownInput()
70      {
71          if (LOG.isDebugEnabled())
72              LOG.debug("ishut {}", this);
73          _ishut=true;
74          if (_oshut)
75              close();
76      }
77  
78      @Override
79      public void shutdownOutput()
80      {
81          if (LOG.isDebugEnabled())
82              LOG.debug("oshut {}", this);
83          _oshut = true;
84          if (_channel.isOpen())
85          {
86              try
87              {
88                  if (!_socket.isOutputShutdown())
89                      _socket.shutdownOutput();
90              }
91              catch (IOException e)
92              {
93                  LOG.debug(e);
94              }
95              finally
96              {
97                  if (_ishut)
98                  {
99                      close();
100                 }
101             }
102         }
103     }
104 
105     @Override
106     public boolean isOutputShutdown()
107     {
108         return _oshut || !_channel.isOpen() || _socket.isOutputShutdown();
109     }
110 
111     @Override
112     public boolean isInputShutdown()
113     {
114         return _ishut || !_channel.isOpen() || _socket.isInputShutdown();
115     }
116 
117     @Override
118     public void close()
119     {
120         super.close();
121         if (LOG.isDebugEnabled())
122             LOG.debug("close {}", this);
123         try
124         {
125             _channel.close();
126         }
127         catch (IOException e)
128         {
129             LOG.debug(e);
130         }
131         finally
132         {
133             _ishut=true;
134             _oshut=true;
135         }
136     }
137 
138     @Override
139     public int fill(ByteBuffer buffer) throws IOException
140     {
141         if (_ishut)
142             return -1;
143 
144         int pos=BufferUtil.flipToFill(buffer);
145         try
146         {
147             int filled = _channel.read(buffer);
148             if (LOG.isDebugEnabled()) // Avoid boxing of variable 'filled'
149                 LOG.debug("filled {} {}", filled, this);
150 
151             if (filled>0)
152                 notIdle();
153             else if (filled==-1)
154                 shutdownInput();
155 
156             return filled;
157         }
158         catch(IOException e)
159         {
160             LOG.debug(e);
161             shutdownInput();
162             return -1;
163         }
164         finally
165         {
166             BufferUtil.flipToFlush(buffer,pos);
167         }
168     }
169 
170     @Override
171     public boolean flush(ByteBuffer... buffers) throws IOException
172     {
173         long flushed=0;
174         try
175         {
176             if (buffers.length==1)
177                 flushed=_channel.write(buffers[0]);
178             else if (buffers.length>1)
179                 flushed=_channel.write(buffers,0,buffers.length);
180             else
181             {
182                 for (ByteBuffer b : buffers)
183                 {
184                     if (b.hasRemaining())
185                     {
186                         int l=_channel.write(b);
187                         if (l>0)
188                             flushed+=l;
189                         if (b.hasRemaining())
190                             break;
191                     }
192                 }
193             }
194             if (LOG.isDebugEnabled())
195                 LOG.debug("flushed {} {}", flushed, this);
196         }
197         catch (IOException e)
198         {
199             throw new EofException(e);
200         }
201 
202         if (flushed>0)
203             notIdle();
204 
205         for (ByteBuffer b : buffers)
206             if (!BufferUtil.isEmpty(b))
207                 return false;
208 
209         return true;
210     }
211 
212     public ByteChannel getChannel()
213     {
214         return _channel;
215     }
216 
217     @Override
218     public Object getTransport()
219     {
220         return _channel;
221     }
222 
223     public Socket getSocket()
224     {
225         return _socket;
226     }
227 
228     @Override
229     protected void onIncompleteFlush()
230     {
231         throw new UnsupportedOperationException();
232     }
233 
234     @Override
235     protected void needsFillInterest() throws IOException
236     {
237         throw new UnsupportedOperationException();
238     }
239 }