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.util.concurrent.atomic.AtomicReference;
22
23 import org.eclipse.jetty.client.api.Connection;
24 import org.eclipse.jetty.client.api.Request;
25 import org.eclipse.jetty.util.Promise;
26
27 public abstract class MultiplexHttpDestination<C extends Connection> extends HttpDestination implements Promise<Connection>
28 {
29 private final AtomicReference<ConnectState> connect = new AtomicReference<>(ConnectState.DISCONNECTED);
30 private C connection;
31
32 protected MultiplexHttpDestination(HttpClient client, Origin origin)
33 {
34 super(client, origin);
35 }
36
37 @Override
38 protected void send()
39 {
40 while (true)
41 {
42 ConnectState current = connect.get();
43 switch (current)
44 {
45 case DISCONNECTED:
46 {
47 if (!connect.compareAndSet(current, ConnectState.CONNECTING))
48 break;
49 newConnection(this);
50 return;
51 }
52 case CONNECTING:
53 {
54
55 return;
56 }
57 case CONNECTED:
58 {
59 if (process(connection, false))
60 break;
61 return;
62 }
63 default:
64 {
65 abort(new IllegalStateException("Invalid connection state " + current));
66 return;
67 }
68 }
69 }
70 }
71
72 @Override
73 @SuppressWarnings("unchecked")
74 public void succeeded(Connection result)
75 {
76 C connection = this.connection = (C)result;
77 if (connect.compareAndSet(ConnectState.CONNECTING, ConnectState.CONNECTED))
78 {
79 process(connection, true);
80 }
81 else
82 {
83 connection.close();
84 failed(new IllegalStateException());
85 }
86 }
87
88 @Override
89 public void failed(Throwable x)
90 {
91 connect.set(ConnectState.DISCONNECTED);
92 }
93
94 protected boolean process(final C connection, boolean dispatch)
95 {
96 HttpClient client = getHttpClient();
97 final HttpExchange exchange = getHttpExchanges().poll();
98 if (LOG.isDebugEnabled())
99 LOG.debug("Processing {} on {}", exchange, connection);
100 if (exchange == null)
101 return false;
102
103 final Request request = exchange.getRequest();
104 Throwable cause = request.getAbortCause();
105 if (cause != null)
106 {
107 if (LOG.isDebugEnabled())
108 LOG.debug("Aborted before processing {}: {}", exchange, cause);
109
110
111
112 exchange.abort(cause);
113 }
114 else
115 {
116 if (dispatch)
117 {
118 client.getExecutor().execute(new Runnable()
119 {
120 @Override
121 public void run()
122 {
123 send(connection, exchange);
124 }
125 });
126 }
127 else
128 {
129 send(connection, exchange);
130 }
131 }
132 return true;
133 }
134
135 @Override
136 public void close()
137 {
138 super.close();
139 C connection = this.connection;
140 if (connection != null)
141 connection.close();
142 }
143
144 @Override
145 public void close(Connection connection)
146 {
147 super.close(connection);
148 while (true)
149 {
150 ConnectState current = connect.get();
151 if (connect.compareAndSet(current, ConnectState.DISCONNECTED))
152 {
153 if (getHttpClient().isRemoveIdleDestinations())
154 getHttpClient().removeDestination(this);
155 break;
156 }
157 }
158 }
159
160 protected abstract void send(C connection, HttpExchange exchange);
161
162 private enum ConnectState
163 {
164 DISCONNECTED, CONNECTING, CONNECTED
165 }
166 }