import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import utils.BigFractionUtils;
import utils.PrimeResult;

import java.math.BigInteger;
import java.time.Duration;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

/**
 * This class shows how to apply Project Reactor features
 * asynchronously to perform various Flux operators, including
 * create(), interval(), map(), filter(), doOnNext(), doFinally(),
 * take(), subscribe(), then(), range(), subscribeOn(), publishOn(),
 * and various thread pools.  The Mono fromRunnable() operator is
 * also shown.
 */
@SuppressWarnings("ALL")
public class FluxEx {
    /**
     * Maximum random number value.
     */
    private static final int sMAX_VALUE = 10_000_000;

    /**
     * Max number of iterations.
     */
    private static final int sMAX_ITERATIONS = 10;

    /**
     * Starting point of the randomly-generated numbers.
     */
    private static final int sLOWER_BOUND = sMAX_VALUE - sMAX_ITERATIONS;

    /**
     * A memoizer cache that maps candidate primes to their smallest
     * factor (if they aren't prime) or 0 if they are prime.
     */ 
    private static final Map<BigInteger, BigInteger> sPrimeCache
        = new ConcurrentHashMap<>();

    /**
     * Random number generator.
     */
    private static final Random sRANDOM = new Random();

    /**
     * 0.5 second duration for sleeping for the Flux.interval() operator.
     */
    private static final Duration sSLEEP_DURATION =
        Duration.ofMillis(500);

    /**
     * Define a predicate that only matches odd numbers.
     */
    private static Predicate<BigInteger> sOnlyOdd = bigInteger ->
        !bigInteger.mod(BigInteger.TWO).equals(BigInteger.ZERO);

    /**
     * Generate a random BigInteger (with the goal of generating duplicates).
     */
    private static Function<Long, BigInteger> sGenerateRandomBigInteger = __ ->
        BigInteger.valueOf(sLOWER_BOUND +
                           sRANDOM.nextInt(sMAX_VALUE));

    /**
     * Test a stream of random BigIntegers to determine which values
     * are prime using an asynchronous time-driven Flux stream.
     */
    public static Mono<Void> testIsPrimeTimed() {
        // We use a StringBuffer because it is thread-safe!
        StringBuffer sb =
            new StringBuffer(">> Calling testIsPrimeTimed()\n");

        // Callback that writes the BigInteger to the StringBuffer.
        Consumer<BigInteger> logBigInteger =
            s -> FluxEx.print(s, sb);

        return Flux
            // Factory method creates a flow of random big integers
            // that are generated at a periodic interval in a
            // background thread.
            .create(makeTimedFluxSink())

            // Print the big integer as a debugging aid.
            .doOnNext(logBigInteger)

            // Use a memoizer to check if each random big integer is
            // prime or not on the background thread.
            .map(bigInteger ->
                 FluxEx.checkIfPrime(bigInteger, sb))

            // Process each big integer on the background thread.
            .doOnNext(bigInteger ->
                       FluxEx.processResult(bigInteger,
                                            sb))

            // Display results after all elements in flux stream are
            // processed and return an empty mono to synchronize with
            // AsyncTaskBarrier.
            .then(Mono.fromRunnable(() ->
                                    BigFractionUtils.display(sb.toString())));
    }

    /**
     * A factory method that bridges the reactive world with the
     * callback-style world to emit a time-based flow of random big
     * integers at a periodic interval.
     */
    private static Consumer<FluxSink<BigInteger>> makeTimedFluxSink() {
        // FluxSink emits any number of next() signals followed by
        // zero or one onError()/onComplete().
        return sink -> Flux
            // Generate a Long stream starting at 0 periodically in a
            // background thread from the Schedulers.parallel() thread
            // pool.
            .interval(sSLEEP_DURATION)

            // Generate random numbers between min and max values to
            // ensure some duplicates.
            .map(sGenerateRandomBigInteger)

            // Eliminate even numbers from consideration since they
            // aren't prime!
            .filter(sOnlyOdd)

            // Only take sMAX_ITERATIONS of big integers.
            .take(sMAX_ITERATIONS)

            // Start the processing and emit each random number until
            // complete or an error occurs.
            .subscribe(sink::next,
                       // Shutdown the input stream on error.
                       error -> sink.complete(),
                       // Shutdown the input stream.
                       sink::complete);
    }

