View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2015 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;
20  
21  import java.io.IOException;
22  import java.util.Arrays;
23  
24  import org.eclipse.jetty.client.api.Connection;
25  import org.eclipse.jetty.client.api.Request;
26  import org.eclipse.jetty.util.Callback;
27  import org.eclipse.jetty.util.component.ContainerLifeCycle;
28  import org.eclipse.jetty.util.thread.Sweeper;
29  
30  public abstract class PoolingHttpDestination<C extends Connection> extends HttpDestination implements Callback
31  {
32      private final ConnectionPool connectionPool;
33  
34      public PoolingHttpDestination(HttpClient client, Origin origin)
35      {
36          super(client, origin);
37          this.connectionPool = newConnectionPool(client);
38          Sweeper sweeper = client.getBean(Sweeper.class);
39          if (sweeper != null)
40              sweeper.offer(connectionPool);
41      }
42  
43      protected ConnectionPool newConnectionPool(HttpClient client)
44      {
45          return new ConnectionPool(this, client.getMaxConnectionsPerDestination(), this);
46      }
47  
48      public ConnectionPool getConnectionPool()
49      {
50          return connectionPool;
51      }
52  
53      @Override
54      public void succeeded()
55      {
56          send();
57      }
58  
59      @Override
60      public void failed(final Throwable x)
61      {
62          abort(x);
63      }
64  
65      public void send()
66      {
67          if (getHttpExchanges().isEmpty())
68              return;
69          C connection = acquire();
70          if (connection != null)
71              process(connection);
72      }
73  
74      @SuppressWarnings("unchecked")
75      public C acquire()
76      {
77          return (C)connectionPool.acquire();
78      }
79  
80      /**
81       * <p>Processes a new connection making it idle or active depending on whether requests are waiting to be sent.</p>
82       * <p>A new connection is created when a request needs to be executed; it is possible that the request that
83       * triggered the request creation is executed by another connection that was just released, so the new connection
84       * may become idle.</p>
85       * <p>If a request is waiting to be executed, it will be dequeued and executed by the new connection.</p>
86       *
87       * @param connection the new connection
88       */
89      public void process(final C connection)
90      {
91          HttpClient client = getHttpClient();
92          final HttpExchange exchange = getHttpExchanges().poll();
93          if (LOG.isDebugEnabled())
94              LOG.debug("Processing exchange {} on {} of {}", exchange, connection, this);
95          if (exchange == null)
96          {
97              if (!connectionPool.release(connection))
98                  connection.close();
99  
100             if (!client.isRunning())
101             {
102                 if (LOG.isDebugEnabled())
103                     LOG.debug("{} is stopping", client);
104                 connection.close();
105             }
106         }
107         else
108         {
109             final Request request = exchange.getRequest();
110             Throwable cause = request.getAbortCause();
111             if (cause != null)
112             {
113                 if (LOG.isDebugEnabled())
114                     LOG.debug("Aborted before processing {}: {}", exchange, cause);
115                 // It may happen that the request is aborted before the exchange
116                 // is created. Aborting the exchange a second time will result in
117                 // a no-operation, so we just abort here to cover that edge case.
118                 exchange.abort(cause);
119             }
120             else
121             {
122                 send(connection, exchange);
123             }
124         }
125     }
126 
127     protected abstract void send(C connection, HttpExchange exchange);
128 
129     @Override
130     public void release(Connection c)
131     {
132         @SuppressWarnings("unchecked")
133         C connection = (C)c;
134         if (LOG.isDebugEnabled())
135             LOG.debug("{} released", connection);
136         HttpClient client = getHttpClient();
137         if (client.isRunning())
138         {
139             if (connectionPool.isActive(connection))
140             {
141                 process(connection);
142             }
143             else
144             {
145                 if (LOG.isDebugEnabled())
146                     LOG.debug("{} explicit", connection);
147             }
148         }
149         else
150         {
151             if (LOG.isDebugEnabled())
152                 LOG.debug("{} is stopped", client);
153             connection.close();
154         }
155     }
156 
157     @Override
158     public void close(Connection oldConnection)
159     {
160         super.close(oldConnection);
161 
162         connectionPool.remove(oldConnection);
163 
164         if (getHttpExchanges().isEmpty())
165         {
166             if (getHttpClient().isRemoveIdleDestinations() && connectionPool.isEmpty())
167             {
168                 // There is a race condition between this thread removing the destination
169                 // and another thread queueing a request to this same destination.
170                 // If this destination is removed, but the request queued, a new connection
171                 // will be opened, the exchange will be executed and eventually the connection
172                 // will idle timeout and be closed. Meanwhile a new destination will be created
173                 // in HttpClient and will be used for other requests.
174                 getHttpClient().removeDestination(this);
175             }
176         }
177         else
178         {
179             // We need to execute queued requests even if this connection failed.
180             // We may create a connection that is not needed, but it will eventually
181             // idle timeout, so no worries.
182             C newConnection = acquire();
183             if (newConnection != null)
184                 process(newConnection);
185         }
186     }
187 
188     public void close()
189     {
190         super.close();
191         connectionPool.close();
192     }
193 
194     @Override
195     public void dump(Appendable out, String indent) throws IOException
196     {
197         super.dump(out, indent);
198         ContainerLifeCycle.dump(out, indent, Arrays.asList(connectionPool));
199     }
200 
201     @Override
202     public String toString()
203     {
204         return String.format("%s,pool=%s", super.toString(), connectionPool);
205     }
206 }