@Test public void testContentAcrossChunksAsyncRequest() { String boundary = "boundary"; final byte[] chunk1 = ("--" + boundary + "\n" + "Content-Id: part1\n" + "\n" + "thi").getBytes(); final byte[] chunk11 = ("s-is-the-1st-slice-of-the-body\n").getBytes(); final byte[] chunk12 = ("t").getBytes(); final byte[] chunk2 = ("his-is-the-2nd-slice-of-the-body\n" + "--" + boundary + "--").getBytes(); final CountDownLatch latch = new CountDownLatch(2); Consumer<BodyPart> consumer = (part) -> { latch.countDown(); assertThat(part.headers().values("Content-Id"), hasItems("part1")); DataChunkSubscriber subscriber = new DataChunkSubscriber(); part.content().subscribe(subscriber); subscriber.content().thenAccept(body -> { assertThat(body, is(equalTo( "this-is-the-1st-slice-of-the-body\n" + "this-is-the-2nd-slice-of-the-body"))); }).exceptionally((ex) -> { System.out.println("UH-OH... " + ex); return null; }).thenAccept((_i) -> { latch.countDown(); }); }; BodyPartSubscriber testSubscriber = new BodyPartSubscriber(SUBSCRIBER_TYPE.ONE_BY_ONE, consumer); partsPublisher(boundary, List.of(chunk1, chunk11, chunk12, chunk2)).subscribe(testSubscriber); try { boolean b = testSubscriber.complete.orTimeout(200, TimeUnit.MILLISECONDS).join(); assertThat(b, is(equalTo(true))); } catch(CompletionException error) { assertThat(error, is(nullValue())); } waitOnLatch(latch); }
@Override public void subscribe(Subscriber<? super ReadableBodyPart> subscriber) { Objects.requireNonNull(subscriber); if (!halfInit(UPSTREAM_INIT)) { Multi.<ReadableBodyPart>error(new IllegalStateException("Only one Subscriber allowed")) .subscribe(subscriber); return; } this.downstream = subscriber; downstream.onSubscribe(new Subscription() { @Override public void request(long n) { long curr = n <= 0 ? partsRequested.getAndSet(-1) : partsRequested.getAndUpdate(v -> Long.MAX_VALUE - v > n ? v + n : v < 0 ? v : Long.MAX_VALUE); if (curr == 0) { drain(); } } @Override public void cancel() { cancelled = true; if (partsRequested.getAndSet(-1) == 0) { drain(); } } }); deferredInit(); }
MultiPartDecoder implements Processor<DataChunk, ReadableBodyPart> { @Override public void subscribe(Subscriber<? super ReadableBodyPart> subscriber) { Objects.requireNonNull(subscriber); if (!halfInit(UPSTREAM_INIT)) { Multi.<ReadableBodyPart>error(new IllegalStateException("Only one Subscriber allowed")) .subscribe(subscriber); return; } this.downstream = subscriber; downstream.onSubscribe(new Subscription() { @Override public void request(long n) { long curr = n <= 0 ? partsRequested.getAndSet(-1) : partsRequested.getAndUpdate(v -> Long.MAX_VALUE - v > n ? v + n : v < 0 ? v : Long.MAX_VALUE); if (curr == 0) { drain(); } } @Override public void cancel() { cancelled = true; if (partsRequested.getAndSet(-1) == 0) { drain(); } } }); deferredInit(); } }
MultiPartDecoder implements Processor<DataChunk, ReadableBodyPart> { @Override public void subscribe(Subscriber<? super ReadableBodyPart> subscriber) { Objects.requireNonNull(subscriber); if (!halfInit(UPSTREAM_INIT)) { Multi.<ReadableBodyPart>error(new IllegalStateException("Only one Subscriber allowed")) .subscribe(subscriber); return; } this.downstream = subscriber; downstream.onSubscribe(new Subscription() { @Override public void request(long n) { long curr = n <= 0 ? partsRequested.getAndSet(-1) : partsRequested.getAndUpdate(v -> Long.MAX_VALUE - v > n ? v + n : v < 0 ? v : Long.MAX_VALUE); if (curr == 0) { drain(); } } @Override public void cancel() { cancelled = true; if (partsRequested.getAndSet(-1) == 0) { drain(); } } }); deferredInit(); } MultiPartDecoder(String boundary, MessageBodyReaderContext context); }
MultiPartDecoder implements Processor<DataChunk, ReadableBodyPart> { @Override public void subscribe(Subscriber<? super ReadableBodyPart> subscriber) { Objects.requireNonNull(subscriber); if (!halfInit(UPSTREAM_INIT)) { Multi.<ReadableBodyPart>error(new IllegalStateException("Only one Subscriber allowed")) .subscribe(subscriber); return; } this.downstream = subscriber; downstream.onSubscribe(new Subscription() { @Override public void request(long n) { long curr = n <= 0 ? partsRequested.getAndSet(-1) : partsRequested.getAndUpdate(v -> Long.MAX_VALUE - v > n ? v + n : v < 0 ? v : Long.MAX_VALUE); if (curr == 0) { drain(); } } @Override public void cancel() { cancelled = true; if (partsRequested.getAndSet(-1) == 0) { drain(); } } }); deferredInit(); } MultiPartDecoder(String boundary, MessageBodyReaderContext context); static MultiPartDecoder create(String boundary, MessageBodyReaderContext context); @Override void subscribe(Subscriber<? super ReadableBodyPart> subscriber); @Override void onSubscribe(Subscription subscription); @Override void onNext(DataChunk chunk); @Override void onError(Throwable throwable); @Override void onComplete(); }
MultiPartDecoder implements Processor<DataChunk, ReadableBodyPart> { @Override public void subscribe(Subscriber<? super ReadableBodyPart> subscriber) { Objects.requireNonNull(subscriber); if (!halfInit(UPSTREAM_INIT)) { Multi.<ReadableBodyPart>error(new IllegalStateException("Only one Subscriber allowed")) .subscribe(subscriber); return; } this.downstream = subscriber; downstream.onSubscribe(new Subscription() { @Override public void request(long n) { long curr = n <= 0 ? partsRequested.getAndSet(-1) : partsRequested.getAndUpdate(v -> Long.MAX_VALUE - v > n ? v + n : v < 0 ? v : Long.MAX_VALUE); if (curr == 0) { drain(); } } @Override public void cancel() { cancelled = true; if (partsRequested.getAndSet(-1) == 0) { drain(); } } }); deferredInit(); } MultiPartDecoder(String boundary, MessageBodyReaderContext context); static MultiPartDecoder create(String boundary, MessageBodyReaderContext context); @Override void subscribe(Subscriber<? super ReadableBodyPart> subscriber); @Override void onSubscribe(Subscription subscription); @Override void onNext(DataChunk chunk); @Override void onError(Throwable throwable); @Override void onComplete();  }