FixedStreamMessageTest.java

class FixedStreamMessageTest {

    @ArgumentsSource(IntsProvider.class)
    @ParameterizedTest
    void spec_306_requestAfterCancel(List<Integer> nums) throws InterruptedException {
        final Integer[] array = nums.stream().toArray(Integer[]::new);
        final StreamMessage<Integer> message = StreamMessage.of(array);
        final CompletableFuture<Subscription> subscriptionFuture = new CompletableFuture<>();
        message.subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                subscriptionFuture.complete(s);
            }

            @Override
            public void onNext(Integer integer) {}

            @Override
            public void onError(Throwable t) {}

            @Override
            public void onComplete() {}
        }, ImmediateEventExecutor.INSTANCE);

        final Subscription subscription = subscriptionFuture.join();

        subscription.cancel();
        assertThatCode(() -> subscription.request(1)).doesNotThrowAnyException();
    }

    private static class IntsProvider implements ArgumentsProvider {

        @Override
        public Stream<? extends Arguments> provideArguments(ExtensionContext context) throws Exception {
            return Stream.of(ImmutableList.of(),
                             ImmutableList.of(1),
                             ImmutableList.of(1, 2),
                             ImmutableList.of(1, 2, 3),
                             ImmutableList.of(1, 2, 3, 4))
                    .map(Arguments::of);
        }
    }
}