### Question:
MultiFromByteChannel implements Multi<ByteBuffer> { LazyValue<ScheduledExecutorService> executor() { return executor; } MultiFromByteChannel(IoMulti.MultiFromByteChannelBuilder builder); @Override void subscribe(Flow.Subscriber<? super ByteBuffer> subscriberParam);  }### Answer:
@Test void testChunky() { PeriodicalChannel pc = createChannelWithNoAvailableData(25, 3); Multi<ByteBuffer> publisher = IoMulti.multiFromByteChannelBuilder(pc) .retrySchema(RetrySchema.constant(2)) .build(); byte[] bytes = Multi.create(publisher) .collect(new BufferCollector()) .await(5, TimeUnit.SECONDS); assertThat(bytes.length, is(TEST_DATA_SIZE)); assertByteSequence(bytes); assertThat(pc.threads.size(), is(2)); assertThat(pc.isOpen(), is(false)); MultiFromByteChannel multi = (MultiFromByteChannel) publisher; LazyValue<ScheduledExecutorService> executor = multi.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 testChunkyWithCustomExecutor() { ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1); PeriodicalChannel pc = createChannelWithNoAvailableData(25, 3); Multi<ByteBuffer> publisher = IoMulti.multiFromByteChannelBuilder(pc) .retrySchema(RetrySchema.constant(2)) .executor(executorService) .build(); byte[] bytes = Multi.create(publisher) .collect(new BufferCollector()) .await(5, TimeUnit.SECONDS); assertThat(bytes.length, is(TEST_DATA_SIZE)); assertByteSequence(bytes); assertThat(pc.threads.size(), is(2)); assertThat(pc.isOpen(), is(false)); MultiFromByteChannel multi = (MultiFromByteChannel) publisher; LazyValue<ScheduledExecutorService> executor = multi.executor(); assertThat("Executor was provided", executor.isLoaded(), is(true)); ScheduledExecutorService usedExecutor = executor.get(); assertThat("Executor should not have been shut down, as we use an explicit one", usedExecutor.isShutdown(), is(false)); assertThat("Executor should have been our instance", usedExecutor, is(sameInstance(executorService))); executorService.shutdownNow(); }
@Test void testOnClosedChannel() { PeriodicalChannel pc = new PeriodicalChannel(i -> 1024, TEST_DATA_SIZE); pc.close(); Multi<ByteBuffer> publisher = IoMulti.multiFromByteChannelBuilder(pc) .retrySchema(RetrySchema.constant(0)) .build(); try { Multi.create(publisher) .collect(new BufferCollector()) .await(5, TimeUnit.SECONDS); fail("Did not throw expected CompletionException!"); } catch (CompletionException e) { assertThat(e.getCause(), instanceOf(ClosedChannelException.class)); } MultiFromByteChannel multi = (MultiFromByteChannel) publisher; LazyValue<ScheduledExecutorService> executor = multi.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 testOnClosedInProgress() throws Exception { PeriodicalChannel pc = createChannelWithNoAvailableData(5, 2); Multi<ByteBuffer> publisher = IoMulti.multiFromByteChannelBuilder(pc) .retrySchema(RetrySchema.constant(TimeUnit.SECONDS.toMillis(2))) .build(); Single<byte[]> data = Multi.create(publisher) .collect(new BufferCollector()); data.thenRun(() -> { }); Thread.sleep(1000); pc.close(); CompletionException c = assertThrows(CompletionException.class, () -> data.await(5, TimeUnit.SECONDS)); assertThat(c.getCause(), instanceOf(ClosedChannelException.class)); MultiFromByteChannel multi = (MultiFromByteChannel) publisher; LazyValue<ScheduledExecutorService> executor = multi.executor(); assertThat("Executor should have been used", executor.isLoaded(), is(true)); assertThat("Executor should have been shut down", executor.get().isShutdown(), is(true)); }