@Test public void testPartContentSubscriberThrottling() { String boundary = "boundary"; final byte[] chunk1 = ("--" + boundary + "\n" + "Content-Id: part1\n" + "\n" + "body 1.aaaa\n").getBytes(); final byte[] chunk2 = "body 1.bbbb\n".getBytes(); final byte[] chunk3 = ("body 1.cccc\n" + "--" + boundary + "\n" + "Content-Id: part2\n" + "\n" + "This is the 2nd").getBytes(); final byte[] chunk4 = ("body.\n" + "--" + boundary + "--").getBytes(); final CountDownLatch latch = new CountDownLatch(3); Consumer<BodyPart> consumer = (part) -> { latch.countDown(); if (latch.getCount() == 2) { assertThat(part.headers().values("Content-Id"), hasItems("part1")); } part.content().subscribe(new Subscriber<DataChunk>() { @Override public void onSubscribe(Subscription subscription) { subscription.request(1); } @Override public void onNext(DataChunk item) { latch.countDown(); } @Override public void onError(Throwable throwable) { } @Override public void onComplete() { } }); }; BodyPartSubscriber testSubscriber = new BodyPartSubscriber(SUBSCRIBER_TYPE.ONE_BY_ONE, consumer); partsPublisher(boundary, List.of(chunk1, chunk2, chunk3, chunk4)).subscribe(testSubscriber); waitOnLatchNegative(latch, "the 2nd part should not be processed"); assertThat(latch.getCount(), is(equalTo(1L))); try { testSubscriber.complete.orTimeout(200, TimeUnit.MILLISECONDS).join(); throw new IllegalStateException("Not expecting to make progress, unless the part is consumed"); } catch(CompletionException e) { Throwable error = e.getCause(); assertThat(error.getClass(), is(equalTo(TimeoutException.class))); } }
@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();  }