/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * license agreements; and to You under the Apache License, version 2.0:
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * This file is part of the Apache Pekko project, which was derived from Akka.
 */

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

package org.apache.pekko.pattern;

import org.apache.pekko.actor.*;
import org.apache.pekko.dispatch.Futures;
import org.apache.pekko.testkit.PekkoJUnitActorSystemResource;
import org.apache.pekko.testkit.PekkoSpec;
import org.apache.pekko.testkit.TestProbe;
import org.apache.pekko.util.Timeout;
import org.junit.ClassRule;
import org.junit.Test;
import org.scalatestplus.junit.JUnitSuite;
import scala.concurrent.Await;
import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;

import java.util.Arrays;
import java.util.concurrent.*;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

import static org.apache.pekko.pattern.Patterns.ask;
import static org.apache.pekko.pattern.Patterns.pipe;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.junit.Assert.assertEquals;

/** Copyright (C) 2009-2018 Lightbend Inc. <https://www.lightbend.com> */
public class PatternsTest extends JUnitSuite {

  public static final class ExplicitAskTestActor extends AbstractActor {

    public static final class Message implements NoSerializationVerificationNeeded {

      public Message(final String text, ActorRef replyTo) {
        this.text = text;
        this.replyTo = replyTo;
      }

      public final String text;
      public final ActorRef replyTo;
    }

    public Receive createReceive() {
      return receiveBuilder()
          .match(Message.class, message -> message.replyTo.tell(message.text, getSelf()))
          .build();
    }
  }

  public static final class StopActor extends AbstractActor {

    @Override
    public Receive createReceive() {
      return receiveBuilder()
          .match(String.class, message -> sender().tell("Pong", getSelf()))
          .build();
    }
  }

  @ClassRule
  public static PekkoJUnitActorSystemResource actorSystemResource =
      new PekkoJUnitActorSystemResource("JavaAPI", PekkoSpec.testConf());

  private final ActorSystem system = actorSystemResource.getSystem();

  private final ExecutionContext ec = system.dispatcher();

  @Test
  public void useAsk() throws Exception {
    ActorRef testActor = system.actorOf(Props.create(JavaAPITestActor.class), "test");
    scala.concurrent.duration.Duration timeout =
        scala.concurrent.duration.Duration.create(3, "seconds");
    assertEquals(
        "Ask should return expected answer",
        JavaAPITestActor.ANSWER,
        Await.result(ask(testActor, "hey!", 3000), timeout));
  }

  @Test
  public void useAskWithActorSelection() throws Exception {
    scala.concurrent.duration.Duration timeout =
        scala.concurrent.duration.Duration.create(3, "seconds");
    ActorRef testActor = system.actorOf(Props.create(JavaAPITestActor.class), "test2");
    ActorSelection selection = system.actorSelection("/user/test2");
    ActorIdentity id =
        (ActorIdentity) Await.result(ask(selection, new Identify("yo!"), 3000), timeout);
    assertEquals(
        "Ask (Identify) should return the proper ActorIdentity", testActor, id.getActorRef().get());
  }

  @Test
  public void testCSAsk() throws Exception {
    ActorRef target = system.actorOf(Props.create(JavaAPITestActor.class));
    CompletionStage<String> result =
        Patterns.ask(target, "hello", Duration.ofSeconds(3)).thenApply(o -> (String) o);

    String actual = result.toCompletableFuture().get(3, SECONDS);
    assertEquals(JavaAPITestActor.ANSWER, actual);
  }

  @Test
  public void testCSAskWithActorSelection() throws Exception {
    ActorRef target = system.actorOf(Props.create(JavaAPITestActor.class), "test3");

    ActorSelection selection = system.actorSelection("/user/test3");
    ActorIdentity id =
        Patterns.ask(selection, new Identify("hello"), Duration.ofSeconds(3))
            .toCompletableFuture()
            .thenApply(o -> (ActorIdentity) o)
            .get(3, SECONDS);

    assertEquals(target, id.getActorRef().get());
  }

