1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
82
83
84
85
86
87
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
116
117
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
169
170
171
172
173
174 getHttpClient().removeDestination(this);
175 }
176 }
177 else
178 {
179
180
181
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 }