@@ -0,0 +1,1294 @@
+package reactor.pipe.flow;%0A%0Aimport org.junit.Test;%0Aimport org.reactivestreams.Publisher;%0Aimport reactor.Publishers;%0Aimport reactor.fn.tuple.Tuple;%0Aimport reactor.fn.tuple.Tuple2;%0Aimport reactor.pipe.AbstractFirehoseTest;%0Aimport reactor.pipe.Pipe;%0Aimport reactor.pipe.concurrent.AVar;%0Aimport reactor.pipe.key.Key;%0A%0Aimport java.util.Arrays;%0Aimport java.util.List;%0A%0Aimport static org.hamcrest.core.Is.is;%0Aimport static org.junit.Assert.assertThat;%0A%0Apublic class LocalFlowBuilderTest extends AbstractFirehoseTest %7B%0A%0A  @Test%0A  public void upDownstreamTest() throws InterruptedException %7B%0A    AVar%3CInteger%3E result = new AVar%3C%3E(1);%0A%0A    FlowBuilder localFlowBuilder = new LocalFlowBuilder(firehose);%0A%0A    List%3CTuple2%3CString, Integer%3E%3E l = Arrays.asList(Tuple.of(%22source%22, 1),%0A                                                    Tuple.of(%22source%22, 2));%0A    Publisher%3CTuple2%3CString, Integer%3E%3E p = Publishers.from(l);%0A%0A    localFlowBuilder.%3CString, Integer%3Eflow(%22local flow%22)%0A      .upstream((String k, Integer v) -%3E Key.wrap(k),%0A                p)%0A      .subscribe(Key.wrap(%22source%22),%0A                 Pipe.%3CInteger%3Ebuild().map((Integer i) -%3E i + 1))%0A      .downstream(result::set);%0A%0A    localFlowBuilder.start();%0A    assertThat(result.get(LATCH_TIMEOUT, LATCH_TIME_UNIT),%0A               is(2));%0A  %7D%0A%0A%0A%7D

Add flow builder tests.