@Test
public void testBuildingSubscriberFromSpec() throws ExecutionException, InterruptedException {
    Processor<ByteBuffer, MyDomainObject> parser = createParser();

    CompletionSubscriber<ByteBuffer, List<MyDomainObject>> subscriber = ReactiveStreams.<ByteBuffer> builder()
            .via(parser)
            .toList()
            .build();

    CompletionStage<List<MyDomainObject>> result = subscriber.getCompletion();

    List<MyDomainObject> domainObjects = Arrays.asList(new MyDomainObject("Clement", "Neo"),
            new MyDomainObject("Tintin", "Milou"));
    Publisher<ByteBuffer> publisher = ReactiveStreams.fromIterable(domainObjects)
            .map(obj -> String.format("%s,%s\n", obj.field1, obj.field2))
            .map(line -> ByteBuffer.wrap(line.getBytes()))
            .buildRs();

    publisher.subscribe(subscriber);
    List<MyDomainObject> objects = result.toCompletableFuture().get();
    assertThat(objects.toString()).contains("Clement => Neo", "Tintin => Milou");
}
 

@Test
public void testValidSubscriber() {
    engine = new Engine();
    CompletionSubscriber<Integer, Optional<Integer>> built = ReactiveStreams.<Integer> builder()
            .map(i -> i + 1)
            .findFirst()
            .build(engine);

    assertThat(built).isNotNull();

    ReactiveStreams.of(5, 4, 3).buildRs().subscribe(built);
    Optional<Integer> integer = built.getCompletion().toCompletableFuture().join();
    assertThat(integer).contains(6);
}
 

@Test
public void completionSubscriberShouldDelegateToSubscriber() {
    Deque<Object> calls = new ArrayDeque<>();
    CompletionSubscriber subscriber = CompletionSubscriber.of(new Subscriber() {
        @Override
        public void onSubscribe(Subscription s) {
            calls.add(s);
        }

        @Override
        public void onNext(Object o) {
            calls.add(o);
        }

        @Override
        public void onError(Throwable t) {
            calls.add(t);
        }

        @Override
        public void onComplete() {
            calls.add("onComplete");
        }
    }, new CompletableFuture<>());

    subscriber.onSubscribe(Mocks.SUBSCRIPTION);
    assertSame(calls.removeFirst(), Mocks.SUBSCRIPTION);
    subscriber.onNext("element");
    assertEquals(calls.removeFirst(), "element");
    Exception e = new Exception();
    subscriber.onError(e);
    assertSame(calls.removeFirst(), e);
    subscriber.onComplete();
    assertEquals(calls.removeFirst(),"onComplete");
    assertTrue(calls.isEmpty());
}
 

@Incoming("sink")
public Subscriber<Message<String>> sink() {
    CompletionSubscriber<Message<String>, List<Message<String>>> subscriber = ReactiveStreams.<Message<String>> builder()
            .toList().build();
    subscriber.getCompletion().thenAccept(result::addAll);
    return subscriber;
}
 

@Test
public void testValidSubscriber() {
    engine = new Engine();
    CompletionSubscriber<Integer, Optional<Integer>> built = ReactiveStreams.<Integer> builder()
            .map(i -> i + 1)
            .findFirst()
            .build(engine);

    assertThat(built).isNotNull();

    ReactiveStreams.of(5, 4, 3).buildRs().subscribe(built);
    Optional<Integer> integer = built.getCompletion().toCompletableFuture().join();
    assertThat(integer).contains(6);
}
 

public DefaultSubscriberWithCompletionStage(Processor<T, T> processor, CompletionStage<R> result) {
    subscriber = CompletionSubscriber.of(processor, result);
}
 

@Test
public void completionSubscriberShouldReturnSameCompletionStage() {
    CompletableFuture future = new CompletableFuture();
    assertSame(CompletionSubscriber.of(Mocks.SUBSCRIBER, future).getCompletion(), future);
}
 

@Test(expectedExceptions = NullPointerException.class)
public void completionSubscriberShouldNotAcceptNullSubscriber() {
    CompletionSubscriber.of(null, new CompletableFuture<>());
}
 

@Test(expectedExceptions = NullPointerException.class)
public void completionSubscriberShouldNotAcceptNullCompletionStage() {
    CompletionSubscriber.of(Mocks.SUBSCRIBER, null);
}
 

@Override
public CompletionSubscriber<T, R> build() {
    return build(ReactiveStreamsEngineResolver.instance());
}
 

@Override
public CompletionSubscriber<T, R> build(ReactiveStreamsEngine engine) {
    Objects.requireNonNull(engine, "Engine must not be null");
    SubscriberWithCompletionStage<T, R> subscriberWithCompletionStage = engine.buildSubscriber(toGraph());
    return CompletionSubscriber.of(subscriberWithCompletionStage.getSubscriber(), subscriberWithCompletionStage.getCompletion());
}
 

public DefaultSubscriberWithCompletionStage(Processor<T, T> processor, CompletionStage<R> result) {
    subscriber = CompletionSubscriber.of(processor, result);
}