package reactor.pipe;

import reactor.fn.*;
import reactor.pipe.concurrent.Atom;
import reactor.pipe.key.Key;
import reactor.pipe.operation.PartitionOperation;
import reactor.pipe.operation.SlidingWindowOperation;
import org.pcollections.PVector;
import org.pcollections.TreePVector;

import javax.lang.model.type.NullType;
import java.util.LinkedList;
import java.util.List;

public class MatchedPipe<V> extends FinalizedMatchedStream<V> {

  MatchedPipe() {
    super(new LinkedList<>());
  }

  protected MatchedPipe(List<MatchedPipe.StreamSupplier> suppliers) {
    super(suppliers);
  }

  @SuppressWarnings(value = {"unchecked"})
  public <SRC extends Key, V1> MatchedPipe<V1> map(Function<V, V1> mapper) {
    this.suppliers.add(new StreamSupplier<SRC, V, V1>() {
      @Override
      public <DST extends Key> KeyedConsumer<SRC, V> get(SRC src,
                                                         DST dst,
                                                         NamedPipe<V1> pipe) {
        return (key, value) -> {
          pipe.notify(dst.clone(key), mapper.apply(value));
        };
      }
    });
    return new MatchedPipe<>(suppliers);
  }

  @SuppressWarnings(value = {"unchecked"})
  public <SRC extends Key, ST, V1> MatchedPipe<V1> map(BiFunction<Atom<ST>, V, V1> mapper,
                                                       ST init) {
    this.suppliers.add(new StreamSupplier<SRC, V, V1>() {
      @Override
      public <DST extends Key> KeyedConsumer<SRC, V> get(SRC src,
                                                         DST dst,
                                                         NamedPipe<V1> pipe) {
        Atom<ST> st = pipe.stateProvider().makeAtom(src, init);

        return (key, value) -> {
          pipe.notify(dst.clone(key), mapper.apply(st, value));
        };
      }
    });
    return new MatchedPipe<>(suppliers);
  }

  @SuppressWarnings(value = {"unchecked"})
  public <SRC extends Key> MatchedPipe<V> filter(Predicate<V> predicate) {
    this.suppliers.add(new StreamSupplier<SRC, V, V>() {
      @Override
      public <DST extends Key> KeyedConsumer<SRC, V> get(SRC src,
                                                         DST dst,
                                                         NamedPipe<V> pipe) {
        return (key, value) -> {
          if (predicate.test(value)) {
            pipe.notify(dst.clone(key), value);
          }
        };
      }
    });

    return new MatchedPipe<>(suppliers);
  }

  @SuppressWarnings(value = {"unchecked"})
  public <SRC extends Key> MatchedPipe<List<V>> slide(UnaryOperator<List<V>> drop) {
    this.suppliers.add(new StreamSupplier<SRC, V, V>() {
      @Override
      public <DST extends Key> KeyedConsumer<SRC, V> get(SRC src,
                                                         DST dst,
                                                         NamedPipe<V> pipe) {
        Atom<PVector<V>> buffer = pipe.stateProvider().makeAtom(src, TreePVector.empty());

        return new SlidingWindowOperation<SRC, DST, V>(pipe.firehose(),
                                                       buffer,
                                                       drop,
                                                       dst);
      }
    });

    return new MatchedPipe<>(suppliers);
  }

  @SuppressWarnings(value = {"unchecked"})
  public <SRC extends Key> MatchedPipe<List<V>> partition(Predicate<List<V>> emit) {
    this.suppliers.add(new StreamSupplier<SRC, V, V>() {
      @Override
      public <DST extends Key> KeyedConsumer<SRC, V> get(SRC src,
                                                         DST dst,
                                                         NamedPipe<V> pipe) {
        Atom<PVector<V>> buffer = pipe.stateProvider().makeAtom(src, TreePVector.empty());

        return new PartitionOperation<SRC, DST, V>(pipe.firehose(),
                                                   buffer,
                                                   emit,
                                                   dst);
      }
    });

    return new MatchedPipe<>(suppliers);
  }

  @SuppressWarnings(value = {"unchecked"})
  public <SRC extends Key> FinalizedMatchedStream consume(KeyedConsumer<SRC, V> consumer) {
    this.suppliers.add(new StreamSupplier<SRC, V, NullType>() {

      @Override
      public <DST extends Key> KeyedConsumer<SRC, V> get(SRC src,
                                                         DST dst,
                                                         NamedPipe<NullType> pipe) {
        return consumer;
      }
    });
    return new FinalizedMatchedStream(suppliers);
  }

  @SuppressWarnings(value = {"unchecked"})
  public <SRC extends Key> FinalizedMatchedStream consume(Consumer<V> consumer) {
    this.suppliers.add(new StreamSupplier<SRC, V, NullType>() {
      @Override
      public <DST extends Key> KeyedConsumer<SRC, V> get(SRC src,
                                                         DST dst,
                                                         NamedPipe<NullType> pipe) {
        return (key, value) -> consumer.accept(value);
      }
    });
    return new FinalizedMatchedStream(suppliers);
  }

  @FunctionalInterface
  public static interface StreamSupplier<SRC, V, V1> {
    public <DST extends Key> KeyedConsumer<SRC, V> get(SRC src,
                                                       DST dst,
                                                       NamedPipe<V1> pipe);
  }


}

@@ -1149,32 +1149,664 @@
 uppliers);%0A  %7D%0A%0A
+  @SuppressWarnings(value = %7B%22unchecked%22%7D)%0A  public %3CSRC extends Key, V1%3E MatchedPipe%3CV1%3E map(Supplier%3CFunction%3CV, V1%3E%3E supplier) %7B%0A    this.suppliers.add(new StreamSupplier%3CSRC, V, V1%3E() %7B%0A      @Override%0A      public %3CDST extends Key%3E KeyedConsumer%3CSRC, V%3E get(SRC src,%0A                                                         DST dst,%0A                                                         NamedPipe%3CV1%3E pipe) %7B%0A        Function%3CV, V1%3E mapper = supplier.get();%0A        return (key, value) -%3E %7B%0A          pipe.notify(dst.clone(key), mapper.apply(value));%0A        %7D;%0A      %7D%0A    %7D);%0A    return new MatchedPipe%3C%3E(suppliers);%0A  %7D%0A%0A
   @SuppressWarni
@@ -2501,32 +2501,34 @@
 uppliers);%0A  %7D%0A%0A
+%0A%0A
   @SuppressWarni

Add map with supplier