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.client.http;
20  
21  import java.nio.channels.AsynchronousCloseException;
22  import java.util.concurrent.TimeoutException;
23  import java.util.concurrent.atomic.AtomicBoolean;
24  import java.util.concurrent.atomic.AtomicInteger;
25  
26  import org.eclipse.jetty.client.HttpConnection;
27  import org.eclipse.jetty.client.HttpDestination;
28  import org.eclipse.jetty.client.HttpExchange;
29  import org.eclipse.jetty.client.api.Connection;
30  import org.eclipse.jetty.client.api.Request;
31  import org.eclipse.jetty.client.api.Response;
32  import org.eclipse.jetty.io.AbstractConnection;
33  import org.eclipse.jetty.io.EndPoint;
34  import org.eclipse.jetty.util.Promise;
35  import org.eclipse.jetty.util.log.Log;
36  import org.eclipse.jetty.util.log.Logger;
37  import org.eclipse.jetty.util.thread.Sweeper;
38  
39  public class HttpConnectionOverHTTP extends AbstractConnection implements Connection, Sweeper.Sweepable
40  {
41      private static final Logger LOG = Log.getLogger(HttpConnectionOverHTTP.class);
42  
43      private final AtomicBoolean closed = new AtomicBoolean();
44      private final AtomicInteger sweeps = new AtomicInteger();
45      private final Promise<Connection> promise;
46      private final Delegate delegate;
47      private final HttpChannelOverHTTP channel;
48      private long idleTimeout;
49  
50      /**
51       * @deprecated use {@link #HttpConnectionOverHTTP(EndPoint, HttpDestination, Promise)} instead
52       */
53      @Deprecated
54      public HttpConnectionOverHTTP(EndPoint endPoint, HttpDestination destination)
55      {
56          this(endPoint, destination, new Promise.Adapter<Connection>());
57          throw new UnsupportedOperationException("Deprecated, use HttpConnectionOverHTTP(EndPoint, HttpDestination, Promise<Connection>) instead");
58      }
59  
60      public HttpConnectionOverHTTP(EndPoint endPoint, HttpDestination destination, Promise<Connection> promise)
61      {
62          super(endPoint, destination.getHttpClient().getExecutor(), destination.getHttpClient().isDispatchIO());
63          this.promise = promise;
64          this.delegate = new Delegate(destination);
65          this.channel = newHttpChannel();
66      }
67  
68      protected HttpChannelOverHTTP newHttpChannel()
69      {
70          return new HttpChannelOverHTTP(this);
71      }
72  
73      public HttpChannelOverHTTP getHttpChannel()
74      {
75          return channel;
76      }
77  
78      public HttpDestinationOverHTTP getHttpDestination()
79      {
80          return (HttpDestinationOverHTTP)delegate.getHttpDestination();
81      }
82  
83      @Override
84      public void send(Request request, Response.CompleteListener listener)
85      {
86          delegate.send(request, listener);
87      }
88  
89      protected void send(HttpExchange exchange)
90      {
91          delegate.send(exchange);
92      }
93  
94      @Override
95      public void onOpen()
96      {
97          super.onOpen();
98          fillInterested();
99          promise.succeeded(this);
100     }
101 
102     public boolean isClosed()
103     {
104         return closed.get();
105     }
106 
107     @Override
108     protected boolean onReadTimeout()
109     {
110         if (LOG.isDebugEnabled())
111             LOG.debug("{} idle timeout", this);
112         close(new TimeoutException());
113         return false;
114     }
115 
116     @Override
117     public void onFillable()
118     {
119         HttpExchange exchange = channel.getHttpExchange();
120         if (exchange != null)
121         {
122             channel.receive();
123         }
124         else
125         {
126             // If there is no exchange, then could be either a remote close,
127             // or garbage bytes; in both cases we close the connection
128             close();
129         }
130     }
131 
132     public void release()
133     {
134         // Restore idle timeout
135         getEndPoint().setIdleTimeout(idleTimeout);
136         getHttpDestination().release(this);
137     }
138 
139     @Override
140     public void close()
141     {
142         close(new AsynchronousCloseException());
143     }
144 
145     protected void close(Throwable failure)
146     {
147         if (softClose())
148         {
149             // First close then abort, to be sure that the connection cannot be reused
150             // from an onFailure() handler or by blocking code waiting for completion.
151             getHttpDestination().close(this);
152             getEndPoint().shutdownOutput();
153             if (LOG.isDebugEnabled())
154                 LOG.debug("{} oshut", this);
155             getEndPoint().close();
156             if (LOG.isDebugEnabled())
157                 LOG.debug("{} closed", this);
158 
159             abort(failure);
160         }
161     }
162 
163     public boolean softClose()
164     {
165         return closed.compareAndSet(false, true);
166     }
167 
168     protected boolean abort(Throwable failure)
169     {
170         HttpExchange exchange = channel.getHttpExchange();
171         return exchange != null && exchange.getRequest().abort(failure);
172     }
173 
174     @Override
175     public boolean sweep()
176     {
177         if (!closed.get())
178             return false;
179 
180         if (sweeps.incrementAndGet() < 4)
181             return false;
182 
183         return true;
184     }
185 
186     @Override
187     public String toString()
188     {
189         return String.format("%s@%h(l:%s <-> r:%s,closed=%b)[%s]",
190                 getClass().getSimpleName(),
191                 this,
192                 getEndPoint().getLocalAddress(),
193                 getEndPoint().getRemoteAddress(),
194                 closed.get(),
195                 channel);
196     }
197 
198     private class Delegate extends HttpConnection
199     {
200         private Delegate(HttpDestination destination)
201         {
202             super(destination);
203         }
204 
205         @Override
206         protected void send(HttpExchange exchange)
207         {
208             Request request = exchange.getRequest();
209             normalizeRequest(request);
210 
211             // Save the old idle timeout to restore it
212             EndPoint endPoint = getEndPoint();
213             idleTimeout = endPoint.getIdleTimeout();
214             endPoint.setIdleTimeout(request.getIdleTimeout());
215 
216             // One channel per connection, just delegate the send
217             if (channel.associate(exchange))
218                 channel.send();
219             else
220                 channel.release();
221         }
222 
223         @Override
224         public void close()
225         {
226             HttpConnectionOverHTTP.this.close();
227         }
228 
229         @Override
230         public String toString()
231         {
232             return HttpConnectionOverHTTP.this.toString();
233         }
234     }
235 }