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