/*
 * Copyright (C) 2017-2021 Lightbend Inc. <https://www.lightbend.com>
 */

package akka.stream.javadsl;

import akka.NotUsed;
import akka.japi.Pair;
import akka.japi.Creator;
import akka.stream.impl.JavaFlowAndRsConverters;

/**
 * For use only with `JDK 9+`.
 * <p>
 * Provides support for `java.util.concurrent.Flow.*` interfaces which mirror the Reactive Streams
 * interfaces from `org.reactivestreams`. See <a href="http//www.reactive-streams.org/">reactive-streams.org</a>.
 */
public final class JavaFlowSupport {

  private static final NotUsed NotUsed = akka.NotUsed.getInstance();

  private JavaFlowSupport() {
    throw new RuntimeException("No instances allowed!");
  }

  /**
   * {@link akka.stream.javadsl.Flow]] factories operating with {@code java.util.concurrent.Flow.*} interfaces.
   */
  public static final class Source {
    private Source() {
      throw new RuntimeException("No instances allowed!");
    }

    /**
     * Helper to create {@code Source} from {@link java.util.concurrent.Flow.Publisher}.
     * <p>
     * Construct a transformation starting with given publisher. The transformation steps
     * are executed by a series of {@link java.util.concurrent.Flow.Processor} instances
     * that mediate the flow of elements downstream and the propagation of
     * back-pressure upstream.
     * <p>
     * See also {@code Source.fromPublisher} if wanting to integrate with {@link org.reactivestreams.Publisher} instead
     * (which carries the same semantics, however existed before RS's inclusion in Java 9).
     */
    public static <T> akka.stream.javadsl.Source<T, NotUsed> fromPublisher(java.util.concurrent.Flow.Publisher<T> publisher) {
      return akka.stream.javadsl.Source.<T>fromPublisher(JavaFlowAndRsConverters.asRs(publisher));
    }

    /**
     * Creates a {@code Source} that is materialized as a {@link java.util.concurrent.Flow.Subscriber}.
     * <p>
     * See also {@code Source.asSubscriber} if wanting to integrate with {@link org.reactivestreams.Subscriber} instead
     * (which carries the same semantics, however existed before RS's inclusion in Java 9).
     */
    //#asSubscriber
    public static <T> akka.stream.javadsl.Source<T, java.util.concurrent.Flow.Subscriber<T>> asSubscriber() {
    //#asSubscriber
      return akka.stream.javadsl.Source.<T>asSubscriber().mapMaterializedValue(JavaFlowAndRsConverters::asJava);
    }
  }

  /**
   * {@link akka.stream.javadsl.Flow]] factories operating with {@code java.util.concurrent.Flow.*} interfaces.
   */
  public static final class Flow {
    private Flow() {
      throw new RuntimeException("No instances allowed!");
    }

    /**
     * Creates a Flow from a {@link java.util.concurrent.Flow.Processor}
     */
    public static <I, O> akka.stream.javadsl.Flow<I, O, NotUsed> fromProcessor(Creator<java.util.concurrent.Flow.Processor<I, O>> processorFactory) throws Exception {
      return fromProcessorMat(() -> Pair.apply(processorFactory.create(), NotUsed));
    }


    /**
     * Creates a Flow from a {@link java.util.concurrent.Flow.Processor>> and returns a materialized value.
     */
    public static <I, O, M> akka.stream.javadsl.Flow<I, O, M> fromProcessorMat(
      akka.japi.Creator<akka.japi.Pair<java.util.concurrent.Flow.Processor<I, O>, M>> processorFactory) throws Exception {
      final Pair<java.util.concurrent.Flow.Processor<I, O>, M> value = processorFactory.create();
      final java.util.concurrent.Flow.Processor<I, O> processor = value.first();
      final M mat = value.second();

      return akka.stream.javadsl.Flow.fromProcessorMat(() ->
        akka.japi.Pair.apply(JavaFlowAndRsConverters.asRs(processor), mat)
      );

    }

    /**
     * Converts this Flow to a {@code RunnableGraph} that materializes to a Reactive Streams {@link java.util.concurrent.Flow.Processor}
     * which implements the operations encapsulated by this Flow. Every materialization results in a new Processor
     * instance, i.e. the returned {@code RunnableGraph} is reusable.
     *
     * @return A {@code RunnableGraph} that materializes to a {@code Processor} when {@code run()} is called on it.
     */
    public static <In, Out, Mat> akka.stream.javadsl.RunnableGraph<java.util.concurrent.Flow.Processor<In, Out>> toProcessor(akka.stream.javadsl.Flow<In, Out, Mat> flow) {
      final akka.stream.javadsl.Source<In, java.util.concurrent.Flow.Subscriber<In>> source = JavaFlowSupport.Source.<In>asSubscriber();
      final akka.stream.javadsl.Sink<Out, java.util.concurrent.Flow.Publisher<Out>> sink = JavaFlowSupport.Sink.<Out>asPublisher(AsPublisher.WITHOUT_FANOUT);

      // have to jump though scaladsl for the toMat because type inference of the Keep.both
      return
        source.via(flow).toMat(sink, Keep.both())
          .mapMaterializedValue(pair -> {
            final java.util.concurrent.Flow.Subscriber<In> sub = pair.first();
            final java.util.concurrent.Flow.Publisher<Out> pub = pair.second();

            return new java.util.concurrent.Flow.Processor<In, Out>() {
              @Override public void onError(Throwable t) { sub.onError(t); }
              @Override public void onSubscribe(java.util.concurrent.Flow.Subscription s) { sub.onSubscribe(s); }
              @Override public void onComplete() { sub.onComplete(); }
              @Override public void onNext(In t) { sub.onNext(t); }
              @Override public void subscribe(java.util.concurrent.Flow.Subscriber<? super Out> s) { pub.subscribe(s); }
            };
          });
    }
  }

  /**
   * {@link akka.stream.javadsl.Sink} factories operating with {@code java.util.concurrent.Flow.*} interfaces.
   */
  public static final class Sink {
    private Sink() {
      throw new RuntimeException("No instances allowed!");
    }

    /**
     * A `Sink` that materializes into a {@link java.util.concurrent.Flow.Publisher}.
     * <p>
     * If {@code fanout} is {@code WITH_FANOUT}, the materialized {@code Publisher} will support multiple {@code Subscriber}s and
     * the size of the {@code inputBuffer} configured for this operator becomes the maximum number of elements that
     * the fastest {@link java.util.concurrent.Flow.Subscriber} can be ahead of the slowest one before slowing
     * the processing down due to back pressure.
     * <p>
     * If {@code fanout} is {@code WITHOUT_FANOUT} then the materialized {@code Publisher} will only support a single {@code Subscriber} and
       * reject any additional {@code Subscriber}s.
     */
    public static <T> akka.stream.javadsl.Sink<T, java.util.concurrent.Flow.Publisher<T>> asPublisher(AsPublisher fanout) {
      return akka.stream.javadsl.Sink.<T>asPublisher(fanout).mapMaterializedValue(JavaFlowAndRsConverters::asJava);
    }

      /**
       * Helper to create <<Sink>> from <<java.util.concurrent.Flow.Subscriber>>.
       */
      public static <T> akka.stream.javadsl.Sink<T, NotUsed> fromSubscriber(java.util.concurrent.Flow.Subscriber<T> s) {
        return akka.stream.javadsl.Sink.fromSubscriber(JavaFlowAndRsConverters.asRs(s));
      }

  }

}
