author | prappo |
Thu, 09 Nov 2017 15:24:39 +0300 | |
branch | http-client-branch |
changeset 55792 | 0936888d5a4a |
parent 55768 | 8674257c75ce |
child 55798 | fa84be3c77e4 |
permissions | -rw-r--r-- |
55763 | 1 |
/* |
2 |
* Copyright (c) 2017, Oracle and/or its affiliates. All rights reserved. |
|
3 |
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. |
|
4 |
* |
|
5 |
* This code is free software; you can redistribute it and/or modify it |
|
6 |
* under the terms of the GNU General Public License version 2 only, as |
|
7 |
* published by the Free Software Foundation. Oracle designates this |
|
8 |
* particular file as subject to the "Classpath" exception as provided |
|
9 |
* by Oracle in the LICENSE file that accompanied this code. |
|
10 |
* |
|
11 |
* This code is distributed in the hope that it will be useful, but WITHOUT |
|
12 |
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or |
|
13 |
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License |
|
14 |
* version 2 for more details (a copy is included in the LICENSE file that |
|
15 |
* accompanied this code). |
|
16 |
* |
|
17 |
* You should have received a copy of the GNU General Public License version |
|
18 |
* 2 along with this work; if not, write to the Free Software Foundation, |
|
19 |
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. |
|
20 |
* |
|
21 |
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA |
|
22 |
* or visit www.oracle.com if you need additional information or have any |
|
23 |
* questions. |
|
24 |
*/ |
|
25 |
||
26 |
package jdk.incubator.http.internal.common; |
|
27 |
||
28 |
import java.lang.System.Logger.Level; |
|
29 |
import java.nio.ByteBuffer; |
|
30 |
import java.util.List; |
|
31 |
import java.util.Objects; |
|
32 |
import java.util.concurrent.CompletableFuture; |
|
33 |
import java.util.concurrent.Executor; |
|
34 |
import java.util.concurrent.Flow; |
|
35 |
import java.util.concurrent.atomic.AtomicReference; |
|
36 |
import java.util.function.Consumer; |
|
37 |
import javax.net.ssl.SSLEngine; |
|
38 |
import javax.net.ssl.SSLHandshakeException; |
|
39 |
import javax.net.ssl.SSLEngineResult.HandshakeStatus; |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
40 |
import jdk.incubator.http.internal.common.SubscriberWrapper.SchedulingAction; |
55763 | 41 |
import static javax.net.ssl.SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING; |
42 |
import static javax.net.ssl.SSLEngineResult.HandshakeStatus.FINISHED; |
|
43 |
||
44 |
public class SSLTube implements FlowTube { |
|
45 |
||
46 |
static final boolean DEBUG = Utils.DEBUG; // revisit: temporary developer's flag. |
|
47 |
final System.Logger debug = |
|
48 |
Utils.getDebugLogger(this::dbgString, DEBUG); |
|
49 |
||
50 |
private final FlowTube tube; |
|
51 |
private final SSLSubscriberWrapper readSubscriber; |
|
52 |
private final SSLSubscriptionWrapper writeSubscription; |
|
53 |
private final SSLFlowDelegate sslDelegate; |
|
54 |
private final SSLEngine engine; |
|
55 |
private volatile boolean finished; |
|
56 |
||
57 |
public SSLTube(SSLEngine engine, Executor executor, FlowTube tube) { |
|
58 |
Objects.requireNonNull(engine); |
|
59 |
Objects.requireNonNull(executor); |
|
60 |
this.tube = Objects.requireNonNull(tube); |
|
61 |
writeSubscription = new SSLSubscriptionWrapper(); |
|
62 |
readSubscriber = new SSLSubscriberWrapper(); |
|
63 |
this.engine = engine; |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
64 |
sslDelegate = new SSLTubeFlowDelegate(engine, |
55763 | 65 |
executor, |
66 |
readSubscriber, |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
67 |
tube); |
55763 | 68 |
tube.subscribe(sslDelegate.upstreamReader()); |
69 |
sslDelegate.upstreamWriter().onSubscribe(writeSubscription); |
|
70 |
} |
|
71 |
||
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
72 |
// the other possibility would be to pass a lambda to the |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
73 |
// constructor of SSLFlowDelegate (instead of subclassing it). |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
74 |
final class SSLTubeFlowDelegate extends SSLFlowDelegate { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
75 |
SSLTubeFlowDelegate(SSLEngine engine, Executor executor, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
76 |
SSLSubscriberWrapper readSubscriber, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
77 |
FlowTube tube) { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
78 |
super(engine, executor, readSubscriber, tube); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
79 |
} |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
80 |
protected SchedulingAction enterReadScheduling() { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
81 |
readSubscriber.processPendingSubscriber(); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
82 |
return SchedulingAction.CONTINUE; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
83 |
} |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
84 |
} |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
85 |
|
55763 | 86 |
public CompletableFuture<String> getALPN() { |
87 |
return sslDelegate.alpn(); |
|
88 |
} |
|
89 |
||
90 |
@Override |
|
91 |
public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> s) { |
|
92 |
readSubscriber.dropSubscription(); |
|
93 |
readSubscriber.setDelegate(s); |
|
94 |
s.onSubscribe(readSubscription); |
|
95 |
} |
|
96 |
||
97 |
/** |
|
98 |
* Tells whether, or not, this FlowTube has finished receiving data. |
|
99 |
* |
|
100 |
* @return true when one of this FlowTube Subscriber's OnError or onComplete |
|
101 |
* methods have been invoked |
|
102 |
*/ |
|
103 |
@Override |
|
104 |
public boolean isFinished() { |
|
105 |
return finished; |
|
106 |
} |
|
107 |
||
108 |
private volatile Flow.Subscription readSubscription; |
|
109 |
||
110 |
// The DelegateWrapper wraps a subscribed {@code Flow.Subscriber} and |
|
111 |
// tracks the subscriber's state. In particular it makes sure that |
|
112 |
// onComplete/onError are not called before onSubscribed. |
|
113 |
final static class DelegateWrapper implements FlowTube.TubeSubscriber { |
|
114 |
private final FlowTube.TubeSubscriber delegate; |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
115 |
private final System.Logger debug; |
55763 | 116 |
volatile boolean subscribedCalled; |
117 |
volatile boolean subscribedDone; |
|
118 |
volatile boolean completed; |
|
119 |
volatile Throwable error; |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
120 |
DelegateWrapper(Flow.Subscriber<? super List<ByteBuffer>> delegate, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
121 |
System.Logger debug) { |
55763 | 122 |
this.delegate = FlowTube.asTubeSubscriber(delegate); |
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
123 |
this.debug = debug; |
55763 | 124 |
} |
125 |
||
126 |
@Override |
|
127 |
public void dropSubscription() { |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
128 |
if (subscribedCalled && !completed) { |
55763 | 129 |
delegate.dropSubscription(); |
130 |
} |
|
131 |
} |
|
132 |
||
133 |
@Override |
|
134 |
public void onNext(List<ByteBuffer> item) { |
|
135 |
assert subscribedCalled; |
|
136 |
delegate.onNext(item); |
|
137 |
} |
|
138 |
||
139 |
@Override |
|
140 |
public void onSubscribe(Flow.Subscription subscription) { |
|
141 |
onSubscribe(delegate::onSubscribe, subscription); |
|
142 |
} |
|
143 |
||
144 |
@Override |
|
145 |
public void onConnection(Flow.Subscription subscription) { |
|
146 |
onSubscribe(delegate::onConnection, subscription); |
|
147 |
} |
|
148 |
||
149 |
private void onSubscribe(Consumer<Flow.Subscription> method, |
|
150 |
Flow.Subscription subscription) { |
|
151 |
subscribedCalled = true; |
|
152 |
method.accept(subscription); |
|
153 |
Throwable x; |
|
154 |
boolean finished; |
|
155 |
synchronized (this) { |
|
156 |
subscribedDone = true; |
|
157 |
x = error; |
|
158 |
finished = completed; |
|
159 |
} |
|
160 |
if (x != null) { |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
161 |
debug.log(Level.DEBUG, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
162 |
"Subscriber completed before subscribe: forwarding %s", |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
163 |
(Object)x); |
55763 | 164 |
delegate.onError(x); |
165 |
} else if (finished) { |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
166 |
debug.log(Level.DEBUG, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
167 |
"Subscriber completed before subscribe: calling onComplete()"); |
55763 | 168 |
delegate.onComplete(); |
169 |
} |
|
170 |
} |
|
171 |
||
172 |
@Override |
|
173 |
public void onError(Throwable t) { |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
174 |
if (completed) { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
175 |
debug.log(Level.DEBUG, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
176 |
"Subscriber already completed: ignoring %s", |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
177 |
(Object)t); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
178 |
return; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
179 |
} |
55763 | 180 |
boolean subscribed; |
181 |
synchronized (this) { |
|
182 |
if (completed) return; |
|
183 |
error = t; |
|
184 |
completed = true; |
|
185 |
subscribed = subscribedDone; |
|
186 |
} |
|
187 |
if (subscribed) { |
|
188 |
delegate.onError(t); |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
189 |
} else { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
190 |
debug.log(Level.DEBUG, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
191 |
"Subscriber not yet subscribed: stored %s", |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
192 |
(Object)t); |
55763 | 193 |
} |
194 |
} |
|
195 |
||
196 |
@Override |
|
197 |
public void onComplete() { |
|
198 |
if (completed) return; |
|
199 |
boolean subscribed; |
|
200 |
synchronized (this) { |
|
201 |
if (completed) return; |
|
202 |
completed = true; |
|
203 |
subscribed = subscribedDone; |
|
204 |
} |
|
205 |
if (subscribed) { |
|
206 |
delegate.onComplete(); |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
207 |
} else { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
208 |
debug.log(Level.DEBUG, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
209 |
"Subscriber not yet subscribed: stored completed=true"); |
55763 | 210 |
} |
211 |
} |
|
212 |
||
213 |
@Override |
|
214 |
public String toString() { |
|
215 |
return "DelegateWrapper:" + delegate.toString(); |
|
216 |
} |
|
217 |
||
218 |
} |
|
219 |
||
220 |
// Used to read data from the SSLTube. |
|
221 |
final class SSLSubscriberWrapper implements FlowTube.TubeSubscriber { |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
222 |
private AtomicReference<DelegateWrapper> pendingDelegate = |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
223 |
new AtomicReference<>(); |
55763 | 224 |
private volatile DelegateWrapper subscribed; |
225 |
private volatile boolean onCompleteReceived; |
|
226 |
private final AtomicReference<Throwable> errorRef |
|
227 |
= new AtomicReference<>(); |
|
228 |
||
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
229 |
// setDelegate can be called asynchronously when the SSLTube flow |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
230 |
// is connected. At this time the permanent subscriber (this class) |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
231 |
// may already be subscribed (readSubscription != null) or not. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
232 |
// 1. If it's already subscribed (readSubscription != null), we |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
233 |
// are going to signal the SSLFlowDelegate reader, and make sure |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
234 |
// onSubscribed is called within the reader flow |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
235 |
// 2. If it's not yet subscribed (readSubscription == null), then |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
236 |
// we're going to wait for onSubscribe/onConnection to be called. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
237 |
// |
55763 | 238 |
void setDelegate(Flow.Subscriber<? super List<ByteBuffer>> delegate) { |
239 |
debug.log(Level.DEBUG, "SSLSubscriberWrapper (reader) got delegate: %s", |
|
240 |
delegate); |
|
241 |
assert delegate != null; |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
242 |
DelegateWrapper delegateWrapper = new DelegateWrapper(delegate, debug); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
243 |
DelegateWrapper previous; |
55763 | 244 |
Flow.Subscription subscription; |
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
245 |
boolean handleNow; |
55763 | 246 |
synchronized (this) { |
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
247 |
previous = pendingDelegate.getAndSet(delegateWrapper); |
55763 | 248 |
subscription = readSubscription; |
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
249 |
handleNow = this.errorRef.get() != null || finished; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
250 |
} |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
251 |
if (previous != null) { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
252 |
previous.dropSubscription(); |
55763 | 253 |
} |
254 |
if (subscription == null) { |
|
255 |
debug.log(Level.DEBUG, "SSLSubscriberWrapper (reader) no subscription yet"); |
|
256 |
return; |
|
257 |
} |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
258 |
if (handleNow || !sslDelegate.resumeReader()) { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
259 |
processPendingSubscriber(); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
260 |
} |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
261 |
} |
55763 | 262 |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
263 |
// Can be called outside of the flow if an error has already been |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
264 |
// raise. Otherwise, must be called within the SSLFlowDelegate |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
265 |
// downstream reader flow. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
266 |
// If there is a subscription, and if there is a pending delegate, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
267 |
// calls dropSubscription() on the previous delegate (if any), |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
268 |
// then subscribe the pending delegate. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
269 |
void processPendingSubscriber() { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
270 |
Flow.Subscription subscription; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
271 |
DelegateWrapper delegateWrapper, previous; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
272 |
synchronized (this) { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
273 |
delegateWrapper = pendingDelegate.get(); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
274 |
if (delegateWrapper == null) return; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
275 |
subscription = readSubscription; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
276 |
previous = subscribed; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
277 |
} |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
278 |
if (subscription == null) { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
279 |
debug.log(Level.DEBUG, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
280 |
"SSLSubscriberWrapper (reader) %s", |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
281 |
"processPendingSubscriber: no subscription yet"); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
282 |
return; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
283 |
} |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
284 |
delegateWrapper = pendingDelegate.getAndSet(null); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
285 |
if (delegateWrapper == null) return; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
286 |
if (previous != null) { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
287 |
previous.dropSubscription(); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
288 |
} |
55763 | 289 |
onNewSubscription(delegateWrapper, |
290 |
delegateWrapper::onSubscribe, |
|
291 |
subscription); |
|
292 |
} |
|
293 |
||
294 |
@Override |
|
295 |
public void dropSubscription() { |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
296 |
DelegateWrapper subscriberImpl = subscribed; |
55763 | 297 |
if (subscriberImpl != null) { |
298 |
subscriberImpl.dropSubscription(); |
|
299 |
} |
|
300 |
} |
|
301 |
||
302 |
@Override |
|
303 |
public void onConnection(Flow.Subscription subscription) { |
|
304 |
debug.log(Level.DEBUG, |
|
305 |
"SSLSubscriberWrapper (reader) onConnection(%s)", |
|
306 |
subscription); |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
307 |
onSubscribeImpl(subscription); |
55763 | 308 |
} |
309 |
||
310 |
@Override |
|
311 |
public void onSubscribe(Flow.Subscription subscription) { |
|
312 |
debug.log(Level.DEBUG, |
|
313 |
"SSLSubscriberWrapper (reader) onSubscribe(%s)", |
|
314 |
subscription); |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
315 |
onSubscribeImpl(subscription); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
316 |
} |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
317 |
|
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
318 |
// called in the reader flow, from either onSubscribe or onConnection. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
319 |
private void onSubscribeImpl(Flow.Subscription subscription) { |
55763 | 320 |
assert subscription != null; |
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
321 |
DelegateWrapper subscriberImpl, pending; |
55763 | 322 |
synchronized (this) { |
323 |
readSubscription = subscription; |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
324 |
subscriberImpl = subscribed; |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
325 |
pending = pendingDelegate.get(); |
55763 | 326 |
} |
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
327 |
|
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
328 |
if (subscriberImpl == null && pending == null) { |
55763 | 329 |
debug.log(Level.DEBUG, |
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
330 |
"SSLSubscriberWrapper (reader) onSubscribeImpl: %s", |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
331 |
"no delegate yet"); |
55763 | 332 |
return; |
333 |
} |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
334 |
|
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
335 |
if (pending == null) { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
336 |
// There is no pending delegate, but we have a previously |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
337 |
// subscribed delegate. This is obviously a re-subscribe. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
338 |
// We are in the downstream reader flow, so we should call |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
339 |
// onConnection directly. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
340 |
debug.log(Level.DEBUG, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
341 |
"SSLSubscriberWrapper (reader) onSubscribeImpl: %s", |
55792
0936888d5a4a
http-client-branch: (cleanup) unused imports, unused methods, removed 1 class; typos;
prappo
parents:
55768
diff
changeset
|
342 |
"resubscribing"); |
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
343 |
onNewSubscription(subscriberImpl, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
344 |
subscriberImpl::onConnection, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
345 |
subscription); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
346 |
} else { |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
347 |
// We have some pending subscriber: subscribe it now that we have |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
348 |
// a subscription. If we already had a previous delegate then |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
349 |
// it will get a dropSubscription(). |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
350 |
debug.log(Level.DEBUG, |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
351 |
"SSLSubscriberWrapper (reader) onSubscribeImpl: %s", |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
352 |
"subscribing pending"); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
353 |
processPendingSubscriber(); |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
354 |
} |
55763 | 355 |
} |
356 |
||
357 |
private void onNewSubscription(DelegateWrapper subscriberImpl, |
|
358 |
Consumer<Flow.Subscription> method, |
|
359 |
Flow.Subscription subscription) { |
|
360 |
assert subscriberImpl != null; |
|
361 |
assert method != null; |
|
362 |
assert subscription != null; |
|
363 |
||
364 |
Throwable failed; |
|
365 |
boolean completed; |
|
366 |
// reset any demand that may have been made by the previous |
|
367 |
// subscriber |
|
368 |
sslDelegate.resetReaderDemand(); |
|
369 |
// send the subscription to the subscriber. |
|
370 |
method.accept(subscription); |
|
371 |
||
372 |
// The following twisted logic is just here that we don't invoke |
|
373 |
// onError before onSubscribe. It also prevents race conditions |
|
374 |
// if onError is invoked concurrently with setDelegate. |
|
375 |
synchronized (this) { |
|
376 |
failed = this.errorRef.get(); |
|
377 |
completed = finished; |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
378 |
subscribed = subscriberImpl; |
55763 | 379 |
} |
380 |
if (failed != null) { |
|
381 |
subscriberImpl.onError(failed); |
|
382 |
} else if (completed) { |
|
383 |
subscriberImpl.onComplete(); |
|
384 |
} |
|
385 |
} |
|
386 |
||
387 |
@Override |
|
388 |
public void onNext(List<ByteBuffer> item) { |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
389 |
subscribed.onNext(item); |
55763 | 390 |
} |
391 |
||
392 |
public void onErrorImpl(Throwable throwable) { |
|
393 |
// The following twisted logic is just here that we don't invoke |
|
394 |
// onError before onSubscribe. It also prevents race conditions |
|
395 |
// if onError is invoked concurrently with setDelegate. |
|
396 |
// See setDelegate. |
|
397 |
||
398 |
errorRef.compareAndSet(null, throwable); |
|
399 |
Throwable failed = errorRef.get(); |
|
400 |
finished = true; |
|
401 |
debug.log(Level.DEBUG, "%s: onErrorImpl: %s", this, throwable); |
|
402 |
DelegateWrapper subscriberImpl; |
|
403 |
synchronized (this) { |
|
404 |
subscriberImpl = subscribed; |
|
405 |
} |
|
406 |
if (subscriberImpl != null) { |
|
407 |
subscriberImpl.onError(failed); |
|
408 |
} else { |
|
409 |
debug.log(Level.DEBUG, "%s: delegate null, stored %s", this, failed); |
|
410 |
} |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
411 |
// now if we have any pending subscriber, we should forward |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
412 |
// the error to them immediately as the read scheduler will |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
413 |
// already be stopped. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
414 |
processPendingSubscriber(); |
55763 | 415 |
} |
416 |
||
417 |
@Override |
|
418 |
public void onError(Throwable throwable) { |
|
419 |
assert !finished && !onCompleteReceived; |
|
420 |
onErrorImpl(throwable); |
|
421 |
} |
|
422 |
||
423 |
private boolean handshaking() { |
|
424 |
HandshakeStatus hs = engine.getHandshakeStatus(); |
|
425 |
return !(hs == NOT_HANDSHAKING || hs == FINISHED); |
|
426 |
} |
|
427 |
||
428 |
@Override |
|
429 |
public void onComplete() { |
|
430 |
assert !finished && !onCompleteReceived; |
|
431 |
onCompleteReceived = true; |
|
432 |
DelegateWrapper subscriberImpl; |
|
433 |
synchronized(this) { |
|
434 |
subscriberImpl = subscribed; |
|
435 |
} |
|
436 |
||
437 |
if (handshaking()) { |
|
438 |
onErrorImpl(new SSLHandshakeException( |
|
439 |
"Remote host terminated the handshake")); |
|
440 |
} else if (subscriberImpl != null) { |
|
441 |
finished = true; |
|
442 |
subscriberImpl.onComplete(); |
|
443 |
} |
|
55768
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
444 |
// now if we have any pending subscriber, we should complete |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
445 |
// them immediately as the read scheduler will already be stopped. |
8674257c75ce
http-client-branch: SSLTube should call onSubscribe in the flow
dfuchs
parents:
55763
diff
changeset
|
446 |
processPendingSubscriber(); |
55763 | 447 |
} |
448 |
} |
|
449 |
||
450 |
@Override |
|
451 |
public void connectFlows(TubePublisher writePub, |
|
452 |
TubeSubscriber readSub) { |
|
453 |
debug.log(Level.DEBUG, "connecting flows"); |
|
454 |
readSubscriber.setDelegate(readSub); |
|
455 |
writePub.subscribe(this); |
|
456 |
} |
|
457 |
||
458 |
/** Outstanding write demand from the SSL Flow Delegate. */ |
|
459 |
private final Demand writeDemand = new Demand(); |
|
460 |
||
461 |
final class SSLSubscriptionWrapper implements Flow.Subscription { |
|
462 |
||
463 |
volatile Flow.Subscription delegate; |
|
464 |
||
465 |
void setSubscription(Flow.Subscription sub) { |
|
466 |
long demand = writeDemand.get(); // FIXME: isn't it a racy way of passing the demand? |
|
467 |
delegate = sub; |
|
468 |
debug.log(Level.DEBUG, "setSubscription: demand=%d", demand); |
|
469 |
if (demand > 0) |
|
470 |
sub.request(demand); |
|
471 |
} |
|
472 |
||
473 |
@Override |
|
474 |
public void request(long n) { |
|
475 |
writeDemand.increase(n); |
|
476 |
debug.log(Level.DEBUG, "request: n=%d", n); |
|
477 |
Flow.Subscription sub = delegate; |
|
478 |
if (sub != null && n > 0) { |
|
479 |
sub.request(n); |
|
480 |
} |
|
481 |
} |
|
482 |
||
483 |
@Override |
|
484 |
public void cancel() { |
|
485 |
// TODO: no-op or error? |
|
486 |
} |
|
487 |
} |
|
488 |
||
489 |
/* Subscriber - writing side */ |
|
490 |
@Override |
|
491 |
public void onSubscribe(Flow.Subscription subscription) { |
|
492 |
Objects.requireNonNull(subscription); |
|
493 |
Flow.Subscription x = writeSubscription.delegate; |
|
494 |
if (x != null) |
|
495 |
x.cancel(); |
|
496 |
||
497 |
writeSubscription.setSubscription(subscription); |
|
498 |
} |
|
499 |
||
500 |
@Override |
|
501 |
public void onNext(List<ByteBuffer> item) { |
|
502 |
Objects.requireNonNull(item); |
|
503 |
boolean decremented = writeDemand.tryDecrement(); |
|
504 |
assert decremented : "Unexpected writeDemand: "; |
|
505 |
debug.log(Level.DEBUG, |
|
506 |
"sending %d buffers to SSL flow delegate", item.size()); |
|
507 |
sslDelegate.upstreamWriter().onNext(item); |
|
508 |
} |
|
509 |
||
510 |
@Override |
|
511 |
public void onError(Throwable throwable) { |
|
512 |
Objects.requireNonNull(throwable); |
|
513 |
sslDelegate.upstreamWriter().onError(throwable); |
|
514 |
} |
|
515 |
||
516 |
@Override |
|
517 |
public void onComplete() { |
|
518 |
sslDelegate.upstreamWriter().onComplete(); |
|
519 |
} |
|
520 |
||
521 |
@Override |
|
522 |
public String toString() { |
|
523 |
return dbgString(); |
|
524 |
} |
|
525 |
||
526 |
final String dbgString() { |
|
527 |
return "SSLTube(" + tube + ")"; |
|
528 |
} |
|
529 |
||
530 |
} |