@Test public void testNoClosingBoundary(){ String boundary = "boundary"; final byte[] chunk1 = ("--" + boundary + "\n" + "Content-Type: text/xml; charset=UTF-8\n" + "Content-Id: part1\n" + "\n" + "<foo>bar</foo>\n").getBytes(); DataChunkSubscriber s1 = new DataChunkSubscriber(); BodyPartSubscriber testSubscriber = new BodyPartSubscriber(SUBSCRIBER_TYPE.ONE_BY_ONE, p -> p.content().subscribe(s1)); partsPublisher(boundary, chunk1).subscribe(testSubscriber); try { s1.future.orTimeout(100, TimeUnit.MILLISECONDS).join(); throw new IllegalStateException("Should have terminated exceptionally"); } catch(CompletionException e) { Throwable error = e.getCause(); assertThat(error.getClass(), is(equalTo(MimeParser.ParsingException.class))); assertThat(error.getMessage(), is(equalTo("No closing MIME boundary"))); } try { testSubscriber.complete.orTimeout(200, TimeUnit.MILLISECONDS).join(); throw new IllegalStateException("Not expecting to terminate normally"); } catch(CompletionException e) { Throwable error = e.getCause(); assertThat(error, is(notNullValue())); assertThat(error.getClass(), is(equalTo(MimeParser.ParsingException.class))); assertThat(error.getMessage(), is(equalTo("No closing MIME boundary"))); } }
@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();  }