    /**
     * Test a stream of random BigIntegers to determine which values
     * are prime using an asynchronous Flux stream.
     */
    public static Mono<Void> testIsPrimeAsync() {
        StringBuffer sb =
            new StringBuffer(">> Calling testIsPrimeAsync()\n");

        // Callback that writes the BigInteger to the StringBuffer.
        Consumer<BigInteger> logBigInteger =
            s -> FluxEx.print(s, sb);

        // Create a Scheduler containing one "subscriber" thread.
        Scheduler subscriber = Schedulers
            .newParallel("subscriber", 1);

        return Flux
            // Factory method creates a stream of random big integers
            // that are generated in a background "publisher" thread.
            .create(makeAsyncFluxSink())

            // Print the big integer as a debugging aid.
            .doOnNext(logBigInteger)

            // Arrange to perform the prime-checking computations in
            // the "subscriber" thread.
            .publishOn(subscriber)

            // Use a memoizer to check if each random big integer is
            // prime or not in the "subscriber" thread.
            .map(bigInteger ->
                 FluxEx.checkIfPrime(bigInteger, sb))

            // Process each big integer in the "subscriber" thread.
            .doOnNext(bigInteger ->
                      FluxEx.processResult(bigInteger,
                                           sb))

            // Display results after all elements in the flux stream are
            // processed successfully.
            .doOnComplete(() ->
                          BigFractionUtils.display(sb.toString()))

            // Always dispose of the "subscriber" thread regardless of
            // what happens in the stream.
           .doFinally(___ -> subscriber.dispose())

            // Return an empty Mono to synchronize with
            // AsyncTaskBarrier.
            .then();
    }

    /**
     * A factory method that bridges the reactive world with the
     * callback-style world to emit a flow of random big integers in a
     * background thread.
     */
    private static Consumer<FluxSink<BigInteger>> makeAsyncFluxSink() {
        // Create a Scheduler containing one "publisher" thread.
        Scheduler publisher = Schedulers
            .newParallel("publisher", 1);

        // FluxSink emits any number of next() signals followed by
        // zero or one onError()/onComplete().
        return sink -> Flux
            // Emit up to Integer.MAX_VALUE integers starting at 1.
            .range(1, Integer.MAX_VALUE)

            // Convert to Long since Flux lacks a rangeLong() method.
            .map(Integer::toUnsignedLong)

            // Arrange to emit the random big integers in the
            // "publisher" thread.
            .subscribeOn(publisher)

            // Generate random numbers between min and max values
            // to ensure some duplicates.
            .map(sGenerateRandomBigInteger)

            // Eliminate even numbers from consideration since they
            // aren't prime!
            .filter(sOnlyOdd)

            // Only take sMAX_ITERATIONS Integer objects.
            .take(sMAX_ITERATIONS)

            // Always dispose of the "publisher" thread regardless of
            // what happens.
            .doFinally(___ -> publisher.dispose())

            // Start the processing and emit each random number until
            // complete or an error occurs.
            .subscribe(sink::next,
                       // Shutdown the input stream on error.
                       error -> sink.complete(),
                       // Shutdown the input stream.
                       sink::complete);
    }

    /**
     * This method checks whether the {@code primeCandidate} is prime or not.
     *
     * @param primeCandidate The number to check for the prime factor.
     * @return a utils.PrimeResult that contains the prime candidate and
     *         either 0 (if the prime candidate is prime) or the
     *         smallest factor (if it's not prime)
     */
    static PrimeResult checkIfPrime(BigInteger primeCandidate,
                                    StringBuffer sb) {
        return new PrimeResult
            (primeCandidate,
             // This atomic "check then act" method serves as
             // a "memoizer" cache.
             sPrimeCache.computeIfAbsent(primeCandidate,
                                         pc -> (FluxEx.isPrime(pc, sb))));
    }

    /**
     * This method provides a brute-force determination of whether
     * number {@code primeCandidate} is prime.  Returns 0 if it is
     * prime or the smallest factor if it is not prime.
     */
    static BigInteger isPrime(BigInteger n,
                              StringBuffer sb) {
        print("explicitly checking if " + n + " is prime",
                sb);

        // Even numbers can't be prime.
        if (n.mod(BigInteger.TWO).compareTo(BigInteger.TWO) == 0)
            return BigInteger.TWO;

        for (BigInteger i = BigInteger.valueOf(3);
             n.compareTo(i.multiply(i)) >= 0;
             i = i.add(BigInteger.TWO))
            if (n.mod(i).compareTo(BigInteger.ZERO) == 0)
                return i;
    
        return BigInteger.ZERO;
    }

    /**
     * Process the {@code primeTuple} to print whether a number if prime.
     */
    private static void processResult(PrimeResult primeTuple,
                                      StringBuffer sb) {
        if (!primeTuple.smallestFactor().equals(BigInteger.ZERO)) {
            print("found a non-prime number with smallest factor "
                  + primeTuple.smallestFactor()
                  + " for "
                  + primeTuple.primeCandidate(), sb);
        } else {
            print("found a prime number "
                  + primeTuple.primeCandidate(), sb);
        }
    }

    /**
     * Print string {@code s} with the thread name appended.
     */
    private static void print(String s, StringBuffer sb) {
        sb.append("["
                  + Thread.currentThread().getName()
                  + ", "
                  + System.currentTimeMillis()
                  + "] "
                  + s
                  + "\n");
    }

    /**
     * Print big integer {@code bigInteger} with the thread name appended.
     */
    private static void print(BigInteger bigInteger, StringBuffer sb) {
        print("emitting " + bigInteger, sb);
    }
}
