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.Closeable;
22 import java.io.IOException;
23 import java.nio.channels.AsynchronousCloseException;
24 import java.util.ArrayList;
25 import java.util.List;
26 import java.util.Queue;
27 import java.util.concurrent.RejectedExecutionException;
28
29 import org.eclipse.jetty.client.api.Connection;
30 import org.eclipse.jetty.client.api.Destination;
31 import org.eclipse.jetty.client.api.Response;
32 import org.eclipse.jetty.http.HttpField;
33 import org.eclipse.jetty.http.HttpHeader;
34 import org.eclipse.jetty.http.HttpScheme;
35 import org.eclipse.jetty.io.ClientConnectionFactory;
36 import org.eclipse.jetty.io.ssl.SslClientConnectionFactory;
37 import org.eclipse.jetty.util.BlockingArrayQueue;
38 import org.eclipse.jetty.util.HostPort;
39 import org.eclipse.jetty.util.Promise;
40 import org.eclipse.jetty.util.annotation.ManagedAttribute;
41 import org.eclipse.jetty.util.annotation.ManagedObject;
42 import org.eclipse.jetty.util.component.ContainerLifeCycle;
43 import org.eclipse.jetty.util.component.Dumpable;
44 import org.eclipse.jetty.util.log.Log;
45 import org.eclipse.jetty.util.log.Logger;
46
47 @ManagedObject
48 public abstract class HttpDestination extends ContainerLifeCycle implements Destination, Closeable, Dumpable
49 {
50 protected static final Logger LOG = Log.getLogger(HttpDestination.class);
51
52 private final HttpClient client;
53 private final Origin origin;
54 private final Queue<HttpExchange> exchanges;
55 private final RequestNotifier requestNotifier;
56 private final ResponseNotifier responseNotifier;
57 private final ProxyConfiguration.Proxy proxy;
58 private final ClientConnectionFactory connectionFactory;
59 private final HttpField hostField;
60
61 public HttpDestination(HttpClient client, Origin origin)
62 {
63 this.client = client;
64 this.origin = origin;
65
66 this.exchanges = newExchangeQueue(client);
67
68 this.requestNotifier = new RequestNotifier(client);
69 this.responseNotifier = new ResponseNotifier();
70
71 ProxyConfiguration proxyConfig = client.getProxyConfiguration();
72 proxy = proxyConfig.match(origin);
73 ClientConnectionFactory connectionFactory = client.getTransport();
74 if (proxy != null)
75 {
76 connectionFactory = proxy.newClientConnectionFactory(connectionFactory);
77 if (proxy.isSecure())
78 connectionFactory = newSslClientConnectionFactory(connectionFactory);
79 }
80 else
81 {
82 if (HttpScheme.HTTPS.is(getScheme()))
83 connectionFactory = newSslClientConnectionFactory(connectionFactory);
84 }
85 this.connectionFactory = connectionFactory;
86
87 String host = HostPort.normalizeHost(getHost());
88 if (!client.isDefaultPort(getScheme(), getPort()))
89 host += ":" + getPort();
90 hostField = new HttpField(HttpHeader.HOST, host);
91 }
92
93 protected Queue<HttpExchange> newExchangeQueue(HttpClient client)
94 {
95 return new BlockingArrayQueue<>(client.getMaxRequestsQueuedPerDestination());
96 }
97
98 protected ClientConnectionFactory newSslClientConnectionFactory(ClientConnectionFactory connectionFactory)
99 {
100 return new SslClientConnectionFactory(client.getSslContextFactory(), client.getByteBufferPool(), client.getExecutor(), connectionFactory);
101 }
102
103 public HttpClient getHttpClient()
104 {
105 return client;
106 }
107
108 public Origin getOrigin()
109 {
110 return origin;
111 }
112
113 public Queue<HttpExchange> getHttpExchanges()
114 {
115 return exchanges;
116 }
117
118 public RequestNotifier getRequestNotifier()
119 {
120 return requestNotifier;
121 }
122
123 public ResponseNotifier getResponseNotifier()
124 {
125 return responseNotifier;
126 }
127
128 public ProxyConfiguration.Proxy getProxy()
129 {
130 return proxy;
131 }
132
133 public ClientConnectionFactory getClientConnectionFactory()
134 {
135 return connectionFactory;
136 }
137
138 @Override
139 @ManagedAttribute(value = "The destination scheme", readonly = true)
140 public String getScheme()
141 {
142 return origin.getScheme();
143 }
144
145 @Override
146 @ManagedAttribute(value = "The destination host", readonly = true)
147 public String getHost()
148 {
149
150
151 return origin.getAddress().getHost();
152 }
153
154 @Override
155 @ManagedAttribute(value = "The destination port", readonly = true)
156 public int getPort()
157 {
158 return origin.getAddress().getPort();
159 }
160
161 @ManagedAttribute(value = "The number of queued requests", readonly = true)
162 public int getQueuedRequestCount()
163 {
164 return exchanges.size();
165 }
166
167 public Origin.Address getConnectAddress()
168 {
169 return proxy == null ? origin.getAddress() : proxy.getAddress();
170 }
171
172 public HttpField getHostField()
173 {
174 return hostField;
175 }
176
177 protected void send(HttpRequest request, List<Response.ResponseListener> listeners)
178 {
179 if (!getScheme().equalsIgnoreCase(request.getScheme()))
180 throw new IllegalArgumentException("Invalid request scheme " + request.getScheme() + " for destination " + this);
181 if (!getHost().equalsIgnoreCase(request.getHost()))
182 throw new IllegalArgumentException("Invalid request host " + request.getHost() + " for destination " + this);
183 int port = request.getPort();
184 if (port >= 0 && getPort() != port)
185 throw new IllegalArgumentException("Invalid request port " + port + " for destination " + this);
186
187 HttpExchange exchange = new HttpExchange(this, request, listeners);
188
189 if (client.isRunning())
190 {
191 if (enqueue(exchanges, exchange))
192 {
193 if (!client.isRunning() && exchanges.remove(exchange))
194 {
195 request.abort(new RejectedExecutionException(client + " is stopping"));
196 }
197 else
198 {
199 if (LOG.isDebugEnabled())
200 LOG.debug("Queued {} for {}", request, this);
201 requestNotifier.notifyQueued(request);
202 send();
203 }
204 }
205 else
206 {
207 if (LOG.isDebugEnabled())
208 LOG.debug("Max queue size {} exceeded by {} for {}", client.getMaxRequestsQueuedPerDestination(), request, this);
209 request.abort(new RejectedExecutionException("Max requests per destination " + client.getMaxRequestsQueuedPerDestination() + " exceeded for " + this));
210 }
211 }
212 else
213 {
214 request.abort(new RejectedExecutionException(client + " is stopped"));
215 }
216 }
217
218 protected boolean enqueue(Queue<HttpExchange> queue, HttpExchange exchange)
219 {
220 return queue.offer(exchange);
221 }
222
223 public abstract void send();
224
225 public void newConnection(Promise<Connection> promise)
226 {
227 createConnection(promise);
228 }
229
230 protected void createConnection(Promise<Connection> promise)
231 {
232 client.newConnection(this, promise);
233 }
234
235 public boolean remove(HttpExchange exchange)
236 {
237 return exchanges.remove(exchange);
238 }
239
240 public void close()
241 {
242 abort(new AsynchronousCloseException());
243 if (LOG.isDebugEnabled())
244 LOG.debug("Closed {}", this);
245 }
246
247 public void release(Connection connection)
248 {
249 }
250
251 public void close(Connection connection)
252 {
253 }
254
255
256
257
258
259
260 public void abort(Throwable cause)
261 {
262
263
264
265
266 for (HttpExchange exchange : new ArrayList<>(exchanges))
267 exchange.getRequest().abort(cause);
268 }
269
270 @Override
271 public String dump()
272 {
273 return ContainerLifeCycle.dump(this);
274 }
275
276 @Override
277 public void dump(Appendable out, String indent) throws IOException
278 {
279 ContainerLifeCycle.dumpObject(out, toString());
280 }
281
282 public String asString()
283 {
284 return origin.asString();
285 }
286
287 @Override
288 public String toString()
289 {
290 return String.format("%s[%s]@%x%s,queue=%d",
291 HttpDestination.class.getSimpleName(),
292 asString(),
293 hashCode(),
294 proxy == null ? "" : "(via " + proxy + ")",
295 exchanges.size());
296 }
297 }