package com.conveyal.r5.analyst.broker;

import com.conveyal.r5.analyst.cluster.AnalystWorker;
import com.conveyal.r5.analyst.cluster.JobSimulator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * This test is not an automatic unit test. It is an integration test that must be started manually, because it takes
 * a long time to run. It will start up a broker and some local workers, then submit a large job to the broker. The
 * workers will fail to complete tasks some percentage of the time, but eventually the whole job should be finished
 * because the broker will redeliver lost tasks to the workers.
 *
 * Progress can be followed with:
 * watch --interval 1 curl http://localhost:9001/jobs
 */
public class RedeliveryTest {

    private static final Logger LOG = LoggerFactory.getLogger(RedeliveryTest.class);
    static final int N_TASKS = 100;
    static final int N_WORKERS = 4;
    static final int FAILURE_RATE = 20; // percent

    public static void main(String[] params) {

        // Start a broker in a new thread.
        Properties brokerConfig = new Properties();
        brokerConfig.setProperty("graphs-bucket", "FAKE");
        brokerConfig.setProperty("pointsets-bucket", "FAKE");
        brokerConfig.setProperty("work-offline", "true");
        BrokerMain brokerMain = new BrokerMain(brokerConfig);
        Thread brokerThread = new Thread(brokerMain); // TODO combine broker and brokermain, set offline mode.
        brokerThread.start();

        // Start some workers.
        // Do not set any initial graph, because the workers are only going to simulate doing any work.
        Properties workerConfig = new Properties();
        List<Thread> workerThreads = new ArrayList<>();
        for (int i = 0; i < N_WORKERS; i++) {
            AnalystWorker worker = new AnalystWorker(workerConfig);
            worker.dryRunFailureRate = FAILURE_RATE;
            Thread workerThread = new Thread(worker);
            workerThreads.add(workerThread);
            workerThread.start();
        }

        // Feed some work to the broker.
        JobSimulator jobSimulator = new JobSimulator();
        jobSimulator.nOrigins = N_TASKS;
        jobSimulator.graphId = "GRAPH";
        jobSimulator.sendFakeJob();

        // Wait for all tasks to be marked finished
        while (brokerMain.broker.anyJobsActive()) {
            try {
                LOG.info("Some jobs are still not complete.");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        LOG.info("All jobs finished.");
        System.exit(0);
    }

}

@@ -2187,16 +2187,17 @@
 imulator
+A
  = new J
@@ -2231,16 +2231,17 @@
 imulator
+A
 .nOrigin
@@ -2273,16 +2273,17 @@
 imulator
+A
 .graphId
@@ -2314,16 +2314,380 @@
 imulator
+A.sendFakeJob();%0A%0A        try %7B%0A            Thread.sleep(5000);%0A        %7D catch (InterruptedException e) %7B %7D%0A%0A        // Feed another job to the broker, to ensure redelivery can deal with multiple jobs.%0A        JobSimulator jobSimulatorB = new JobSimulator();%0A        jobSimulatorB.nOrigins = N_TASKS;%0A        jobSimulatorB.graphId = %22GRAPH%22;%0A        jobSimulatorB
 .sendFak
@@ -2912,24 +2912,24 @@
 leep(2000);%0A
+
             
@@ -2966,67 +2966,8 @@
 e) %7B
-%0A                throw new RuntimeException(e);%0A           
  %7D%0A 

Update redelivery test to enqueue multiple jobs.