### Question:
Pipeline implements SuspendableCallable<Long> { public Pipeline(final ReceivePort<? extends S> from, final SendPort<? super T> to, final SuspendableAction2<S, Channel<T>> transformer, final int parallelism, boolean closeTo, final SuspendableCallable<Channel<T>> resultChannelBuilder, final StrandFactory strandFactory) { this.from = from; this.to = to; this.transformer = transformer; this.parallelism = parallelism <= 0 ? 0 : parallelism; this.jobs = Channels.newChannel(this.parallelism, Channels.OverflowPolicy.BLOCK, true, false); this.results = Channels.newChannel(this.parallelism, Channels.OverflowPolicy.BLOCK, false, true); this.closeTo = closeTo; this.resultChannelBuilder = resultChannelBuilder != null ? resultChannelBuilder : resultChannelBuilderDefault; this.strandFactory = strandFactory; } Pipeline(final ReceivePort<? extends S> from, final SendPort<? super T> to, final SuspendableAction2<S, Channel<T>> transformer, final int parallelism, boolean closeTo, final SuspendableCallable<Channel<T>> resultChannelBuilder, final StrandFactory strandFactory);  Pipeline(final ReceivePort<? extends S> from, final SendPort<? super T> to, final SuspendableAction2<S, Channel<T>> transformer, final int parallelism, boolean closeTo, final SuspendableCallable<Channel<T>> resultChannelBuilder);  Pipeline(final ReceivePort<? extends S> from, final SendPort<? super T> to, final SuspendableAction2<S, Channel<T>> transformer, final int parallelism, boolean closeTo);  Pipeline(final ReceivePort<? extends S> from, final SendPort<? super T> to, final SuspendableAction2<S, Channel<T>> transformer, final int parallelism);  Pipeline(final ReceivePort<? extends S> from, final SendPort<? super T> to, final SuspendableAction2<S, Channel<T>> transformer); long getTransferred(); @Override Long run();  }### Answer:
@Test public void testPipeline() throws Exception { final Channel<Integer> i = newChannel(); final Channel<Integer> o = newChannel(); final Pipeline<Integer, Integer> p = new Pipeline<>( i, o, new SuspendableAction2<Integer, Channel<Integer>>() { @Override public void call(final Integer i, final Channel<Integer> out) throws SuspendExecution, InterruptedException { out.send(i + 1); out.close(); } }, parallelism); final Fiber<Long> pf = new Fiber("pipeline", scheduler, p).start(); final Fiber receiver = new Fiber("receiver", scheduler, new SuspendableRunnable() { @Override public void run() throws SuspendExecution, InterruptedException { final Integer m1 = o.receive(); final Integer m2 = o.receive(); final Integer m3 = o.receive(); final Integer m4 = o.receive(); assertThat(m1, notNullValue()); assertThat(m2, notNullValue()); assertThat(m3, notNullValue()); assertThat(m4, notNullValue()); assertThat(ImmutableSet.of(m1, m2, m3, m4), equalTo(ImmutableSet.of(2, 3, 4, 5))); try { pf.join(); } catch (ExecutionException ex) { throw new AssertionError(ex); } assertNull(o.tryReceive()); assertTrue(o.isClosed()); } }).start(); i.send(1); i.send(2); i.send(3); i.send(4); i.close(); long transferred = pf.get(); assertThat(transferred, equalTo(p.getTransferred())); assertThat(transferred, equalTo(4l)); receiver.join(); }