package reactor.pipe;

import reactor.pipe.key.Key;

import java.util.List;
import reactor.fn.UnaryOperator;

import reactor.fn.Consumer;
import reactor.fn.Function;
import reactor.fn.Predicate;


public class AnonymousPipe<V> {

  private final Key       rootKey;
  private final Key       upstream;
  private final NamedPipe pipe;

  public AnonymousPipe(Key upstream, NamedPipe pipe) {
    this.upstream = upstream;
    this.rootKey = upstream;
    this.pipe = pipe;
  }

  public AnonymousPipe(Key rootKey, Key upstream, NamedPipe pipe) {
    this.upstream = upstream;
    this.pipe = pipe;
    this.rootKey = rootKey;
  }

  @SuppressWarnings(value = {"unchecked"})
  public <V1> AnonymousPipe<V1> map(Function<V, V1> mapper) {
    Key downstream = upstream.derive();

    pipe.map(upstream, downstream, mapper);

    return new AnonymousPipe<>(rootKey, downstream, pipe);
  }

  @SuppressWarnings(value = {"unchecked"})
  public AnonymousPipe<V> filter(Predicate<V> predicate) {
    Key downstream = upstream.derive();

    pipe.filter(upstream, downstream, predicate);

    return new AnonymousPipe<>(rootKey, downstream, pipe);
  }

  @SuppressWarnings(value = {"unchecked"})
  public AnonymousPipe<List<V>> slide(UnaryOperator<List<V>> drop) {
    Key downstream = upstream.derive();

    pipe.slide(upstream, downstream, drop);

    return new AnonymousPipe<>(rootKey, downstream, pipe);
  }

  @SuppressWarnings(value = {"unchecked"})
  public AnonymousPipe<List<V>> partition(Predicate<List<V>> emit) {
    Key downstream = upstream.derive();

    pipe.partition(upstream, downstream, emit);

    return new AnonymousPipe<>(rootKey, downstream, pipe);
  }

  @SuppressWarnings(value = {"unchecked"})
  public AnonymousPipe<V> consume(Consumer<V> consumer) {
    pipe.consume(upstream, consumer);
    return this;
  }

  @SuppressWarnings(value = {"unchecked"})
  public AnonymousPipe<V> redirect(Key destination) {
    pipe.consume(upstream, new KeyedConsumer<Key, V>() {
      @Override
      public void accept(Key key, V value) {
        pipe.notify(destination, value);
      }
    });

    return this;
  }

  @SuppressWarnings(value = {"unchecked"})
  public Runnable cancellableConsumer(Consumer<V> consumer) {
    pipe.consume(upstream, consumer);
    return () -> {
      pipe.unregister(upstream);
    };
  }

  @SuppressWarnings(value = {"unchecked"})
  public void notify(V v) {
    this.pipe.notify(upstream, v);
  }

  public void unregister() {
    this.pipe.unregister(new Predicate<Key>() {
      @Override
      public boolean test(Key k) {
        return k.isDerivedFrom(rootKey) || k.equals(rootKey);
      }
    });
  }
}

@@ -35,163 +35,102 @@
 tor.
-pipe.key.Key;%0A%0Aimport java.util.List;%0Aimport reactor.fn.UnaryOperator;%0A%0Aimport reactor.fn.Consumer;%0Aimport reactor.fn.Function;%0Aimport reactor.fn.Predicate
+fn.*;%0Aimport reactor.pipe.concurrent.Atom;%0Aimport reactor.pipe.key.Key;%0A%0Aimport java.util.List
 ;%0A%0A%0A
@@ -807,32 +807,358 @@
 am, pipe);%0A  %7D%0A%0A
+  @SuppressWarnings(value = %7B%22unchecked%22%7D)%0A  public %3CV1, ST%3E AnonymousPipe%3CV1%3E map(BiFunction%3CAtom%3CST%3E, V, V1%3E mapper,%0A                                        ST init) %7B%0A    Key downstream = upstream.derive();%0A%0A    pipe.map(upstream, downstream, mapper, init);%0A%0A    return new AnonymousPipe%3C%3E(rootKey, downstream, pipe);%0A  %7D%0A%0A
   @SuppressWarni

Add stateful mapper to anonymous pipe