1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.client.http;
20
21 import java.nio.channels.AsynchronousCloseException;
22 import java.util.concurrent.TimeoutException;
23 import java.util.concurrent.atomic.AtomicBoolean;
24 import java.util.concurrent.atomic.AtomicInteger;
25
26 import org.eclipse.jetty.client.HttpConnection;
27 import org.eclipse.jetty.client.HttpDestination;
28 import org.eclipse.jetty.client.HttpExchange;
29 import org.eclipse.jetty.client.api.Connection;
30 import org.eclipse.jetty.client.api.Request;
31 import org.eclipse.jetty.client.api.Response;
32 import org.eclipse.jetty.io.AbstractConnection;
33 import org.eclipse.jetty.io.EndPoint;
34 import org.eclipse.jetty.util.Promise;
35 import org.eclipse.jetty.util.log.Log;
36 import org.eclipse.jetty.util.log.Logger;
37 import org.eclipse.jetty.util.thread.Sweeper;
38
39 public class HttpConnectionOverHTTP extends AbstractConnection implements Connection, Sweeper.Sweepable
40 {
41 private static final Logger LOG = Log.getLogger(HttpConnectionOverHTTP.class);
42
43 private final AtomicBoolean closed = new AtomicBoolean();
44 private final AtomicInteger sweeps = new AtomicInteger();
45 private final Promise<Connection> promise;
46 private final Delegate delegate;
47 private final HttpChannelOverHTTP channel;
48 private long idleTimeout;
49
50
51
52
53 @Deprecated
54 public HttpConnectionOverHTTP(EndPoint endPoint, HttpDestination destination)
55 {
56 this(endPoint, destination, new Promise.Adapter<Connection>());
57 throw new UnsupportedOperationException("Deprecated, use HttpConnectionOverHTTP(EndPoint, HttpDestination, Promise<Connection>) instead");
58 }
59
60 public HttpConnectionOverHTTP(EndPoint endPoint, HttpDestination destination, Promise<Connection> promise)
61 {
62 super(endPoint, destination.getHttpClient().getExecutor(), destination.getHttpClient().isDispatchIO());
63 this.promise = promise;
64 this.delegate = new Delegate(destination);
65 this.channel = newHttpChannel();
66 }
67
68 protected HttpChannelOverHTTP newHttpChannel()
69 {
70 return new HttpChannelOverHTTP(this);
71 }
72
73 public HttpChannelOverHTTP getHttpChannel()
74 {
75 return channel;
76 }
77
78 public HttpDestinationOverHTTP getHttpDestination()
79 {
80 return (HttpDestinationOverHTTP)delegate.getHttpDestination();
81 }
82
83 @Override
84 public void send(Request request, Response.CompleteListener listener)
85 {
86 delegate.send(request, listener);
87 }
88
89 protected void send(HttpExchange exchange)
90 {
91 delegate.send(exchange);
92 }
93
94 @Override
95 public void onOpen()
96 {
97 super.onOpen();
98 fillInterested();
99 promise.succeeded(this);
100 }
101
102 public boolean isClosed()
103 {
104 return closed.get();
105 }
106
107 @Override
108 protected boolean onReadTimeout()
109 {
110 if (LOG.isDebugEnabled())
111 LOG.debug("{} idle timeout", this);
112 close(new TimeoutException());
113 return false;
114 }
115
116 @Override
117 public void onFillable()
118 {
119 HttpExchange exchange = channel.getHttpExchange();
120 if (exchange != null)
121 {
122 channel.receive();
123 }
124 else
125 {
126
127
128 close();
129 }
130 }
131
132 public void release()
133 {
134
135 getEndPoint().setIdleTimeout(idleTimeout);
136 getHttpDestination().release(this);
137 }
138
139 @Override
140 public void close()
141 {
142 close(new AsynchronousCloseException());
143 }
144
145 protected void close(Throwable failure)
146 {
147 if (softClose())
148 {
149
150
151 getHttpDestination().close(this);
152 getEndPoint().shutdownOutput();
153 if (LOG.isDebugEnabled())
154 LOG.debug("{} oshut", this);
155 getEndPoint().close();
156 if (LOG.isDebugEnabled())
157 LOG.debug("{} closed", this);
158
159 abort(failure);
160 }
161 }
162
163 public boolean softClose()
164 {
165 return closed.compareAndSet(false, true);
166 }
167
168 protected boolean abort(Throwable failure)
169 {
170 HttpExchange exchange = channel.getHttpExchange();
171 return exchange != null && exchange.getRequest().abort(failure);
172 }
173
174 @Override
175 public boolean sweep()
176 {
177 if (!closed.get())
178 return false;
179
180 if (sweeps.incrementAndGet() < 4)
181 return false;
182
183 return true;
184 }
185
186 @Override
187 public String toString()
188 {
189 return String.format("%s@%h(l:%s <-> r:%s,closed=%b)[%s]",
190 getClass().getSimpleName(),
191 this,
192 getEndPoint().getLocalAddress(),
193 getEndPoint().getRemoteAddress(),
194 closed.get(),
195 channel);
196 }
197
198 private class Delegate extends HttpConnection
199 {
200 private Delegate(HttpDestination destination)
201 {
202 super(destination);
203 }
204
205 @Override
206 protected void send(HttpExchange exchange)
207 {
208 Request request = exchange.getRequest();
209 normalizeRequest(request);
210
211
212 EndPoint endPoint = getEndPoint();
213 idleTimeout = endPoint.getIdleTimeout();
214 endPoint.setIdleTimeout(request.getIdleTimeout());
215
216
217 if (channel.associate(exchange))
218 channel.send();
219 else
220 channel.release();
221 }
222
223 @Override
224 public void close()
225 {
226 HttpConnectionOverHTTP.this.close();
227 }
228
229 @Override
230 public String toString()
231 {
232 return HttpConnectionOverHTTP.this.toString();
233 }
234 }
235 }