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