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.util.List;
22 import java.util.concurrent.CopyOnWriteArrayList;
23 import java.util.concurrent.Executor;
24 import java.util.concurrent.TimeoutException;
25 import java.util.concurrent.atomic.AtomicReference;
26
27 import org.eclipse.jetty.util.Callback;
28 import org.eclipse.jetty.util.ExecutorCallback;
29 import org.eclipse.jetty.util.log.Log;
30 import org.eclipse.jetty.util.log.Logger;
31
32
33
34
35
36
37
38
39 public abstract class AbstractConnection implements Connection
40 {
41 private static final Logger LOG = Log.getLogger(AbstractConnection.class);
42
43 private final List<Listener> listeners = new CopyOnWriteArrayList<>();
44 private final AtomicReference<State> _state = new AtomicReference<>(State.IDLE);
45 private final long _created=System.currentTimeMillis();
46 private final EndPoint _endPoint;
47 private final Executor _executor;
48 private final Callback<Void> _readCallback;
49 private int _inputBufferSize=8192;
50
51 public AbstractConnection(EndPoint endp, Executor executor)
52 {
53 this(endp, executor, true);
54 }
55
56 public AbstractConnection(EndPoint endp, Executor executor, final boolean dispatchCompletion)
57 {
58 if (executor == null)
59 throw new IllegalArgumentException("Executor must not be null!");
60 _endPoint = endp;
61 _executor = executor;
62 _readCallback = new ExecutorCallback<Void>(executor)
63 {
64 @Override
65 protected void onCompleted(Void context)
66 {
67 if (_state.compareAndSet(State.INTERESTED,State.FILLING))
68 {
69 try
70 {
71 onFillable();
72 }
73 finally
74 {
75 loop:while(true)
76 {
77 switch(_state.get())
78 {
79 case IDLE:
80 case INTERESTED:
81 throw new IllegalStateException();
82
83 case FILLING:
84 if (_state.compareAndSet(State.FILLING,State.IDLE))
85 break loop;
86 break;
87
88 case FILLING_INTERESTED:
89 if (_state.compareAndSet(State.FILLING_INTERESTED,State.INTERESTED))
90 {
91 getEndPoint().fillInterested(null, _readCallback);
92 break loop;
93 }
94 break;
95 }
96 }
97
98 }
99 }
100 else
101 LOG.warn(new Throwable());
102 }
103
104 @Override
105 protected void onFailed(Void context, Throwable x)
106 {
107 onFillInterestedFailed(x);
108 }
109
110 @Override
111 protected boolean alwaysDispatchCompletion()
112 {
113 return dispatchCompletion;
114 }
115
116 @Override
117 public String toString()
118 {
119 return String.format("AC.ReadCB@%x", AbstractConnection.this.hashCode());
120 }
121 };
122 }
123
124 @Override
125 public void addListener(Listener listener)
126 {
127 listeners.add(listener);
128 }
129
130 public int getInputBufferSize()
131 {
132 return _inputBufferSize;
133 }
134
135 public void setInputBufferSize(int inputBufferSize)
136 {
137 _inputBufferSize = inputBufferSize;
138 }
139
140 public Executor getExecutor()
141 {
142 return _executor;
143 }
144
145
146
147
148
149
150
151 public void fillInterested()
152 {
153 LOG.debug("fillInterested {}",this);
154
155 loop:while(true)
156 {
157 switch(_state.get())
158 {
159 case IDLE:
160 if (_state.compareAndSet(State.IDLE,State.INTERESTED))
161 {
162 getEndPoint().fillInterested(null, _readCallback);
163 break loop;
164 }
165 break;
166
167 case FILLING:
168 if (_state.compareAndSet(State.FILLING,State.FILLING_INTERESTED))
169 break loop;
170 break;
171
172 case FILLING_INTERESTED:
173 case INTERESTED:
174 break loop;
175 }
176 }
177 }
178
179
180
181
182
183 public abstract void onFillable();
184
185
186
187
188
189 public void onFillInterestedFailed(Throwable cause)
190 {
191 LOG.debug("{} onFillInterestedFailed {}", this, cause);
192 if (_endPoint.isOpen())
193 {
194 boolean close = true;
195 if (cause instanceof TimeoutException)
196 close = onReadTimeout();
197 if (close)
198 {
199 if (_endPoint.isOutputShutdown())
200 _endPoint.close();
201 else
202 _endPoint.shutdownOutput();
203 }
204 }
205 }
206
207
208
209
210
211 protected boolean onReadTimeout()
212 {
213 return true;
214 }
215
216
217 final AtomicReference<Throwable> _opened = new AtomicReference<>(null);
218 @Override
219 public void onOpen()
220 {
221 LOG.debug("onOpen {}", this);
222
223 for (Listener listener : listeners)
224 listener.onOpened(this);
225
226 if (!_opened.compareAndSet(null,new Throwable()))
227 {
228 LOG.warn("ALREADY OPENED ", _opened.get());
229 LOG.warn("EXTRA OPEN AT ",new Throwable());
230 }
231 }
232
233 @Override
234 public void onClose()
235 {
236 LOG.debug("onClose {}",this);
237
238 for (Listener listener : listeners)
239 listener.onClosed(this);
240 }
241
242 @Override
243 public EndPoint getEndPoint()
244 {
245 return _endPoint;
246 }
247
248 @Override
249 public void close()
250 {
251 getEndPoint().close();
252 }
253
254 @Override
255 public int getMessagesIn()
256 {
257 return 0;
258 }
259
260 @Override
261 public int getMessagesOut()
262 {
263 return 0;
264 }
265
266 @Override
267 public long getBytesIn()
268 {
269
270 return 0;
271 }
272
273 @Override
274 public long getBytesOut()
275 {
276
277 return 0;
278 }
279
280 @Override
281 public long getCreatedTimeStamp()
282 {
283 return _created;
284 }
285
286 @Override
287 public String toString()
288 {
289 return String.format("%s@%x{%s}", getClass().getSimpleName(), hashCode(), _state.get());
290 }
291
292 private enum State
293 {
294 IDLE, INTERESTED, FILLING, FILLING_INTERESTED
295 }
296 }