  @Test
  public void testCSAskWithReplyToTimeout() throws Exception {
    final String expected = "hello";

    final ActorRef echo = system.actorOf(Props.create(ExplicitAskTestActor.class));
    final CompletionStage<String> response =
        Patterns.askWithReplyTo(
                echo,
                replyTo -> new ExplicitAskTestActor.Message(expected, replyTo),
                Duration.ofSeconds(3))
            .thenApply(o -> (String) o);

    final String actual = response.toCompletableFuture().get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test
  public void testCSAskWithReplyToTimeoutMillis() throws Exception {
    final String expected = "hello";

    final ActorRef echo = system.actorOf(Props.create(ExplicitAskTestActor.class));
    final CompletionStage<String> response =
        Patterns.askWithReplyTo(
                echo,
                replyTo -> new ExplicitAskTestActor.Message(expected, replyTo),
                Duration.ofSeconds(3))
            .thenApply(o -> (String) o);

    final String actual = response.toCompletableFuture().get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test
  public void testCSAskSelectionWithReplyToTimeoutMillis() throws Exception {
    final String expected = "hello";

    final ActorRef echo = system.actorOf(Props.create(ExplicitAskTestActor.class));
    final ActorSelection selection = system.actorSelection(echo.path());
    final CompletionStage<String> response =
        Patterns.askWithReplyTo(
                selection,
                replyTo -> new ExplicitAskTestActor.Message(expected, replyTo),
                Duration.ofSeconds(3))
            .thenApply(o -> (String) o);

    final String actual = response.toCompletableFuture().get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test
  public void testAskWithReplyToTimeoutMillis() throws Exception {
    final String expected = "hello";

    final ActorRef echo = system.actorOf(Props.create(ExplicitAskTestActor.class));
    final Future<Object> response =
        Patterns.askWithReplyTo(
            echo, replyTo -> new ExplicitAskTestActor.Message(expected, replyTo), 3000);

    final Object actual = Await.result(response, FiniteDuration.apply(3, SECONDS));
    assertEquals(expected, actual);
  }

  @Test
  public void testAskSelectionWithReplyToTimeoutMillis() throws Exception {
    final String expected = "hello";

    final ActorRef echo = system.actorOf(Props.create(ExplicitAskTestActor.class));
    final ActorSelection selection = system.actorSelection(echo.path());
    final Future<Object> response =
        Patterns.askWithReplyTo(
            selection, replyTo -> new ExplicitAskTestActor.Message(expected, replyTo), 3000);

    final Object actual = Await.result(response, FiniteDuration.apply(3, SECONDS));
    assertEquals(expected, actual);
  }

  @Test
  public void testAskWithReplyToTimeout() throws Exception {
    final String expected = "hello";

    final ActorRef echo = system.actorOf(Props.create(ExplicitAskTestActor.class));
    final Future<Object> response =
        Patterns.askWithReplyTo(
            echo,
            replyTo -> new ExplicitAskTestActor.Message(expected, replyTo),
            Timeout.apply(3, SECONDS));

    final Object actual = Await.result(response, FiniteDuration.apply(3, SECONDS));
    assertEquals(expected, actual);
  }

  @Test
  public void usePipe() throws Exception {
    TestProbe probe = new TestProbe(system);
    pipe(Futures.successful("ho!"), system.dispatcher()).to(probe.ref());
    probe.expectMsg("ho!");
  }

  @Test
  public void usePipeWithActorSelection() throws Exception {
    TestProbe probe = new TestProbe(system);
    ActorSelection selection = system.actorSelection(probe.ref().path());
    pipe(Futures.successful("hi!"), system.dispatcher()).to(selection);
    probe.expectMsg("hi!");
  }

  @Test
  public void testCSPipeToActorRef() throws Exception {
    TestProbe probe = new TestProbe(system);
    CompletableFuture<String> f = new CompletableFuture<>();
    f.complete("ho!");
    Patterns.pipe(f, ec).to(probe.ref());
    probe.expectMsg("ho!");
  }

  @Test
  public void testCSPipeToActorSelection() throws Exception {
    TestProbe probe = new TestProbe(system);
    ActorSelection selection = system.actorSelection(probe.ref().path());
    CompletableFuture<String> f = new CompletableFuture<>();
    f.complete("hi!");
    Patterns.pipe(f, ec).to(selection);
    probe.expectMsg("hi!");
  }

  @Test
  public void testRetryCompletionStageNoDelay() throws Exception {
    final String expected = "hello";

    CompletionStage<String> retriedFuture =
        Patterns.retry(() -> CompletableFuture.completedFuture(expected), 3, ec);

    String actual = retriedFuture.toCompletableFuture().get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test
  public void testRetryCompletionStageRandomDelay() throws Exception {
    final String expected = "hello";
    final AtomicInteger counter = new AtomicInteger(0);
    CompletionStage<String> retriedFuture =
        Patterns.retry(
            () -> {
              if (counter.incrementAndGet() <= 3) {
                final CompletableFuture<String> empty = new CompletableFuture<>();
                empty.completeExceptionally(new RuntimeException("failed by purpose."));
                return empty;
              } else {
                return CompletableFuture.completedFuture(expected);
              }
            },
            3,
            Duration.ofMillis(100),
            Duration.ofMillis(200),
            0.2d,
            system.scheduler(),
            ec);

    String actual = retriedFuture.toCompletableFuture().get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test
  public void testRetry() throws Exception {
    final String expected = "hello";

    Future<String> retriedFuture =
        Patterns.retry(
            () -> Futures.successful(expected),
            3,
            scala.concurrent.duration.Duration.apply(200, "millis"),
            system.scheduler(),
            ec);

    String actual = Await.result(retriedFuture, FiniteDuration.apply(3, SECONDS));
    assertEquals(expected, actual);
  }

  @Test
  public void testCSRetry() throws Exception {
    final String expected = "hello";

    Callable<CompletionStage<String>> attempt = () -> CompletableFuture.completedFuture(expected);

    CompletionStage<String> retriedStage =
        Patterns.retry(attempt, 3, Duration.ofMillis(200), system.scheduler(), ec);

    final String actual = retriedStage.toCompletableFuture().get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test(expected = IllegalStateException.class)
  public void testAfterFailedCallable() throws Exception {
    Callable<Future<String>> failedCallable =
        () -> Futures.failed(new IllegalStateException("Illegal!"));

    Future<String> delayedFuture =
        Patterns.after(
            scala.concurrent.duration.Duration.create(200, "millis"),
            system.scheduler(),
            ec,
            failedCallable);

    Future<String> resultFuture = Futures.firstCompletedOf(Arrays.asList(delayedFuture), ec);
    Await.result(resultFuture, scala.concurrent.duration.FiniteDuration.apply(3, SECONDS));
  }

  @Test(expected = IllegalStateException.class)
  public void testAfterFailedFuture() throws Exception {

    Future<String> delayedFuture =
        Patterns.after(
            scala.concurrent.duration.Duration.create(200, "millis"),
            system.scheduler(),
            ec,
            () -> Futures.failed(new IllegalStateException("Illegal!")));

    Future<String> resultFuture = Futures.firstCompletedOf(Arrays.asList(delayedFuture), ec);
    Await.result(resultFuture, FiniteDuration.apply(3, SECONDS));
  }

  @Test
  public void testAfterSuccessfulCallable() throws Exception {
    final String expected = "Hello";

    Future<String> delayedFuture =
        Patterns.after(
            scala.concurrent.duration.Duration.create(200, "millis"),
            system.scheduler(),
            ec,
            () -> Futures.successful(expected));

    Future<String> resultFuture = Futures.firstCompletedOf(Arrays.asList(delayedFuture), ec);
    final String actual = Await.result(resultFuture, FiniteDuration.apply(3, SECONDS));

    assertEquals(expected, actual);
  }

  @Test
  public void testAfterSuccessfulFuture() throws Exception {
    final String expected = "Hello";

    Future<String> delayedFuture =
        Patterns.after(
            scala.concurrent.duration.Duration.create(200, "millis"),
            system.scheduler(),
            ec,
            () -> Futures.successful(expected));

    Future<String> resultFuture = Futures.firstCompletedOf(Arrays.asList(delayedFuture), ec);

    final String actual = Await.result(resultFuture, FiniteDuration.apply(3, SECONDS));
    assertEquals(expected, actual);
  }

  @Test
  public void testAfterFiniteDuration() throws Exception {
    final String expected = "Hello";

    Future<String> delayedFuture =
        Patterns.after(
            scala.concurrent.duration.Duration.create(200, "millis"),
            system.scheduler(),
            ec,
            () -> Futures.successful("world"));

    Future<String> immediateFuture = Futures.future(() -> expected, ec);

    Future<String> resultFuture =
        Futures.firstCompletedOf(Arrays.asList(delayedFuture, immediateFuture), ec);

    final String actual = Await.result(resultFuture, FiniteDuration.apply(3, SECONDS));
    assertEquals(expected, actual);
  }

  @Test(expected = ExecutionException.class)
  public void testCSAfterFailedCallable() throws Exception {
    Callable<CompletionStage<String>> failedCallable =
        () -> {
          CompletableFuture<String> f = new CompletableFuture<>();
          f.completeExceptionally(new IllegalStateException("Illegal!"));
          return f;
        };

    CompletionStage<String> delayedStage =
        Patterns.after(Duration.ofMillis(200), system.scheduler(), ec, failedCallable);

    delayedStage.toCompletableFuture().get(3, SECONDS);
  }

  @Test(expected = ExecutionException.class)
  public void testCSAfterFailedFuture() throws Exception {
    Callable<CompletionStage<String>> failedFuture =
        () -> {
          CompletableFuture<String> f = new CompletableFuture<>();
          f.completeExceptionally(new IllegalStateException("Illegal!"));
          return f;
        };

    CompletionStage<String> delayedStage =
        Patterns.after(Duration.ofMillis(200), system.scheduler(), ec, failedFuture);

    String result = delayedStage.toCompletableFuture().get(3, SECONDS);
  }

  @Test
  public void testCSAfterSuccessfulCallable() throws Exception {
    final String expected = "Hello";

    final Callable<CompletionStage<String>> cf =
        () -> {
          CompletableFuture<String> f = CompletableFuture.completedFuture(expected);
          return f;
        };

    CompletionStage<String> delayedStage =
        Patterns.after(Duration.ofMillis(200), system.scheduler(), ec, cf);

    final String actual = delayedStage.toCompletableFuture().get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test
  public void testCSAfterSuccessfulFuture() throws Exception {
    final String expected = "Hello";

    final CompletionStage<String> f = CompletableFuture.completedFuture(expected);

    CompletionStage<String> delayedStage =
        Patterns.after(
            Duration.ofMillis(200),
            system.scheduler(),
            ec,
            () -> CompletableFuture.completedFuture(expected));

    final String actual = delayedStage.toCompletableFuture().get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test
  public void testCSAfterDuration() throws Exception {
    final String expected = "Hello";

    final CompletionStage<String> f = CompletableFuture.completedFuture("world!");

    CompletionStage<String> delayedStage =
        Patterns.after(
            Duration.ofMillis(200),
            system.scheduler(),
            ec,
            () -> CompletableFuture.completedFuture("world!"));

    CompletableFuture<String> immediateStage = CompletableFuture.completedFuture(expected);
    CompletableFuture<Object> resultStage =
        CompletableFuture.anyOf(delayedStage.toCompletableFuture(), immediateStage);

    final String actual = (String) resultStage.get(3, SECONDS);
    assertEquals(expected, actual);
  }

  @Test
  public void testGracefulStop() throws Exception {
    ActorRef target = system.actorOf(Props.create(StopActor.class));
    Future<Boolean> result =
        Patterns.gracefulStop(target, FiniteDuration.apply(200, TimeUnit.MILLISECONDS));

    Boolean actual = Await.result(result, FiniteDuration.apply(3, SECONDS));
    assertEquals(true, actual);
  }

  @Test
  public void testCSGracefulStop() throws Exception {
    ActorRef target = system.actorOf(Props.create(StopActor.class));
    CompletionStage<Boolean> result = Patterns.gracefulStop(target, Duration.ofMillis(200));

    Boolean actual = result.toCompletableFuture().get(3, SECONDS);
    assertEquals(true, actual);
  }
}
