1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
37
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())
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 }