package net.jgp.books.spark.ch10.x.utils.streaming.app;

import java.io.File;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import net.jgp.books.spark.ch10.x.utils.streaming.lib.FieldType;
import net.jgp.books.spark.ch10.x.utils.streaming.lib.RecordGeneratorUtils;
import net.jgp.books.spark.ch10.x.utils.streaming.lib.RecordStructure;
import net.jgp.books.spark.ch10.x.utils.streaming.lib.RecordWriterUtils;
import net.jgp.books.spark.ch10.x.utils.streaming.lib.StreamingUtils;

/**
 * This application generates a series of random records at random interval
 * during one minute. All parameters are random or configurable.
 *
 * @author jgp
 */
public class RecordsInFilesGeneratorApp {
    private static Logger log = LoggerFactory
            .getLogger(RecordsInFilesGeneratorApp.class);

    /**
     * Streaming duration in seconds.
     */
    public int streamDuration = 60;

    /**
     * Maximum number of records send at the same time.
     */
    public int batchSize = 10;

    /**
     * Wait time between two batches of records, in seconds, with an element
     * of variability. If you say 10 seconds, the system will wait between 5s
     * and 15s, if you say 20s, the system will wait between 10s and 30s, and
     * so on.
     */
    public int waitTime = 5;

    public static void main(String[] args) {
        // corresponds with the following execution entry from the pom.xml file
        // /project/build/plugins/plugin[artifactId='exec-maven-plugin']/executions/execution[id='generate-records-in-files']
        // whereby the following if statement will evaluate false
        String outputDirectory = StreamingUtils.getInputDirectory();

        // Improvement: Both this class and
        // net.jgp.books.spark.ch10.lab400_read_records_from_multiple_streams.ReadRecordFromMultipleFileStreamApp
        // use the same subDirs generated by StreamingUtils when there are input arguments
        if (args.length == 2
                && args[0].compareToIgnoreCase("--output-directory") == 0) {

            if ("subdir1".equals(args[1])) {
                // corresponds with the following execution entry from the pom.xml file
                // /project/build/plugins/plugin[artifactId='exec-maven-plugin']/executions/execution[id='generate-records-in-files-subdir1']
                outputDirectory = StreamingUtils.getInputSubDirectory1();
            } else if ("subdir2".equals(args[1])) {
                // corresponds with the following execution entry from the pom.xml file
                // /project/build/plugins/plugin[artifactId='exec-maven-plugin']/executions/execution[id='generate-records-in-files-subdir2']
                outputDirectory = StreamingUtils.getInputSubDirectory2();
            } else {
                log.warn("The second argument {} is not recognized as a valid subdirectory, " +
                                "therefore {} will be used to store the records.",
                        args[1], outputDirectory);
            }

            log.info("All records created by this instance will be stored in {}", outputDirectory);
        }
        RecordStructure rs = new RecordStructure("contact")
                .add("fname", FieldType.FIRST_NAME)
                .add("mname", FieldType.FIRST_NAME)
                .add("lname", FieldType.LAST_NAME)
                .add("age", FieldType.AGE)
                .add("ssn", FieldType.SSN);

        RecordsInFilesGeneratorApp app = new RecordsInFilesGeneratorApp();
        app.start(rs, outputDirectory);
    }

    private void start(RecordStructure rs, String outputDirectory) {
        log.debug("-> start (..., {})", outputDirectory);
        long start = System.currentTimeMillis();
        while (start + streamDuration * 1000 > System.currentTimeMillis()) {
            int maxRecord = RecordGeneratorUtils.getRandomInt(batchSize) + 1;
            RecordWriterUtils.write(
                    rs.getRecordName() + "_" + System.currentTimeMillis() + ".txt",
                    rs.getRecords(maxRecord, false),
                    outputDirectory);
            try {
                Thread.sleep(RecordGeneratorUtils.getRandomInt(waitTime * 1000)
                        + waitTime * 1000 / 2);
            } catch (InterruptedException e) {
                // Simply ignore the interruption
            }
        }
    }
}
