### Question:
ReadableByteChannelPublisher implements Flow.Publisher<DataChunk> { LazyValue<ScheduledExecutorService> executor() { return executor; } @Deprecated(since = "2.0.0", forRemoval = true)  ReadableByteChannelPublisher(ReadableByteChannel channel, RetrySchema retrySchema); @Override void subscribe(Flow.Subscriber<? super DataChunk> subscriberParam);  }### Answer:
@Test void chunky() throws Exception { PeriodicalChannel pc = createChannelWithNoAvailableData(25, 3); ReadableByteChannelPublisher publisher = new ReadableByteChannelPublisher(pc, RetrySchema.constant(2)); byte[] bytes = ContentReaders.readBytes(publisher).get(5, TimeUnit.SECONDS); assertThat(bytes.length, is(TEST_DATA_SIZE)); assertByteSequence(bytes); assertThat(pc.threads.size(), is(2)); assertThat(pc.isOpen(), is(false)); LazyValue<ScheduledExecutorService> executor = publisher.executor(); assertThat("Executor should have been used", executor.isLoaded(), is(true)); assertThat("Executor should have been shut down", executor.get().isShutdown(), is(true)); }
@Test void onClosedChannel() throws Exception { PeriodicalChannel pc = new PeriodicalChannel(i -> 1024, TEST_DATA_SIZE); pc.close(); ReadableByteChannelPublisher publisher = new ReadableByteChannelPublisher(pc, RetrySchema.constant(0)); try { ContentReaders.readBytes(publisher).get(5, TimeUnit.SECONDS); fail("Did not throw expected ExecutionException!"); } catch (ExecutionException e) { assertThat(e.getCause(), instanceOf(ClosedChannelException.class)); } LazyValue<ScheduledExecutorService> executor = publisher.executor(); assertThat("Executor should have not been used", executor.isLoaded(), is(false)); }
@Test @Disabled("This test uses a sleep, so could cause issues on slow environments") void onClosedInProgress() throws Exception { PeriodicalChannel pc = createChannelWithNoAvailableData(5, 2); RetrySchema schema = RetrySchema.constant(TimeUnit.SECONDS.toMillis(2)); ReadableByteChannelPublisher publisher = new ReadableByteChannelPublisher(pc, schema); Single<byte[]> data = ContentReaders.readBytes(publisher); data.thenRun(() -> { }); Thread.sleep(1000); pc.close(); CompletionException c = assertThrows(CompletionException.class, () -> data.await(5, TimeUnit.SECONDS)); assertThat(c.getCause(), instanceOf(ClosedChannelException.class)); LazyValue<ScheduledExecutorService> executor = publisher.executor(); assertThat("Executor should have been used", executor.isLoaded(), is(true)); assertThat("Executor should have been shut down", executor.get().isShutdown(), is(true)); }