package com.tesco.payments.async;

import akka.actor.*;
import akka.japi.pf.DeciderBuilder;
import akka.pattern.Backoff;
import akka.pattern.BackoffSupervisor;
import akka.testkit.javadsl.TestKit;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import scala.Option;
import scala.concurrent.duration.Duration;

import java.util.concurrent.TimeUnit;

import static akka.actor.SupervisorStrategy.escalate;
import static akka.actor.SupervisorStrategy.restart;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;

public class TestSupervisionDocs {
    static int counter = 0;

    private static ActorSystem system;

    @Test public void supervisedActorIsRestarted() {
        supervised().tell("foo", ActorRef.noSender());

        await().until(() -> counter == 2);
    }

    private ActorRef supervised() {
        return system.actorOf(supervised(SupervisedActor.props()), "child-supervisor");
    }

    private Props supervised(final Props supervised) {
        return BackoffSupervisor.props(Backoff.onStop(
                supervised, "child-actor",
                Duration.create(100, MILLISECONDS),
                Duration.create(1, SECONDS),
                0.2).withSupervisorStrategy(
                new OneForOneStrategy(10, Duration.create(1, TimeUnit.MINUTES),
                        DeciderBuilder
                                .match(RestartMe.class, e -> {
                                    System.out.println("RETRY caught------------------");
                                    return restart();
                                })
                                .matchAny(e -> escalate()).build())));
    }

    @BeforeAll
    static void beforeClass() {
        system = ActorSystem.apply("TestActorSystem");
    }

    @AfterAll
    static void afterAll() {
        TestKit.shutdownActorSystem(system);
    }

    static class SupervisedActor extends AbstractActor {

        private ActorRef originator;

        @Override public Receive createReceive() {
            return receiveBuilder()
                    .match(String.class, m -> {
                        originator = getSender();
                        System.out.println("received. Count=" + counter);
                        if (counter++ < 3) {
                            throw new RestartMe() ;
                        }
                    })
                    .matchAny(m -> System.out.println("matched any: " + m))
                    .build();
        }

        @Override public void preRestart(Throwable reason, Option<Object> message) {
            System.out.println("SupervisedActor restarting with message " + message.get() + " of type " + message.get().getClass().getSimpleName());
            getContext().getParent().tell(message.get(), originator);
        }

        @Override public void postRestart(Throwable reason) {
            System.out.println("SupervisedActor restarted! " + getSelf().path());
        }

        public static Props props() {
            return Props.create(SupervisedActor.class);
        }
    }

    private static class RestartMe extends RuntimeException {
    }
}