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.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   * <p>A convenience base implementation of {@link Connection}.</p>
34   * <p>This class uses the capabilities of the {@link EndPoint} API to provide a
35   * more traditional style of async reading.  A call to {@link #fillInterested()}
36   * will schedule a callback to {@link #onFillable()} or {@link #onFillInterestedFailed(Throwable)}
37   * as appropriate.</p>
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      * <p>Utility method to be called to register read interest.</p>
147      * <p>After a call to this method, {@link #onFillable()} or {@link #onFillInterestedFailed(Throwable)}
148      * will be called back as appropriate.</p>
149      * @see #onFillable()
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      * <p>Callback method invoked when the endpoint is ready to be read.</p>
181      * @see #fillInterested()
182      */
183     public abstract void onFillable();
184 
185     /**
186      * <p>Callback method invoked when the endpoint failed to be ready to be read.</p>
187      * @param cause the exception that caused the failure
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      * <p>Callback method invoked when the endpoint failed to be ready to be read after a timeout</p>
209      * @return true to signal that the endpoint must be closed, false to keep the endpoint open
210      */
211     protected boolean onReadTimeout()
212     {
213         return true;
214     }
215 
216     // TODO remove this when open/close refactored
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         // TODO Auto-generated method stub
270         return 0;
271     }
272 
273     @Override
274     public long getBytesOut()
275     {
276         // TODO Auto-generated method stub
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 }