author | iignatyev |
Tue, 25 Sep 2018 18:26:46 -0700 | |
changeset 51876 | d2b381ea8477 |
parent 49765 | ee6f7a61f3a5 |
child 53467 | 97cf88608d76 |
child 56451 | 9585061fdb04 |
permissions | -rw-r--r-- |
48083 | 1 |
/* |
49765 | 2 |
* Copyright (c) 2016, 2018, Oracle and/or its affiliates. All rights reserved. |
48083 | 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 |
||
49765 | 26 |
package jdk.internal.net.http; |
48083 | 27 |
|
49765 | 28 |
import java.io.BufferedReader; |
29 |
import java.io.FilePermission; |
|
48083 | 30 |
import java.io.IOException; |
31 |
import java.io.InputStream; |
|
49765 | 32 |
import java.io.InputStreamReader; |
48083 | 33 |
import java.lang.System.Logger.Level; |
49765 | 34 |
import java.net.http.HttpHeaders; |
35 |
import java.net.http.HttpResponse; |
|
48083 | 36 |
import java.nio.ByteBuffer; |
37 |
import java.nio.channels.FileChannel; |
|
49765 | 38 |
import java.nio.charset.Charset; |
48083 | 39 |
import java.nio.file.OpenOption; |
40 |
import java.nio.file.Path; |
|
41 |
import java.security.AccessController; |
|
42 |
import java.security.PrivilegedActionException; |
|
43 |
import java.security.PrivilegedExceptionAction; |
|
44 |
import java.util.ArrayList; |
|
45 |
import java.util.Iterator; |
|
46 |
import java.util.List; |
|
47 |
import java.util.Objects; |
|
48 |
import java.util.Optional; |
|
49 |
import java.util.concurrent.ArrayBlockingQueue; |
|
50 |
import java.util.concurrent.BlockingQueue; |
|
51 |
import java.util.concurrent.CompletableFuture; |
|
52 |
import java.util.concurrent.CompletionStage; |
|
53 |
import java.util.concurrent.Flow; |
|
48408
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
54 |
import java.util.concurrent.Flow.Subscriber; |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
55 |
import java.util.concurrent.Flow.Subscription; |
48083 | 56 |
import java.util.concurrent.atomic.AtomicBoolean; |
49765 | 57 |
import java.util.concurrent.atomic.AtomicReference; |
48083 | 58 |
import java.util.function.Consumer; |
59 |
import java.util.function.Function; |
|
49765 | 60 |
import java.util.stream.Stream; |
61 |
import java.net.http.HttpResponse.BodySubscriber; |
|
62 |
import jdk.internal.net.http.common.Log; |
|
63 |
import jdk.internal.net.http.common.Logger; |
|
64 |
import jdk.internal.net.http.common.MinimalFuture; |
|
65 |
import jdk.internal.net.http.common.Utils; |
|
66 |
import static java.nio.charset.StandardCharsets.UTF_8; |
|
48083 | 67 |
|
49765 | 68 |
public class ResponseSubscribers { |
48083 | 69 |
|
49765 | 70 |
public static class ConsumerSubscriber implements BodySubscriber<Void> { |
48083 | 71 |
private final Consumer<Optional<byte[]>> consumer; |
72 |
private Flow.Subscription subscription; |
|
73 |
private final CompletableFuture<Void> result = new MinimalFuture<>(); |
|
74 |
private final AtomicBoolean subscribed = new AtomicBoolean(); |
|
75 |
||
49765 | 76 |
public ConsumerSubscriber(Consumer<Optional<byte[]>> consumer) { |
48379
5382baab8371
8193698: Null handling in BodyPublisher, BodyHandler, and BodySubscriber convenience static factory methods
chegar
parents:
48083
diff
changeset
|
77 |
this.consumer = Objects.requireNonNull(consumer); |
48083 | 78 |
} |
79 |
||
80 |
@Override |
|
81 |
public CompletionStage<Void> getBody() { |
|
82 |
return result; |
|
83 |
} |
|
84 |
||
85 |
@Override |
|
86 |
public void onSubscribe(Flow.Subscription subscription) { |
|
87 |
if (!subscribed.compareAndSet(false, true)) { |
|
88 |
subscription.cancel(); |
|
89 |
} else { |
|
90 |
this.subscription = subscription; |
|
91 |
subscription.request(1); |
|
92 |
} |
|
93 |
} |
|
94 |
||
95 |
@Override |
|
96 |
public void onNext(List<ByteBuffer> items) { |
|
97 |
for (ByteBuffer item : items) { |
|
98 |
byte[] buf = new byte[item.remaining()]; |
|
99 |
item.get(buf); |
|
100 |
consumer.accept(Optional.of(buf)); |
|
101 |
} |
|
102 |
subscription.request(1); |
|
103 |
} |
|
104 |
||
105 |
@Override |
|
106 |
public void onError(Throwable throwable) { |
|
107 |
result.completeExceptionally(throwable); |
|
108 |
} |
|
109 |
||
110 |
@Override |
|
111 |
public void onComplete() { |
|
112 |
consumer.accept(Optional.empty()); |
|
113 |
result.complete(null); |
|
114 |
} |
|
115 |
||
116 |
} |
|
117 |
||
49765 | 118 |
/** |
119 |
* A Subscriber that writes the flow of data to a given file. |
|
120 |
* |
|
121 |
* Privileged actions are performed within a limited doPrivileged that only |
|
122 |
* asserts the specific, write, file permissions that were checked during |
|
123 |
* the construction of this PathSubscriber. |
|
124 |
*/ |
|
125 |
public static class PathSubscriber implements BodySubscriber<Path> { |
|
126 |
||
127 |
private static final FilePermission[] EMPTY_FILE_PERMISSIONS = new FilePermission[0]; |
|
48083 | 128 |
|
129 |
private final Path file; |
|
49765 | 130 |
private final OpenOption[] options; |
131 |
private final FilePermission[] filePermissions; |
|
48083 | 132 |
private final CompletableFuture<Path> result = new MinimalFuture<>(); |
133 |
||
134 |
private volatile Flow.Subscription subscription; |
|
135 |
private volatile FileChannel out; |
|
136 |
||
49765 | 137 |
private static final String pathForSecurityCheck(Path path) { |
138 |
return path.toFile().getPath(); |
|
48083 | 139 |
} |
140 |
||
49765 | 141 |
/** |
142 |
* Factory for creating PathSubscriber. |
|
143 |
* |
|
144 |
* Permission checks are performed here before construction of the |
|
145 |
* PathSubscriber. Permission checking and construction are deliberately |
|
146 |
* and tightly co-located. |
|
147 |
*/ |
|
148 |
public static PathSubscriber create(Path file, |
|
149 |
List<OpenOption> options) { |
|
150 |
FilePermission filePermission = null; |
|
151 |
SecurityManager sm = System.getSecurityManager(); |
|
152 |
if (sm != null) { |
|
153 |
String fn = pathForSecurityCheck(file); |
|
154 |
FilePermission writePermission = new FilePermission(fn, "write"); |
|
155 |
sm.checkPermission(writePermission); |
|
156 |
filePermission = writePermission; |
|
157 |
} |
|
158 |
return new PathSubscriber(file, options, filePermission); |
|
159 |
} |
|
160 |
||
161 |
// pp so handler implementations in the same package can construct |
|
162 |
/*package-private*/ PathSubscriber(Path file, |
|
163 |
List<OpenOption> options, |
|
164 |
FilePermission... filePermissions) { |
|
165 |
this.file = file; |
|
166 |
this.options = options.stream().toArray(OpenOption[]::new); |
|
167 |
this.filePermissions = |
|
168 |
filePermissions == null ? EMPTY_FILE_PERMISSIONS : filePermissions; |
|
48083 | 169 |
} |
170 |
||
171 |
@Override |
|
172 |
public void onSubscribe(Flow.Subscription subscription) { |
|
173 |
this.subscription = subscription; |
|
49765 | 174 |
if (System.getSecurityManager() == null) { |
175 |
try { |
|
176 |
out = FileChannel.open(file, options); |
|
177 |
} catch (IOException ioe) { |
|
178 |
result.completeExceptionally(ioe); |
|
179 |
return; |
|
180 |
} |
|
181 |
} else { |
|
182 |
try { |
|
183 |
PrivilegedExceptionAction<FileChannel> pa = |
|
184 |
() -> FileChannel.open(file, options); |
|
185 |
out = AccessController.doPrivileged(pa, null, filePermissions); |
|
186 |
} catch (PrivilegedActionException pae) { |
|
187 |
Throwable t = pae.getCause() != null ? pae.getCause() : pae; |
|
188 |
result.completeExceptionally(t); |
|
189 |
subscription.cancel(); |
|
190 |
return; |
|
191 |
} |
|
48083 | 192 |
} |
193 |
subscription.request(1); |
|
194 |
} |
|
195 |
||
196 |
@Override |
|
197 |
public void onNext(List<ByteBuffer> items) { |
|
198 |
try { |
|
199 |
out.write(items.toArray(Utils.EMPTY_BB_ARRAY)); |
|
200 |
} catch (IOException ex) { |
|
201 |
Utils.close(out); |
|
202 |
subscription.cancel(); |
|
203 |
result.completeExceptionally(ex); |
|
204 |
} |
|
205 |
subscription.request(1); |
|
206 |
} |
|
207 |
||
208 |
@Override |
|
209 |
public void onError(Throwable e) { |
|
210 |
result.completeExceptionally(e); |
|
211 |
Utils.close(out); |
|
212 |
} |
|
213 |
||
214 |
@Override |
|
215 |
public void onComplete() { |
|
216 |
Utils.close(out); |
|
217 |
result.complete(file); |
|
218 |
} |
|
219 |
||
220 |
@Override |
|
221 |
public CompletionStage<Path> getBody() { |
|
222 |
return result; |
|
223 |
} |
|
224 |
} |
|
225 |
||
49765 | 226 |
public static class ByteArraySubscriber<T> implements BodySubscriber<T> { |
48083 | 227 |
private final Function<byte[], T> finisher; |
228 |
private final CompletableFuture<T> result = new MinimalFuture<>(); |
|
229 |
private final List<ByteBuffer> received = new ArrayList<>(); |
|
230 |
||
231 |
private volatile Flow.Subscription subscription; |
|
232 |
||
49765 | 233 |
public ByteArraySubscriber(Function<byte[],T> finisher) { |
48083 | 234 |
this.finisher = finisher; |
235 |
} |
|
236 |
||
237 |
@Override |
|
238 |
public void onSubscribe(Flow.Subscription subscription) { |
|
239 |
if (this.subscription != null) { |
|
240 |
subscription.cancel(); |
|
241 |
return; |
|
242 |
} |
|
243 |
this.subscription = subscription; |
|
244 |
// We can handle whatever you've got |
|
245 |
subscription.request(Long.MAX_VALUE); |
|
246 |
} |
|
247 |
||
248 |
@Override |
|
249 |
public void onNext(List<ByteBuffer> items) { |
|
250 |
// incoming buffers are allocated by http client internally, |
|
251 |
// and won't be used anywhere except this place. |
|
252 |
// So it's free simply to store them for further processing. |
|
253 |
assert Utils.hasRemaining(items); |
|
49765 | 254 |
received.addAll(items); |
48083 | 255 |
} |
256 |
||
257 |
@Override |
|
258 |
public void onError(Throwable throwable) { |
|
259 |
received.clear(); |
|
260 |
result.completeExceptionally(throwable); |
|
261 |
} |
|
262 |
||
263 |
static private byte[] join(List<ByteBuffer> bytes) { |
|
264 |
int size = Utils.remaining(bytes, Integer.MAX_VALUE); |
|
265 |
byte[] res = new byte[size]; |
|
266 |
int from = 0; |
|
267 |
for (ByteBuffer b : bytes) { |
|
268 |
int l = b.remaining(); |
|
269 |
b.get(res, from, l); |
|
270 |
from += l; |
|
271 |
} |
|
272 |
return res; |
|
273 |
} |
|
274 |
||
275 |
@Override |
|
276 |
public void onComplete() { |
|
277 |
try { |
|
278 |
result.complete(finisher.apply(join(received))); |
|
279 |
received.clear(); |
|
280 |
} catch (IllegalArgumentException e) { |
|
281 |
result.completeExceptionally(e); |
|
282 |
} |
|
283 |
} |
|
284 |
||
285 |
@Override |
|
286 |
public CompletionStage<T> getBody() { |
|
287 |
return result; |
|
288 |
} |
|
289 |
} |
|
290 |
||
291 |
/** |
|
292 |
* An InputStream built on top of the Flow API. |
|
293 |
*/ |
|
49765 | 294 |
public static class HttpResponseInputStream extends InputStream |
295 |
implements BodySubscriber<InputStream> |
|
48083 | 296 |
{ |
297 |
final static int MAX_BUFFERS_IN_QUEUE = 1; // lock-step with the producer |
|
298 |
||
299 |
// An immutable ByteBuffer sentinel to mark that the last byte was received. |
|
300 |
private static final ByteBuffer LAST_BUFFER = ByteBuffer.wrap(new byte[0]); |
|
301 |
private static final List<ByteBuffer> LAST_LIST = List.of(LAST_BUFFER); |
|
49765 | 302 |
private static final Logger debug = |
303 |
Utils.getDebugLogger("HttpResponseInputStream"::toString, Utils.DEBUG); |
|
48083 | 304 |
|
305 |
// A queue of yet unprocessed ByteBuffers received from the flow API. |
|
306 |
private final BlockingQueue<List<ByteBuffer>> buffers; |
|
307 |
private volatile Flow.Subscription subscription; |
|
308 |
private volatile boolean closed; |
|
309 |
private volatile Throwable failed; |
|
310 |
private volatile Iterator<ByteBuffer> currentListItr; |
|
311 |
private volatile ByteBuffer currentBuffer; |
|
312 |
private final AtomicBoolean subscribed = new AtomicBoolean(); |
|
313 |
||
49765 | 314 |
public HttpResponseInputStream() { |
48083 | 315 |
this(MAX_BUFFERS_IN_QUEUE); |
316 |
} |
|
317 |
||
318 |
HttpResponseInputStream(int maxBuffers) { |
|
319 |
int capacity = (maxBuffers <= 0 ? MAX_BUFFERS_IN_QUEUE : maxBuffers); |
|
320 |
// 1 additional slot needed for LAST_LIST added by onComplete |
|
321 |
this.buffers = new ArrayBlockingQueue<>(capacity + 1); |
|
322 |
} |
|
323 |
||
324 |
@Override |
|
325 |
public CompletionStage<InputStream> getBody() { |
|
326 |
// Returns the stream immediately, before the |
|
327 |
// response body is received. |
|
49765 | 328 |
// This makes it possible for sendAsync().get().body() |
48083 | 329 |
// to complete before the response body is received. |
330 |
return CompletableFuture.completedStage(this); |
|
331 |
} |
|
332 |
||
333 |
// Returns the current byte buffer to read from. |
|
334 |
// If the current buffer has no remaining data, this method will take the |
|
335 |
// next buffer from the buffers queue, possibly blocking until |
|
336 |
// a new buffer is made available through the Flow API, or the |
|
337 |
// end of the flow has been reached. |
|
338 |
private ByteBuffer current() throws IOException { |
|
339 |
while (currentBuffer == null || !currentBuffer.hasRemaining()) { |
|
340 |
// Check whether the stream is closed or exhausted |
|
341 |
if (closed || failed != null) { |
|
342 |
throw new IOException("closed", failed); |
|
343 |
} |
|
344 |
if (currentBuffer == LAST_BUFFER) break; |
|
345 |
||
346 |
try { |
|
347 |
if (currentListItr == null || !currentListItr.hasNext()) { |
|
348 |
// Take a new list of buffers from the queue, blocking |
|
349 |
// if none is available yet... |
|
350 |
||
49765 | 351 |
if (debug.on()) debug.log("Taking list of Buffers"); |
48083 | 352 |
List<ByteBuffer> lb = buffers.take(); |
353 |
currentListItr = lb.iterator(); |
|
49765 | 354 |
if (debug.on()) debug.log("List of Buffers Taken"); |
48083 | 355 |
|
356 |
// Check whether an exception was encountered upstream |
|
357 |
if (closed || failed != null) |
|
358 |
throw new IOException("closed", failed); |
|
359 |
||
360 |
// Check whether we're done. |
|
361 |
if (lb == LAST_LIST) { |
|
362 |
currentListItr = null; |
|
363 |
currentBuffer = LAST_BUFFER; |
|
364 |
break; |
|
365 |
} |
|
366 |
||
367 |
// Request another upstream item ( list of buffers ) |
|
368 |
Flow.Subscription s = subscription; |
|
369 |
if (s != null) { |
|
49765 | 370 |
if (debug.on()) debug.log("Increased demand by 1"); |
48083 | 371 |
s.request(1); |
372 |
} |
|
373 |
assert currentListItr != null; |
|
374 |
if (lb.isEmpty()) continue; |
|
375 |
} |
|
376 |
assert currentListItr != null; |
|
377 |
assert currentListItr.hasNext(); |
|
49765 | 378 |
if (debug.on()) debug.log("Next Buffer"); |
48083 | 379 |
currentBuffer = currentListItr.next(); |
380 |
} catch (InterruptedException ex) { |
|
381 |
// continue |
|
382 |
} |
|
383 |
} |
|
384 |
assert currentBuffer == LAST_BUFFER || currentBuffer.hasRemaining(); |
|
385 |
return currentBuffer; |
|
386 |
} |
|
387 |
||
388 |
@Override |
|
389 |
public int read(byte[] bytes, int off, int len) throws IOException { |
|
390 |
// get the buffer to read from, possibly blocking if |
|
391 |
// none is available |
|
392 |
ByteBuffer buffer; |
|
393 |
if ((buffer = current()) == LAST_BUFFER) return -1; |
|
394 |
||
395 |
// don't attempt to read more than what is available |
|
396 |
// in the current buffer. |
|
397 |
int read = Math.min(buffer.remaining(), len); |
|
398 |
assert read > 0 && read <= buffer.remaining(); |
|
399 |
||
400 |
// buffer.get() will do the boundary check for us. |
|
401 |
buffer.get(bytes, off, read); |
|
402 |
return read; |
|
403 |
} |
|
404 |
||
405 |
@Override |
|
406 |
public int read() throws IOException { |
|
407 |
ByteBuffer buffer; |
|
408 |
if ((buffer = current()) == LAST_BUFFER) return -1; |
|
409 |
return buffer.get() & 0xFF; |
|
410 |
} |
|
411 |
||
412 |
@Override |
|
413 |
public void onSubscribe(Flow.Subscription s) { |
|
414 |
try { |
|
415 |
if (!subscribed.compareAndSet(false, true)) { |
|
416 |
s.cancel(); |
|
417 |
} else { |
|
418 |
// check whether the stream is already closed. |
|
419 |
// if so, we should cancel the subscription |
|
420 |
// immediately. |
|
421 |
boolean closed; |
|
422 |
synchronized (this) { |
|
423 |
closed = this.closed; |
|
424 |
if (!closed) { |
|
425 |
this.subscription = s; |
|
426 |
} |
|
427 |
} |
|
428 |
if (closed) { |
|
429 |
s.cancel(); |
|
430 |
return; |
|
431 |
} |
|
432 |
assert buffers.remainingCapacity() > 1; // should contain at least 2 |
|
49765 | 433 |
if (debug.on()) |
434 |
debug.log("onSubscribe: requesting " |
|
435 |
+ Math.max(1, buffers.remainingCapacity() - 1)); |
|
48083 | 436 |
s.request(Math.max(1, buffers.remainingCapacity() - 1)); |
437 |
} |
|
438 |
} catch (Throwable t) { |
|
439 |
failed = t; |
|
440 |
try { |
|
441 |
close(); |
|
442 |
} catch (IOException x) { |
|
443 |
// OK |
|
444 |
} finally { |
|
445 |
onError(t); |
|
446 |
} |
|
447 |
} |
|
448 |
} |
|
449 |
||
450 |
@Override |
|
451 |
public void onNext(List<ByteBuffer> t) { |
|
452 |
Objects.requireNonNull(t); |
|
453 |
try { |
|
49765 | 454 |
if (debug.on()) debug.log("next item received"); |
48083 | 455 |
if (!buffers.offer(t)) { |
456 |
throw new IllegalStateException("queue is full"); |
|
457 |
} |
|
49765 | 458 |
if (debug.on()) debug.log("item offered"); |
48083 | 459 |
} catch (Throwable ex) { |
460 |
failed = ex; |
|
461 |
try { |
|
462 |
close(); |
|
463 |
} catch (IOException ex1) { |
|
464 |
// OK |
|
465 |
} finally { |
|
466 |
onError(ex); |
|
467 |
} |
|
468 |
} |
|
469 |
} |
|
470 |
||
471 |
@Override |
|
472 |
public void onError(Throwable thrwbl) { |
|
473 |
subscription = null; |
|
474 |
failed = Objects.requireNonNull(thrwbl); |
|
475 |
// The client process that reads the input stream might |
|
476 |
// be blocked in queue.take(). |
|
477 |
// Tries to offer LAST_LIST to the queue. If the queue is |
|
478 |
// full we don't care if we can't insert this buffer, as |
|
479 |
// the client can't be blocked in queue.take() in that case. |
|
480 |
// Adding LAST_LIST to the queue is harmless, as the client |
|
481 |
// should find failed != null before handling LAST_LIST. |
|
482 |
buffers.offer(LAST_LIST); |
|
483 |
} |
|
484 |
||
485 |
@Override |
|
486 |
public void onComplete() { |
|
487 |
subscription = null; |
|
488 |
onNext(LAST_LIST); |
|
489 |
} |
|
490 |
||
491 |
@Override |
|
492 |
public void close() throws IOException { |
|
493 |
Flow.Subscription s; |
|
494 |
synchronized (this) { |
|
495 |
if (closed) return; |
|
496 |
closed = true; |
|
497 |
s = subscription; |
|
498 |
subscription = null; |
|
499 |
} |
|
500 |
// s will be null if already completed |
|
501 |
try { |
|
502 |
if (s != null) { |
|
503 |
s.cancel(); |
|
504 |
} |
|
505 |
} finally { |
|
506 |
buffers.offer(LAST_LIST); |
|
507 |
super.close(); |
|
508 |
} |
|
509 |
} |
|
510 |
||
511 |
} |
|
512 |
||
49765 | 513 |
public static BodySubscriber<Stream<String>> createLineStream() { |
514 |
return createLineStream(UTF_8); |
|
515 |
} |
|
48083 | 516 |
|
49765 | 517 |
public static BodySubscriber<Stream<String>> createLineStream(Charset charset) { |
518 |
Objects.requireNonNull(charset); |
|
519 |
BodySubscriber<InputStream> s = new HttpResponseInputStream(); |
|
520 |
return new MappingSubscriber<InputStream,Stream<String>>(s, |
|
521 |
(InputStream stream) -> { |
|
522 |
return new BufferedReader(new InputStreamReader(stream, charset)) |
|
523 |
.lines().onClose(() -> Utils.close(stream)); |
|
524 |
}); |
|
48083 | 525 |
} |
526 |
||
527 |
/** |
|
528 |
* Currently this consumes all of the data and ignores it |
|
529 |
*/ |
|
49765 | 530 |
public static class NullSubscriber<T> implements BodySubscriber<T> { |
48083 | 531 |
|
532 |
private final CompletableFuture<T> cf = new MinimalFuture<>(); |
|
533 |
private final Optional<T> result; |
|
534 |
private final AtomicBoolean subscribed = new AtomicBoolean(); |
|
535 |
||
49765 | 536 |
public NullSubscriber(Optional<T> result) { |
48083 | 537 |
this.result = result; |
538 |
} |
|
539 |
||
540 |
@Override |
|
541 |
public void onSubscribe(Flow.Subscription subscription) { |
|
542 |
if (!subscribed.compareAndSet(false, true)) { |
|
543 |
subscription.cancel(); |
|
544 |
} else { |
|
545 |
subscription.request(Long.MAX_VALUE); |
|
546 |
} |
|
547 |
} |
|
548 |
||
549 |
@Override |
|
550 |
public void onNext(List<ByteBuffer> items) { |
|
551 |
Objects.requireNonNull(items); |
|
552 |
} |
|
553 |
||
554 |
@Override |
|
555 |
public void onError(Throwable throwable) { |
|
556 |
cf.completeExceptionally(throwable); |
|
557 |
} |
|
558 |
||
559 |
@Override |
|
560 |
public void onComplete() { |
|
561 |
if (result.isPresent()) { |
|
562 |
cf.complete(result.get()); |
|
563 |
} else { |
|
564 |
cf.complete(null); |
|
565 |
} |
|
566 |
} |
|
567 |
||
568 |
@Override |
|
569 |
public CompletionStage<T> getBody() { |
|
570 |
return cf; |
|
571 |
} |
|
572 |
} |
|
48408
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
573 |
|
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
574 |
/** An adapter between {@code BodySubscriber} and {@code Flow.Subscriber}. */ |
49765 | 575 |
public static final class SubscriberAdapter<S extends Subscriber<? super List<ByteBuffer>>,R> |
576 |
implements BodySubscriber<R> |
|
48408
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
577 |
{ |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
578 |
private final CompletableFuture<R> cf = new MinimalFuture<>(); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
579 |
private final S subscriber; |
49765 | 580 |
private final Function<? super S,? extends R> finisher; |
48408
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
581 |
private volatile Subscription subscription; |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
582 |
|
49765 | 583 |
public SubscriberAdapter(S subscriber, Function<? super S,? extends R> finisher) { |
48408
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
584 |
this.subscriber = Objects.requireNonNull(subscriber); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
585 |
this.finisher = Objects.requireNonNull(finisher); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
586 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
587 |
|
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
588 |
@Override |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
589 |
public void onSubscribe(Subscription subscription) { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
590 |
Objects.requireNonNull(subscription); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
591 |
if (this.subscription != null) { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
592 |
subscription.cancel(); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
593 |
} else { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
594 |
this.subscription = subscription; |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
595 |
subscriber.onSubscribe(subscription); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
596 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
597 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
598 |
|
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
599 |
@Override |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
600 |
public void onNext(List<ByteBuffer> item) { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
601 |
Objects.requireNonNull(item); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
602 |
try { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
603 |
subscriber.onNext(item); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
604 |
} catch (Throwable throwable) { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
605 |
subscription.cancel(); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
606 |
onError(throwable); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
607 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
608 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
609 |
|
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
610 |
@Override |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
611 |
public void onError(Throwable throwable) { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
612 |
Objects.requireNonNull(throwable); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
613 |
try { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
614 |
subscriber.onError(throwable); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
615 |
} finally { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
616 |
cf.completeExceptionally(throwable); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
617 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
618 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
619 |
|
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
620 |
@Override |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
621 |
public void onComplete() { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
622 |
try { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
623 |
subscriber.onComplete(); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
624 |
} finally { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
625 |
try { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
626 |
cf.complete(finisher.apply(subscriber)); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
627 |
} catch (Throwable throwable) { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
628 |
cf.completeExceptionally(throwable); |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
629 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
630 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
631 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
632 |
|
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
633 |
@Override |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
634 |
public CompletionStage<R> getBody() { |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
635 |
return cf; |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
636 |
} |
4f830b447edf
8193365: Improve interoperability between HTTP Client's BodyPublisher/BodySubscriber and Flow.Subscriber/Publisher
chegar
parents:
48379
diff
changeset
|
637 |
} |
49765 | 638 |
|
639 |
/** |
|
640 |
* A body subscriber which receives input from an upstream subscriber |
|
641 |
* and maps that subscriber's body type to a new type. The upstream subscriber |
|
642 |
* delegates all flow operations directly to this object. The |
|
643 |
* {@link CompletionStage} returned by {@link #getBody()}} takes the output |
|
644 |
* of the upstream {@code getBody()} and applies the mapper function to |
|
645 |
* obtain the new {@code CompletionStage} type. |
|
646 |
* |
|
647 |
* @param <T> the upstream body type |
|
648 |
* @param <U> this subscriber's body type |
|
649 |
*/ |
|
650 |
public static class MappingSubscriber<T,U> implements BodySubscriber<U> { |
|
651 |
private final BodySubscriber<T> upstream; |
|
652 |
private final Function<? super T,? extends U> mapper; |
|
653 |
||
654 |
public MappingSubscriber(BodySubscriber<T> upstream, |
|
655 |
Function<? super T,? extends U> mapper) { |
|
656 |
this.upstream = Objects.requireNonNull(upstream); |
|
657 |
this.mapper = Objects.requireNonNull(mapper); |
|
658 |
} |
|
659 |
||
660 |
@Override |
|
661 |
public CompletionStage<U> getBody() { |
|
662 |
return upstream.getBody().thenApply(mapper); |
|
663 |
} |
|
664 |
||
665 |
@Override |
|
666 |
public void onSubscribe(Flow.Subscription subscription) { |
|
667 |
upstream.onSubscribe(subscription); |
|
668 |
} |
|
669 |
||
670 |
@Override |
|
671 |
public void onNext(List<ByteBuffer> item) { |
|
672 |
upstream.onNext(item); |
|
673 |
} |
|
674 |
||
675 |
@Override |
|
676 |
public void onError(Throwable throwable) { |
|
677 |
upstream.onError(throwable); |
|
678 |
} |
|
679 |
||
680 |
@Override |
|
681 |
public void onComplete() { |
|
682 |
upstream.onComplete(); |
|
683 |
} |
|
684 |
} |
|
685 |
||
686 |
// A BodySubscriber that returns a Publisher<List<ByteBuffer>> |
|
687 |
static class PublishingBodySubscriber |
|
688 |
implements BodySubscriber<Flow.Publisher<List<ByteBuffer>>> { |
|
689 |
private final MinimalFuture<Flow.Subscription> |
|
690 |
subscriptionCF = new MinimalFuture<>(); |
|
691 |
private final MinimalFuture<SubscriberRef> |
|
692 |
subscribedCF = new MinimalFuture<>(); |
|
693 |
private AtomicReference<SubscriberRef> |
|
694 |
subscriberRef = new AtomicReference<>(); |
|
695 |
private final CompletionStage<Flow.Publisher<List<ByteBuffer>>> body = |
|
696 |
subscriptionCF.thenCompose( |
|
697 |
(s) -> MinimalFuture.completedFuture(this::subscribe)); |
|
698 |
||
699 |
// We use the completionCF to ensure that only one of |
|
700 |
// onError or onComplete is ever called. |
|
701 |
private final MinimalFuture<Void> completionCF; |
|
702 |
private PublishingBodySubscriber() { |
|
703 |
completionCF = new MinimalFuture<>(); |
|
704 |
completionCF.whenComplete( |
|
705 |
(r,t) -> subscribedCF.thenAccept( s -> complete(s, t))); |
|
706 |
} |
|
707 |
||
708 |
// An object that holds a reference to a Flow.Subscriber. |
|
709 |
// The reference is cleared when the subscriber is completed - either |
|
710 |
// normally or exceptionally, or when the subscription is cancelled. |
|
711 |
static final class SubscriberRef { |
|
712 |
volatile Flow.Subscriber<? super List<ByteBuffer>> ref; |
|
713 |
SubscriberRef(Flow.Subscriber<? super List<ByteBuffer>> subscriber) { |
|
714 |
ref = subscriber; |
|
715 |
} |
|
716 |
Flow.Subscriber<? super List<ByteBuffer>> get() { |
|
717 |
return ref; |
|
718 |
} |
|
719 |
Flow.Subscriber<? super List<ByteBuffer>> clear() { |
|
720 |
Flow.Subscriber<? super List<ByteBuffer>> res = ref; |
|
721 |
ref = null; |
|
722 |
return res; |
|
723 |
} |
|
724 |
} |
|
725 |
||
726 |
// A subscription that wraps an upstream subscription and |
|
727 |
// holds a reference to a subscriber. The subscriber reference |
|
728 |
// is cleared when the subscription is cancelled |
|
729 |
final static class SubscriptionRef implements Flow.Subscription { |
|
730 |
final Flow.Subscription subscription; |
|
731 |
final SubscriberRef subscriberRef; |
|
732 |
SubscriptionRef(Flow.Subscription subscription, |
|
733 |
SubscriberRef subscriberRef) { |
|
734 |
this.subscription = subscription; |
|
735 |
this.subscriberRef = subscriberRef; |
|
736 |
} |
|
737 |
@Override |
|
738 |
public void request(long n) { |
|
739 |
if (subscriberRef.get() != null) { |
|
740 |
subscription.request(n); |
|
741 |
} |
|
742 |
} |
|
743 |
@Override |
|
744 |
public void cancel() { |
|
745 |
subscription.cancel(); |
|
746 |
subscriberRef.clear(); |
|
747 |
} |
|
748 |
||
749 |
void subscribe() { |
|
750 |
Subscriber<?> subscriber = subscriberRef.get(); |
|
751 |
if (subscriber != null) { |
|
752 |
subscriber.onSubscribe(this); |
|
753 |
} |
|
754 |
} |
|
755 |
||
756 |
@Override |
|
757 |
public String toString() { |
|
758 |
return "SubscriptionRef/" |
|
759 |
+ subscription.getClass().getName() |
|
760 |
+ "@" |
|
761 |
+ System.identityHashCode(subscription); |
|
762 |
} |
|
763 |
} |
|
764 |
||
765 |
// This is a callback for the subscribedCF. |
|
766 |
// Do not call directly! |
|
767 |
private void complete(SubscriberRef ref, Throwable t) { |
|
768 |
assert ref != null; |
|
769 |
Subscriber<?> s = ref.clear(); |
|
770 |
// maybe null if subscription was cancelled |
|
771 |
if (s == null) return; |
|
772 |
if (t == null) { |
|
773 |
try { |
|
774 |
s.onComplete(); |
|
775 |
} catch (Throwable x) { |
|
776 |
s.onError(x); |
|
777 |
} |
|
778 |
} else { |
|
779 |
s.onError(t); |
|
780 |
} |
|
781 |
} |
|
782 |
||
783 |
private void signalError(Throwable err) { |
|
784 |
if (err == null) { |
|
785 |
err = new NullPointerException("null throwable"); |
|
786 |
} |
|
787 |
completionCF.completeExceptionally(err); |
|
788 |
} |
|
789 |
||
790 |
private void signalComplete() { |
|
791 |
completionCF.complete(null); |
|
792 |
} |
|
793 |
||
794 |
private void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) { |
|
795 |
Objects.requireNonNull(subscriber, "subscriber must not be null"); |
|
796 |
SubscriberRef ref = new SubscriberRef(subscriber); |
|
797 |
if (subscriberRef.compareAndSet(null, ref)) { |
|
798 |
subscriptionCF.thenAccept((s) -> { |
|
799 |
SubscriptionRef subscription = new SubscriptionRef(s,ref); |
|
800 |
try { |
|
801 |
subscription.subscribe(); |
|
802 |
subscribedCF.complete(ref); |
|
803 |
} catch (Throwable t) { |
|
804 |
if (Log.errors()) { |
|
805 |
Log.logError("Failed to call onSubscribe: " + |
|
806 |
"cancelling subscription: " + t); |
|
807 |
Log.logError(t); |
|
808 |
} |
|
809 |
subscription.cancel(); |
|
810 |
} |
|
811 |
}); |
|
812 |
} else { |
|
813 |
subscriber.onSubscribe(new Flow.Subscription() { |
|
814 |
@Override public void request(long n) { } |
|
815 |
@Override public void cancel() { } |
|
816 |
}); |
|
817 |
subscriber.onError(new IllegalStateException( |
|
818 |
"This publisher has already one subscriber")); |
|
819 |
} |
|
820 |
} |
|
821 |
||
822 |
@Override |
|
823 |
public void onSubscribe(Flow.Subscription subscription) { |
|
824 |
subscriptionCF.complete(subscription); |
|
825 |
} |
|
826 |
||
827 |
@Override |
|
828 |
public void onNext(List<ByteBuffer> item) { |
|
829 |
try { |
|
830 |
// cannot be called before onSubscribe() |
|
831 |
assert subscriptionCF.isDone(); |
|
832 |
SubscriberRef ref = subscriberRef.get(); |
|
833 |
// cannot be called before subscriber calls request(1) |
|
834 |
assert ref != null; |
|
835 |
Flow.Subscriber<? super List<ByteBuffer>> |
|
836 |
subscriber = ref.get(); |
|
837 |
if (subscriber != null) { |
|
838 |
// may be null if subscription was cancelled. |
|
839 |
subscriber.onNext(item); |
|
840 |
} |
|
841 |
} catch (Throwable err) { |
|
842 |
signalError(err); |
|
843 |
subscriptionCF.thenAccept(s -> s.cancel()); |
|
844 |
} |
|
845 |
} |
|
846 |
||
847 |
@Override |
|
848 |
public void onError(Throwable throwable) { |
|
849 |
// cannot be called before onSubscribe(); |
|
850 |
assert suppress(subscriptionCF.isDone(), |
|
851 |
"onError called before onSubscribe", |
|
852 |
throwable); |
|
853 |
// onError can be called before request(1), and therefore can |
|
854 |
// be called before subscriberRef is set. |
|
855 |
signalError(throwable); |
|
856 |
} |
|
857 |
||
858 |
@Override |
|
859 |
public void onComplete() { |
|
860 |
// cannot be called before onSubscribe() |
|
861 |
if (!subscriptionCF.isDone()) { |
|
862 |
signalError(new InternalError( |
|
863 |
"onComplete called before onSubscribed")); |
|
864 |
} else { |
|
865 |
// onComplete can be called before request(1), |
|
866 |
// and therefore can be called before subscriberRef |
|
867 |
// is set. |
|
868 |
signalComplete(); |
|
869 |
} |
|
870 |
} |
|
871 |
||
872 |
@Override |
|
873 |
public CompletionStage<Flow.Publisher<List<ByteBuffer>>> getBody() { |
|
874 |
return body; |
|
875 |
} |
|
876 |
||
877 |
private boolean suppress(boolean condition, |
|
878 |
String assertion, |
|
879 |
Throwable carrier) { |
|
880 |
if (!condition) { |
|
881 |
if (carrier != null) { |
|
882 |
carrier.addSuppressed(new AssertionError(assertion)); |
|
883 |
} else if (Log.errors()) { |
|
884 |
Log.logError(new AssertionError(assertion)); |
|
885 |
} |
|
886 |
} |
|
887 |
return true; |
|
888 |
} |
|
889 |
||
890 |
} |
|
891 |
||
892 |
public static BodySubscriber<Flow.Publisher<List<ByteBuffer>>> |
|
893 |
createPublisher() { |
|
894 |
return new PublishingBodySubscriber(); |
|
895 |
} |
|
896 |
||
48083 | 897 |
} |