package eu.dnetlib.dli;

import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
import eu.dnetlib.enabling.tools.DnetStreamSupport;
import eu.dnetlib.pid.resolver.AbstractPIDResolver;

import java.util.List;
import java.util.stream.Collectors;

class SparkJob {
    private Integer jobId;
    private String name;
    private String submissionTime;
    private String status;
    private int numTasks;
    private int numActiveTasks;
    private int numCompletedTasks;

    static SparkJob fromJsonElement(final JsonElement element) {
        if (element.isJsonObject()) {
            final SparkJob result = new SparkJob();
            result.setJobId(element.getAsJsonObject().get("jobId").getAsInt());
            result.setName(element.getAsJsonObject().get("name").getAsString());
            result.setStatus(element.getAsJsonObject().get("status").getAsString());
            result.setSubmissionTime(element.getAsJsonObject().get("submissionTime").getAsString());
            result.setNumTasks(element.getAsJsonObject().get("numTasks").getAsInt());
            result.setNumActiveTasks(element.getAsJsonObject().get("numActiveTasks").getAsInt());
            result.setNumCompletedTasks(element.getAsJsonObject().get("numCompletedTasks").getAsInt());
            return result;
        } else
            return null;
    }

    public int getNumActiveTasks() {
        return numActiveTasks;
    }

    public void setNumActiveTasks(int numActiveTasks) {
        this.numActiveTasks = numActiveTasks;
    }

    public String getSubmissionTime() {
        return submissionTime;
    }

    public void setSubmissionTime(String submissionTime) {
        this.submissionTime = submissionTime;
    }

    public Integer getJobId() {
        return jobId;
    }

    public void setJobId(Integer jobId) {
        this.jobId = jobId;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }

    public int getNumTasks() {
        return numTasks;
    }

    public void setNumTasks(int numTasks) {
        this.numTasks = numTasks;
    }

    public int getNumCompletedTasks() {
        return numCompletedTasks;
    }

    public void setNumCompletedTasks(int numCompletedTasks) {
        this.numCompletedTasks = numCompletedTasks;
    }
}

class SparkApplication {
    private String name;
    private String id;

    static SparkApplication fromJsonElement(final JsonElement element) {
        if (element.isJsonObject()) {
            final String id =  element.getAsJsonObject().get("id").getAsString();
            final String name =  element.getAsJsonObject().get("name").getAsString();
            final SparkApplication result = new SparkApplication();
            result.setId(id);
            result.setName(name);
            return result;
        } else
            return null;

    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }


}


public class SparkInspector {


    public List<SparkApplication> listApplication() {
        final String response = AbstractPIDResolver.requestURL("http://localhost:4040/api/v1/applications");

        final JsonParser parser = new JsonParser();
        final JsonElement root = parser.parse(response);
        if (root.isJsonArray()) {
            return DnetStreamSupport.generateStreamFromIterator(root.getAsJsonArray().iterator()).map(SparkApplication::fromJsonElement).collect(Collectors.toList());
        }
        return null;
    }


    public List<SparkJob> listJob(final SparkApplication application) {
        final String response = AbstractPIDResolver.requestURL("http://localhost:4040/api/v1/applications/"+application.getId()+"/jobs");
        final JsonParser parser = new JsonParser();
        final JsonElement root = parser.parse(response);
        if (root.isJsonArray()) {
            return DnetStreamSupport.generateStreamFromIterator(root.getAsJsonArray().iterator()).map(SparkJob::fromJsonElement).collect(Collectors.toList());
        }
        return null;
    }


    public static void main(String[] args) {
        SparkInspector si = new SparkInspector();
        Gson g = new Gson();
        si.listApplication().forEach(it ->
                System.out.println(g.toJson(si.listJob(it)))

        );
    }
}