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

package jdocs.stream;

import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.actor.Cancellable;
import akka.stream.*;
import akka.stream.javadsl.*;
import akka.testkit.javadsl.TestKit;
import jdocs.AbstractJavaTest;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;

import java.time.Duration;
import java.util.Arrays;

public class StreamBuffersRateDocTest extends AbstractJavaTest {

  static class Job {}

  static ActorSystem system;
  static Materializer mat;

  @BeforeClass
  public static void setup() {
    system = ActorSystem.create("StreamBuffersDocTest");
    mat = ActorMaterializer.create(system);
  }

  @AfterClass
  public static void tearDown() {
    TestKit.shutdownActorSystem(system);
    system = null;
    mat = null;
  }

  final SilenceSystemOut.System System = SilenceSystemOut.get();

  @Test
  public void demonstratePipelining() {
    //#pipelining
    Source.from(Arrays.asList(1, 2, 3))
      .map(i -> {System.out.println("A: " + i); return i;}).async()
      .map(i -> {System.out.println("B: " + i); return i;}).async()
      .map(i -> {System.out.println("C: " + i); return i;}).async()
      .runWith(Sink.ignore(), mat);
    //#pipelining
  }

  @Test
  @SuppressWarnings("unused")
  public void demonstrateBufferSizes() {
    //#materializer-buffer
    final Materializer materializer = ActorMaterializer.create(
      ActorMaterializerSettings.create(system)
        .withInputBuffer(64, 64), system);
    //#materializer-buffer

    //#section-buffer
    final Flow<Integer, Integer, NotUsed> flow1 =
      Flow.of(Integer.class)
      .map(elem -> elem * 2).async()
      .addAttributes(Attributes.inputBuffer(1, 1)); // the buffer size of this map is 1
    final Flow<Integer, Integer, NotUsed> flow2 =
      flow1.via(
        Flow.of(Integer.class)
        .map(elem -> elem / 2)).async(); // the buffer size of this map is the default
    //#section-buffer
  }

  @Test
  public void demonstrateBufferAbstractionLeak() {
    //#buffering-abstraction-leak
    final Duration oneSecond = Duration.ofSeconds(1);
    final Source<String, Cancellable> msgSource =
        Source.tick(oneSecond, oneSecond, "message!");
    final Source<String, Cancellable> tickSource =
        Source.tick(oneSecond.multipliedBy(3), oneSecond.multipliedBy(3), "tick");
    final Flow<String, Integer, NotUsed> conflate =
        Flow.of(String.class).conflateWithSeed(
            first -> 1, (count, elem) -> count + 1);

    RunnableGraph.fromGraph(GraphDSL.create(b -> {
      // this is the asynchronous stage in this graph
      final FanInShape2<String, Integer, Integer> zipper =
          b.add(ZipWith.create((String tick, Integer count) -> count).async());
      b.from(b.add(msgSource)).via(b.add(conflate)).toInlet(zipper.in1());
      b.from(b.add(tickSource)).toInlet(zipper.in0());
      b.from(zipper.out()).to(b.add(Sink.foreach(elem -> System.out.println(elem))));
      return ClosedShape.getInstance();
    })).run(mat);
    //#buffering-abstraction-leak
  }

  @Test
  public void demonstrateExplicitBuffers() {
    final Source<Job, NotUsed> inboundJobsConnector = Source.empty();
    //#explicit-buffers-backpressure
    // Getting a stream of jobs from an imaginary external system as a Source
    final Source<Job, NotUsed> jobs = inboundJobsConnector;
    jobs.buffer(1000, OverflowStrategy.backpressure());
    //#explicit-buffers-backpressure

    //#explicit-buffers-droptail
    jobs.buffer(1000, OverflowStrategy.dropTail());
    //#explicit-buffers-droptail

    //#explicit-buffers-dropnew
    jobs.buffer(1000, OverflowStrategy.dropNew());
    //#explicit-buffers-dropnew

    //#explicit-buffers-drophead
    jobs.buffer(1000, OverflowStrategy.dropHead());
    //#explicit-buffers-drophead

    //#explicit-buffers-dropbuffer
    jobs.buffer(1000, OverflowStrategy.dropBuffer());
    //#explicit-buffers-dropbuffer

    //#explicit-buffers-fail
    jobs.buffer(1000, OverflowStrategy.fail());
    //#explicit-buffers-fail
  }

}
