@Test public void testMultiple() throws IOException { String oneValue = "someBytes"; SubmissionPublisher<ByteBuffer> publisher = new SubmissionPublisher<>(); SubscriberInputStream is = new SubscriberInputStream(); publisher.subscribe(is); new Thread(() -> { for (int i = 0; i < 10; i++) { publisher.submit(ByteBuffer.wrap(oneValue.getBytes(StandardCharsets.UTF_8))); } publisher.close(); }).start(); byte[] buffer = new byte[16]; int len; ByteArrayOutputStream baos = new ByteArrayOutputStream(); while ((len = is.read(buffer)) != -1) { baos.write(buffer, 0, len); } String result = new String(baos.toByteArray(), StandardCharsets.UTF_8); StringBuilder expectedResult = new StringBuilder(); for (int i = 0; i < 10; i++) { expectedResult.append(oneValue); } assertThat(result, is(expectedResult.toString())); }
@Override public int read() throws IOException { try { while (true) { final ByteBuffer currentBuffer = processed.get(); if (currentBuffer != null) { if (currentBuffer.remaining() > 0) { return currentBuffer.get(); } else { if (reinitializeFuture()) { subscription.request(1); } else { return -1; } } } else { return -1; } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } catch (ExecutionException e) { throw new IOException(e.getCause()); } }
SubscriberInputStream extends InputStream implements Flow.Subscriber<ByteBuffer> { @Override public int read() throws IOException { try { while (true) { final ByteBuffer currentBuffer = processed.get(); if (currentBuffer != null) { if (currentBuffer.remaining() > 0) { return currentBuffer.get(); } else { if (reinitializeFuture()) { subscription.request(1); } else { return -1; } } } else { return -1; } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } catch (ExecutionException e) { throw new IOException(e.getCause()); } } }
SubscriberInputStream extends InputStream implements Flow.Subscriber<ByteBuffer> { @Override public int read() throws IOException { try { while (true) { final ByteBuffer currentBuffer = processed.get(); if (currentBuffer != null) { if (currentBuffer.remaining() > 0) { return currentBuffer.get(); } else { if (reinitializeFuture()) { subscription.request(1); } else { return -1; } } } else { return -1; } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } catch (ExecutionException e) { throw new IOException(e.getCause()); } }  }
SubscriberInputStream extends InputStream implements Flow.Subscriber<ByteBuffer> { @Override public int read() throws IOException { try { while (true) { final ByteBuffer currentBuffer = processed.get(); if (currentBuffer != null) { if (currentBuffer.remaining() > 0) { return currentBuffer.get(); } else { if (reinitializeFuture()) { subscription.request(1); } else { return -1; } } } else { return -1; } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } catch (ExecutionException e) { throw new IOException(e.getCause()); } }  @Override int read(); @Override void onSubscribe(Flow.Subscription subscription); @Override void onNext(ByteBuffer item); @Override synchronized void onError(Throwable throwable); @Override synchronized void onComplete(); }
SubscriberInputStream extends InputStream implements Flow.Subscriber<ByteBuffer> { @Override public int read() throws IOException { try { while (true) { final ByteBuffer currentBuffer = processed.get(); if (currentBuffer != null) { if (currentBuffer.remaining() > 0) { return currentBuffer.get(); } else { if (reinitializeFuture()) { subscription.request(1); } else { return -1; } } } else { return -1; } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } catch (ExecutionException e) { throw new IOException(e.getCause()); } }  @Override int read(); @Override void onSubscribe(Flow.Subscription subscription); @Override void onNext(ByteBuffer item); @Override synchronized void onError(Throwable throwable); @Override synchronized void onComplete();  }