@XmlRootElement
public class AgentConfig {

    private Actor actor;
    private Job job;
    private LogServer logServer;
    private String contactPoint;

    public Stream<String> getContactPoint() {
        String[] contacts = contactPoint.split(",") ;
        return Stream.of(contacts)
                .map(p -> String.format("akka.tcp://%s@%s/system/receptionist",Constants.PerformanceSystem, p));
    }

    public void setContactPoint(String contactPoint) {
        this.contactPoint = contactPoint;
    }

    public LogServer getLogServer() {
        return logServer;
    }

    public void setLogServer(LogServer logServer) {
        this.logServer = logServer;
    }

    public Actor getActor() {
        return actor;
    }

    public void setActor(Actor actor) {
        this.actor = actor;
    }

    public Job getJob() {
        return job;
    }

    public void setJob(Job job) {
        this.job = job;
    }

    public String getUrl(String filePath) {
        return getGenericUrl("api/log/stream","filePath",filePath);
    }

    public String getMasterUrl(String filePath) {
        return getGenericUrl("gatling/lib/stream","filePath",filePath);
    }

    public String getAbortUrl() {
        return getGenericUrl("gatling/server/abort","trackingId",StringUtils.EMPTY);
    }

    public String getGenericUrl(String path,String queryStringKey,String queryStringValue) {
        String host = logServer.getHostName();
        if (StringUtils.isEmpty(logServer.getHostName())) {
            try {
                host = InetAddress.getLocalHost().getHostName();
            } catch (UnknownHostException e) {
                ;
            }
        }
        String result = StringUtils.EMPTY;
        final String ENCODING = "UTF-8";
        try {
            result = String.format("http://%s:%s/%s?%s=%s", host, Integer.toString(logServer.getPort()),path, queryStringKey,URLEncoder.encode(queryStringValue, ENCODING));
        } catch (UnsupportedEncodingException e) {
            ;
        }

        return result;
    }

    public String getJobFileUrl(String simulationFilePath) {
        return getGenericUrl("gatling/lib/file","filePath",simulationFilePath);
    }

    @XmlRootElement
    public static class LogServer {
        private int port;
        private String hostName;

        public String getHostName() {
            return hostName;
        }

        public void setHostName(String hostName) {
            this.hostName = hostName;
        }

        public int getPort() {
            return port;
        }

        public void setPort(int port) {
            this.port = port;
        }

    }

    @XmlRootElement
    public static class Actor {
        private int numberOfActors;
        private int port;
        private String role;
        private String executerType;

        public String getRole() {
            return role;
        }

        public void setRole(String role) {
            this.role = role;
        }

        public int getNumberOfActors() {
            return numberOfActors;
        }

        public void setNumberOfActors(int numberOfActors) {
            this.numberOfActors = numberOfActors;
        }

        public int getPort() {
            return port;
        }

        public void setPort(int port) {
            this.port = port;
        }

        public String getExecuterType() {
            return executerType;
        }

        public void setExecuterType(String executerType) {
            this.executerType = executerType;
        }
    }

    @XmlRootElement
    public static class Job {
        private String path;
        private String logDirectory;
        private String command;
        private String artifact;
        private String mainClass;
        private String cpOrJar;
        private int[] exitValues;
        private String jobDirectory;

        public String getJobDirectory(String taskId,String postFix, String filePath) {
            return jobDirectory + taskId + "/" + postFix + "/" + filePath;
        }

        public String getJobDirectory(String taskId,String postFix) {
            return jobDirectory + taskId + "/" + postFix;
        }

        public void setJobDirectory(String jobDirectory) {
            this.jobDirectory = jobDirectory;
        }

        public String getLogDirectory() {
            return logDirectory;
        }

        public void setLogDirectory(String logDirectory) {
            this.logDirectory = logDirectory;
        }

        public String getArtifact() {
            return artifact;
        }

        public void setArtifact(String artifact) {
            this.artifact = artifact;
        }

        public String getJobArtifact(String name) {

            return MessageFormat.format(artifact, name);
        }

        public String getPath() {
            return this.path;
        }

        public void setPath(String path) {
            this.path = path;
        }

        public String getOutPath(String jobName, String jobId) {
            return String.format("%slogs/std/%s/%s.log", logDirectory, jobName, jobId);
        }

        public String getErrorPath(String jobName, String jobId) {
            return String.format("%slogs/errors/%s/%s.log", logDirectory, jobName, jobId);
        }

        public String getResultPath(String roleId,String jobId) {
            return String.format("%s/%s/%s", logDirectory, roleId, jobId);
        }

        public String getCommand() {
            return command;
        }

        public void setCommand(String command) {
            this.command = command;
        }

        public String getMainClass() {
            return mainClass;
        }

        public void setMainClass(String mainClass) {
            this.mainClass = mainClass;
        }

        public String getCpOrJar() {
            return cpOrJar;
        }

        public void setCpOrJar(String cpOrJar) {
            this.cpOrJar = cpOrJar;
        }

        public int[] getExitValues() {
            if (exitValues==null)
                return new int[]{0};
            return exitValues;
        }

        public void setExitValues(int[] exitValues) {
            this.exitValues = exitValues;
        }


    }
} public class JobSummary implements Serializable {

    private List<TaskEvent> taskInfoList;
    private JobInfo jobInfo;

    public JobSummary() {
    }

    public JobSummary(JobInfo jobInfo) {
        taskInfoList = new ArrayList<>(jobInfo.count);
        this.jobInfo = jobInfo;
    }

    public List<TaskEvent> getTaskInfoList() {
        return taskInfoList;
    }

    public void setTaskInfoList(List<TaskEvent> taskInfoList) {
        this.taskInfoList = taskInfoList;
    }

    public JobInfo getJobInfo() {
        return jobInfo;
    }

    public void setJobInfo(JobInfo jobInfo) {
        this.jobInfo = jobInfo;
    }

    public void addTask(TaskEvent taskInfo) {
        getTaskInfoList().add(taskInfo);
    }

    public void addTask(List<TaskEvent> taskInfoList) {
        getTaskInfoList().addAll(taskInfoList);
    }

    public long getStartTime() {
        Optional<TaskEvent> task = getTaskInfoList().stream().min((t1, t2) -> Long.valueOf(t1.getStartTimeStamp()).compareTo(Long.valueOf(t2.getStartTimeStamp())));
        if (task.isPresent())
            return task.get().getStartTimeStamp();
        return 0l;
    }

    public long getEndTime() {
        Optional<TaskEvent> task = getTaskInfoList().stream().max((t1, t2) -> Long.valueOf(t1.getEndTimeStamp()).compareTo(Long.valueOf(t2.getEndTimeStamp())));
        if (task.isPresent())
            return task.get().getEndTimeStamp();
        return 0l;
    }

    public String getStatus() {
        if (getTaskInfoList().stream().allMatch(p -> p.getStatus().equalsIgnoreCase(JobState.JobStatusString.COMPLETED)))
            return JobState.JobStatusString.COMPLETED;

        if (getTaskInfoList().stream().anyMatch(p -> p.getStatus().equalsIgnoreCase(JobState.JobStatusString.FAILED)))
            return JobState.JobStatusString.FAILED;

        if (getTaskInfoList().stream().anyMatch(p -> p.getStatus().equalsIgnoreCase(JobState.JobStatusString.PENDING)))
            return JobState.JobStatusString.PENDING;

        if (getTaskInfoList().stream().anyMatch(p -> p.getStatus().equalsIgnoreCase(JobState.JobStatusString.TIMEDOUT)))
            return JobState.JobStatusString.TIMEDOUT;

        if (getTaskInfoList().stream().anyMatch(p -> p.getStatus().equalsIgnoreCase(JobState.JobStatusString.STARTED)))
            return JobState.JobStatusString.STARTED;

        if (getTaskInfoList().stream().allMatch(p -> p.getStatus().equalsIgnoreCase(JobState.JobStatusString.ACCEPTED)))
            return JobState.JobStatusString.ACCEPTED;

        return JobState.JobStatusString.FAILED;

    }

    public boolean runningJob() {
        String status = getStatus();
        return status.equalsIgnoreCase(JobState.JobStatusString.STARTED) || status.equalsIgnoreCase(JobState.JobStatusString.ACCEPTED) ||
                status.equalsIgnoreCase(JobState.JobStatusString.PENDING);
    }

    public boolean completedJob() {
        return getStatus().equalsIgnoreCase(JobState.JobStatusString.COMPLETED);
    }

    public boolean failedJob() {
        return getStatus().equalsIgnoreCase(JobState.JobStatusString.FAILED);
    }

    public boolean containsWork(String workId) {
        return getTaskInfoList().stream().anyMatch(p -> p.getTaskJobId().equalsIgnoreCase(workId));
    }


    public Optional<TaskEvent> getByWork(String workId) {
        return getTaskInfoList().stream().filter(p -> p.getTaskJobId().equalsIgnoreCase(workId)).findFirst();
    }


    public static final class JobInfo implements Serializable {
        public String partitionAccessKey;
        public String user;
        public String partitionName;
        public String jobName;
        public String trackingId;
        public short count;
        public String fileFullName;
        public boolean hasDataFeed;
        public boolean hasResourcesFeed;
        public String parameterString;
        public String dataFileName;
        public String resourcesFileName;
        public String jarFileName;

        public String getFileNameFromPackageName(){
            return fileFullName.replace('.', '/') + ".scala";
        }
        public JobInfo() {
        }

        public JobInfo(String partitionAccessKey, String user, String partitionName, String jobName, String trackingId, short count, String fileFullName, boolean hasResourcesFeed, String parameterString, String resourcesFilename
        , String jarFileName) {
            this.partitionAccessKey = partitionAccessKey;
            this.user = user;
            this.partitionName = partitionName;
            this.jobName = jobName;
            this.trackingId = trackingId;
            this.count = count;
            this.fileFullName = fileFullName;
            this.hasResourcesFeed = hasResourcesFeed;
            this.parameterString = parameterString;
            this.resourcesFileName = resourcesFilename;
            this.jarFileName =   jarFileName;
        }

        private JobInfo(Builder builder) {
            partitionAccessKey = builder.partitionAccessKey;
            user = builder.user;
            partitionName = builder.partitionName;
            jobName = builder.jobName;
            trackingId = builder.trackingId;
            count = builder.count;
            fileFullName = builder.fileFullName;
            this.hasResourcesFeed = builder.hasResourcesFeed;
            this.parameterString = builder.parameterString;
            this.resourcesFileName = builder.resourcesFileName;
            this.jarFileName = builder.jarFileName;
        }

        public static Builder newBuilder() {
            return new Builder();
        }

        public static final class Builder {
            private String partitionAccessKey;
            private String user;
            private String partitionName;
            private String jobName;
            private String trackingId;
            private short count;
            private String fileFullName;
            private boolean hasResourcesFeed;
            private String parameterString;
            public String resourcesFileName;
            public String jarFileName;

            private Builder() {
            }

            public Builder withPartitionAccessKey(String partitionAccessKey) {
                this.partitionAccessKey = partitionAccessKey;
                return this;
            }
            public Builder withFileFullName(String fileFullName) {
                this.fileFullName = fileFullName;
                return this;
            }

            public Builder withUser(String user) {
                this.user = user;
                return this;
            }

            public Builder withPartitionName(String partitionName) {
                this.partitionName = partitionName;
                return this;
            }

            public Builder withJobName(String jobName) {
                this.jobName = jobName;
                return this;
            }

            public Builder withTrackingId(String trackingId) {
                this.trackingId = trackingId;
                return this;
            }

            public Builder withHasResourcesFeed(boolean hasResourcesFeed) {
            	this.hasResourcesFeed = hasResourcesFeed;
                return this;
            }

            public Builder withParameterString(String parameterString) {
                this.parameterString = parameterString;
                return this;
            }
            public Builder withCount(short count) {
                this.count = count;
                return this;
            }

            public Builder withResourcesFileName(String resourcesFileName) {
                this.resourcesFileName = resourcesFileName;
                return this;
            }

            //
            public JobInfo build() {
                return new JobInfo(this);
            }

            public Builder withJarFileName(String jarFileName) {
                this.jarFileName = jarFileName;
                return this;
            }
        }
    }


} public class Master extends AbstractPersistentActor {

    public static final Object CleanupTick = new Object() {
        @Override
        public String toString() {
            return "CleanupTick";
        }
    };
    private final ActorRef reportExecutor;
    private final FiniteDuration workTimeout;
    private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    private final Cancellable cleanupTask;
    private final AgentConfig agentConfig;
    private HashMap<String, WorkerState> workers = new HashMap<>();
    private Set<String> fileTracker = new HashSet<>();
    private JobState jobDatabase = new JobState();
    private Map<String, UploadFile> fileDatabase = new HashMap<>();
    private Set<String> cancelRequests = new HashSet<>();

    public Master(FiniteDuration workTimeout, AgentConfig agentConfig) {
        this.workTimeout = workTimeout;
        this.agentConfig = agentConfig;
        this.reportExecutor = getContext().watch(getContext().actorOf(Props.create(ReportExecutor.class, agentConfig), "report"));
        ClusterClientReceptionist.get(getContext().system()).registerService(getSelf());
        this.cleanupTask = getContext().system().scheduler().schedule(workTimeout.div(2), workTimeout.div(2), getSelf(), CleanupTick, getContext().dispatcher(), getSelf());
    }

    public static Props props(FiniteDuration workTimeout, AgentConfig agentConfig) {
        return Props.create(Master.class, workTimeout, agentConfig);
    }

    @Override
    public void postStop() {
        cleanupTask.cancel();
    }

    private void notifyWorkers() {
        if (jobDatabase.hasJob()) {
            // could pick a few random instead of all
            for (WorkerState state : workers.values()) {
                if (state.status.isIdle())
                    state.ref.tell(MasterWorkerProtocol.WorkIsReady.getInstance(), getSelf());
            }
        }
    }


    @Override
    public String persistenceId() {
        for (String role : JavaConversions.asJavaIterable((Cluster.get(getContext().system()).selfRoles()))) {
            if (role.startsWith("backend-")) {
                return role + "-master";
            }
        }
        return "master";

    }

    @Override
    public Recovery recovery() {
        return Recovery.create(100);
    }

    @Override
    public Receive createReceiveRecover() {
        return receiveBuilder()
                .match(JobDomainEvent.class, p -> {
                    jobDatabase = jobDatabase.updated(p);
                    log.info("Replayed {}", p.getClass().getSimpleName());
                })
                .match(UploadFile.class, p -> {
                    log.info("Replayed {}", p.getClass().getSimpleName());
                })
                .build();
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(MasterWorkerProtocol.RegisterWorker.class, cmd -> onRegisterWorker(cmd))
                .match(MasterWorkerProtocol.WorkerRequestsFile.class, cmd -> onWorkerRequestsFile(cmd))
                .match(MasterWorkerProtocol.WorkerRequestsWork.class, cmd -> onWorkerRequestsWork(cmd))
                .match(Worker.FileUploadComplete.class, cmd -> onFileUploadComplete(cmd))
                .match(MasterWorkerProtocol.WorkInProgress.class, cmd -> onWorkInProgress(cmd))
                .match(MasterWorkerProtocol.WorkIsDone.class, cmd -> onWorkIsDone(cmd))
                .match(MasterWorkerProtocol.WorkFailed.class, cmd -> onWorkFailed(cmd))
                .match(UploadInfo.class, cmd -> onUploadInfo(cmd))
                .match(ServerInfo.class, cmd -> onServerInfo(cmd))
                .match(TrackingInfo.class, cmd -> onTrackingInfo(cmd))
                .match(Report.class, cmd -> onReport(cmd))
                .match(UploadFile.class, cmd -> onUploadFile(cmd))
                .match(Job.class, cmd -> onJob(cmd))
                .match(MasterClientProtocol.CommandLineJob.class, cmd -> processCmdLineJob(cmd))
                .match(JobSummaryInfo.class, cmd -> onJobSummary())
                .matchEquals(CleanupTick, cmd -> onCleanupTick())
                .matchAny(cmd -> unhandled(cmd))
                .build();
    }


    private void onJobSummary() {
        getSender().tell(ImmutableList.copyOf(jobDatabase.getJobSummary().values()), getSelf());
    }

    private void onCleanupTick() {
        Iterator<Map.Entry<String, WorkerState>> iterator = workers.entrySet().iterator();
        Set<String> tobeRemoved = new HashSet<>();
        while (iterator.hasNext()) {
            Map.Entry<String, WorkerState> entry = iterator.next();
            String workerId = entry.getKey();
            WorkerState state = entry.getValue();
            if (state.status.isBusy()) {
                if (state.status.getDeadLine().isOverdue()) {
                    log.info("Work timed out: {}", state.status.getWorkId());
                    tobeRemoved.add(workerId);
                    persist(new JobState.JobTimedOut(state.status.getWorkId()), event -> {
                        // remove from in progress to pending
                        jobDatabase = jobDatabase.updated(event);
                        notifyWorkers();
                    });
                }
            }
            else {
                if (state.status.getDeadLine().isOverdue()) { //we missed pings from worker or worker is dead
                    tobeRemoved.add(workerId);
                }
            }
        }
        for (String workerId : tobeRemoved) {
            workers.remove(workerId);
        }

    }

    public Optional<String> processCmdLineJob(MasterClientProtocol.CommandLineJob cmdJob) {
        ClientConfig clientConfig = cmdJob.clientConfig;
        getSender().tell(new Ack(cmdJob.clientId), getSelf());
        String trackingId = UUID.randomUUID().toString();
        List<String> parameters = Arrays.asList(clientConfig.getParameterString().split(" "));
        boolean hasResourcesFeed = !clientConfig.getResourcesFeedPath().isEmpty();
        JobSummary.JobInfo jobinfo = JobSummary.JobInfo.newBuilder()
                .withCount(clientConfig.getParallelism())
                .withJobName("gatling")
                .withPartitionAccessKey(clientConfig.getAccessKey())
                .withPartitionName(clientConfig.getPartitionName())
                .withUser(clientConfig.getUserName())
                .withTrackingId(trackingId)
                .withHasResourcesFeed(hasResourcesFeed)
                .withParameterString(clientConfig.getParameterString())
                .withFileFullName(clientConfig.getJarFileName())//class name is not fileName
                .withResourcesFileName((clientConfig.getResourcesFeedFileName()))
                .withJarFileName(clientConfig.getJarFileName())
                .build();
        for (int i = 0; i < clientConfig.getParallelism(); i++) {
            TaskEvent taskEvent = new TaskEvent();
            taskEvent.setJobName("gatling"); //the gatling.sh script is the gateway for simulation files
            taskEvent.setJobInfo(jobinfo);
            taskEvent.setParameters(new ArrayList<>(parameters));
            Job job = new Job(clientConfig.getPartitionName(), taskEvent, trackingId,
                    agentConfig.getAbortUrl(),
                    agentConfig.getJobFileUrl(clientConfig.getJarPath()),
                    agentConfig.getJobFileUrl(clientConfig.getResourcesFeedPath()),
                    true);
            persist(new JobState.JobAccepted(job), event -> {
                // Ack back to original sender
                getSender().tell(new MasterClientProtocol.CommandLineJobAccepted(job), getSelf());
                jobDatabase = jobDatabase.updated(event);
                notifyWorkers();
            });

        }
        getSender().tell(new MasterClientProtocol.CommandLineJobSubmitted(trackingId), getSelf());
        return Optional.of(trackingId);
    }

    private void onJob(Job cmd) {
        final String workId = cmd.jobId;
        // idempotent
        if (jobDatabase.isAccepted(workId)) {
            getSender().tell(new Ack(workId), getSelf());
        } else {
            log.info("Accepted work: {}", workId);
            persist(new JobState.JobAccepted(cmd), event -> {
                // Ack back to original sender
                getSender().tell(new Ack(event.job.jobId), getSelf());
                jobDatabase = jobDatabase.updated(event);
                notifyWorkers();
            });
        }
    }

    private void onUploadFile(Object cmd) {
        log.info("Accepted upload file request: {}", cmd);
        UploadFile request = (UploadFile) cmd;
        persist(request, event -> {
            getSender().tell(new Ack(request.trackingId), getSelf());
            fileDatabase.put(request.trackingId, request);
        });
    }

    private void onReport(Object cmd) {
        log.info("Accepted report request: {}", cmd);
        List<Worker.Result> result = jobDatabase.getCompletedResults(((Report) cmd).trackingId);
        reportExecutor.forward(new GenerateReport((Report) cmd, result), getContext());
    }

    private void onTrackingInfo(Object cmd) {
        TrackingInfo trackingInfo = (TrackingInfo) cmd;
        log.info("Accepted tracking info request: {}", cmd);
        TrackingResult result = jobDatabase.getTrackingInfo(trackingInfo.trackingId);
        log.info("Complete tracking info request: {}", result);
        if (trackingInfo.cancel) {
            cancelRequests.add(trackingInfo.trackingId);
        }
        result.setCancelled(cancelRequests.contains(trackingInfo.trackingId));
        getSender().tell(result, getSelf());
    }

    private void onServerInfo(Object cmd) {
        log.info("Accepted Server info request: {}", cmd);
        getSender().tell(new ServerInfo(workers), getSelf());
    }

    private void onUploadInfo(Object cmd) {
        log.info("Accepted Upload info request: {}", cmd);
        UploadInfo info = (UploadInfo) cmd;
        info.setHosts(fileTracker.stream().filter(p -> p.contains(info.trackingId)).map(p -> p.split("_")[1]).collect(Collectors.toList()));
        getSender().tell(info, getSelf());
    }

    private void onWorkFailed(MasterWorkerProtocol.WorkFailed cmd) {
        final String workId = cmd.workId;
        final String workerId = cmd.workerId;
        log.info("Work {} failed by worker {}", workId, workerId);
        if (jobDatabase.isInProgress(workId)) {
            changeWorkerToIdle(workerId);
            persist(new JobState.JobFailed(workId, cmd.result), event -> {
                jobDatabase = jobDatabase.updated(event);
                notifyWorkers();
            });
        }
    }

    private void onWorkIsDone(MasterWorkerProtocol.WorkIsDone cmd) {
        MasterWorkerProtocol.WorkIsDone workDone = cmd;
        final String workerId = workDone.workerId;
        final String workId = workDone.workId;
        if (jobDatabase.isDone(workId)) {
            getSender().tell(new Ack(workId), getSelf());
        } else if (!jobDatabase.isInProgress(workId)) {
            log.info("Work {} not in progress, reported as done by worker {}", workId, workerId);
        } else {
            log.info("Work {} is done by worker {}", workId, workerId);
            changeWorkerToIdle(workerId);
            persist(new JobState.JobCompleted(workId, cmd.result), event -> {
                jobDatabase = jobDatabase.updated(event);
                getSender().tell(new Ack(event.workId), getSelf());
            });
        }
    }

    private void onWorkInProgress(MasterWorkerProtocol.WorkInProgress cmd) {
        final String workerId = cmd.workerId;
        final String workId = cmd.workId;
        final WorkerState state = workers.get(workerId);
        if (jobDatabase.isInProgress(workId)) {
            if (state != null && state.status.isBusy()) {
                workers.put(workerId, state.copyWithStatus(new Busy(state.status.getWorkId(), workTimeout.fromNow())));
            }
        } else {
            log.info("Work {} not in progress, reported as in progress by worker {}", workId, workerId);
        }
    }

    private void onFileUploadComplete(Worker.FileUploadComplete cmd) {
        Worker.FileUploadComplete result = cmd;
        fileTracker.add(result.result.trackingId + "_" + result.host);
    }

    private void onWorkerRequestsWork(MasterWorkerProtocol.WorkerRequestsWork cmd) {
        log.info("Worker requested work: {}", cmd);
        MasterWorkerProtocol.WorkerRequestsWork workReqMsg = cmd;
        final String workerId = workReqMsg.workerId;
        if (jobDatabase.hasJob()) {
            final WorkerState state = workers.get(workerId);
            if (state != null && state.status.isIdle()) {
                //for (int i=0; i<jobDatabase.getPendingJobsCount(); i++) {
                final Job job = jobDatabase.nextJob();//nextJob for the partition/role
                boolean jobWorkerRoleMatched = workReqMsg.role.equalsIgnoreCase(job.roleId);
                if (jobWorkerRoleMatched) {
                    persist(new JobState.JobStarted(job.jobId, workerId), event -> {
                        jobDatabase = jobDatabase.updated(event);
                        log.info("Giving worker {} some taskEvent {}", workerId, event.workId);
                        workers.put(workerId, state.copyWithStatus(new Busy(event.workId, workTimeout.fromNow())));
                        getSender().tell(job, getSelf());
                    });
                } else {
                    persist(new JobState.JobPostponed(job.jobId), event -> {
                        jobDatabase = jobDatabase.updated(event);
                        log.info("Postponing work: {}", workerId);
                    });
                    extendIdleExpiryTime(workerId);
                }
                //}
            }
        }
        else {
            extendIdleExpiryTime(workerId);
        }
    }

    private void extendIdleExpiryTime(String workerId) {
        if(workers.get(workerId).status.isIdle()) {
            workers.put(workerId, workers.get(workerId).copyWithStatus(new Idle(workTimeout.fromNow())));
        }
    }

    private void onWorkerRequestsFile(MasterWorkerProtocol.WorkerRequestsFile cmd) throws IOException {
        MasterWorkerProtocol.WorkerRequestsFile msg = cmd;
        Set<String> trackingIds = fileDatabase.keySet();
        Optional<String> unsentFile = trackingIds.stream().map(p -> p.concat("_").concat(msg.host)).filter(p -> !fileTracker.contains(p)).findFirst();
        if (unsentFile.isPresent()) {
            String tId = unsentFile.get().split("_")[0];
            UploadFile uploadFile = fileDatabase.get(tId);
            if (uploadFile.type.equalsIgnoreCase("lib")) {
                String soonToBeRemotePath = agentConfig.getMasterUrl(uploadFile.path);
                getSender().tell(new FileJob(null, uploadFile, soonToBeRemotePath), getSelf());
            } else {
                String content = FileUtils.readFileToString(new File(uploadFile.path));
                getSender().tell(new FileJob(content, uploadFile, null), getSelf());
            }
        }
    }

    private void onRegisterWorker(MasterWorkerProtocol.RegisterWorker cmd) {
        String workerId = cmd.workerId;
        if (workers.containsKey(workerId)) {
            workers.put(workerId, workers.get(workerId).copyWithRef(getSender()));
        } else {
            log.info("Worker registered: {}", workerId);
            workers.put(workerId, new WorkerState(getSender(), new Idle(workTimeout.fromNow())));
            if (jobDatabase.hasJob()) {
                getSender().tell(MasterWorkerProtocol.WorkIsReady.getInstance(), getSelf());
            }
        }
    }

    private void changeWorkerToIdle(String workerId) {
        if (workers.get(workerId).status.isBusy()) {
            workers.put(workerId, workers.get(workerId).copyWithStatus(new Idle(workTimeout.fromNow())));
        }
    }

    public static abstract class WorkerStatus {
        protected abstract boolean isIdle();

        private boolean isBusy() {
            return !isIdle();
        }

        protected abstract String getWorkId();

        protected abstract Deadline getDeadLine();
    }

    private static final class Idle extends WorkerStatus {
        private final Deadline deadline;

        private Idle(Deadline deadline) {
            this.deadline = deadline;
        }

        @Override
        protected boolean isIdle() {
            return true;
        }

        @Override
        protected String getWorkId() {
            throw new IllegalAccessError();
        }

        @Override
        protected Deadline getDeadLine() {
            return deadline;
        }

        @Override
        public String toString() {
            return "Idle";
        }
    }

    private static final class Busy extends WorkerStatus {
        private final String workId;
        private final Deadline deadline;

        private Busy(String workId, Deadline deadline) {
            this.workId = workId;
            this.deadline = deadline;
        }

        @Override
        protected boolean isIdle() {
            return false;
        }

        @Override
        protected String getWorkId() {
            return workId;
        }

        @Override
        protected Deadline getDeadLine() {
            return deadline;
        }

        @Override
        public String toString() {
            return "Busy";
        }
    }

    public static final class WorkerState {
        public final ActorRef ref;
        public final WorkerStatus status;

        private WorkerState(ActorRef ref, WorkerStatus status) {
            this.ref = ref;
            this.status = status;
        }

        private WorkerState copyWithRef(ActorRef ref) {
            return new WorkerState(ref, this.status);
        }

        private WorkerState copyWithStatus(WorkerStatus status) {
            return new WorkerState(this.ref, status);
        }

        @Override
        public boolean equals(Object o) {
            if (this == o)
                return true;
            if (o == null || !getClass().equals(o.getClass()))
                return false;

            WorkerState that = (WorkerState) o;

            return ref.equals(that.ref) && status.equals(that.status);

        }

        @Override
        public int hashCode() {
            int result = ref.hashCode();
            result = 31 * result + status.hashCode();
            return result;
        }

        @Override
        public String toString() {
            return "WorkerState{" + "ref=" + ref + ", status=" + status + '}';
        }
    }

    public static final class Job implements Serializable {
        public final Object taskEvent;//task
        public final String jobId;
        public final String roleId;
        public final String trackingId;
        public final boolean isJarSimulation;
        public String abortUrl;
        public String jobFileUrl;
        public String resourcesFileUrl;

        public Job(String roleId, Object job, String trackingId, String abortUrl, String jobFileUrl, String resourcesFileUrl, boolean isJarSimulation) {
            this.jobId = UUID.randomUUID().toString();
            this.roleId = roleId;
            this.taskEvent = job;
            this.trackingId = trackingId;
            this.abortUrl = abortUrl;
            this.jobFileUrl = jobFileUrl;
            this.resourcesFileUrl = resourcesFileUrl;
            this.isJarSimulation = isJarSimulation;
        }

        @Override
        public String toString() {
            return "Job{" +
                    "taskEvent=" + taskEvent +
                    ", jobId='" + jobId + '\'' +
                    ", roleId='" + roleId + '\'' +
                    ", trackingId='" + trackingId + '\'' +
                    ", isJarSimulation=" + isJarSimulation +
                    ", abortUrl='" + abortUrl + '\'' +
                    ", jobFileUrl='" + jobFileUrl + '\'' +
                    ", resourcesFileUrl='" + resourcesFileUrl + '\'' +
                    '}';
        }
    }

    public static final class FileJob implements Serializable {
        public final String content;//task
        public final String remotePath;//task
        public final String jobId;
        public final UploadFile uploadFileRequest;

        public FileJob(String content, UploadFile uploadFileRequest, String remotePath) {
            this.jobId = UUID.randomUUID().toString();
            this.uploadFileRequest = uploadFileRequest;
            this.content = content;
            this.remotePath = remotePath;
        }

        @Override
        public String toString() {
            return "FileJob{" +
                    "content='" + content + '\'' +
                    ", remotePath='" + remotePath + '\'' +
                    ", jobId='" + jobId + '\'' +
                    ", uploadFileRequest=" + uploadFileRequest +
                    '}';
        }
    }

    public static final class UploadFile implements Serializable {
        public final String trackingId;
        public final String path, name, role, type;

        public UploadFile(String trackingId, String path, String name, String role, String type) {
            this.trackingId = trackingId;
            this.path = path;
            this.name = name;
            this.role = role;
            this.type = type;
        }

        public String getFileName() {
            if (type.equalsIgnoreCase("conf") || type.equalsIgnoreCase("lib"))
                return type + "/" + name;
            return "user-files/" + type + "/" + name;
        }

        @Override
        public String toString() {
            return "UploadFile{" +
                    "trackingId='" + trackingId + '\'' +
                    ", path='" + path + '\'' +
                    ", name='" + name + '\'' +
                    ", role='" + role + '\'' +
                    ", type='" + type + '\'' +
                    '}';
        }
    }

    public static final class Report implements Serializable {
        public final String trackingId;
        public final TaskEvent taskEvent;

        public Report(String trackingId, TaskEvent taskEvent) {
            this.trackingId = trackingId;
            this.taskEvent = taskEvent;
        }

        @Override
        public String toString() {
            return "Report{" +
                    "trackingId='" + trackingId + '\'' +
                    '}';
        }

        public String getHtml() {
            return "/resources/" + trackingId + "/index.html";
        }
    }

    public static final class GenerateReport implements Serializable {
        public final Report reportJob;
        public final List<Worker.Result> results;

        public GenerateReport(Report repotJob, List<Worker.Result> results) {
            this.reportJob = repotJob;
            this.results = results;
        }

        @Override
        public String toString() {
            return "GenerateReport{" +
                    "reportJob=" + reportJob +
                    ", results=" + results +
                    '}';
        }
    }

    public static final class TrackingInfo implements Serializable {
        public final String trackingId;
        public final boolean cancel;

        public TrackingInfo(String trackingId) {
            this.trackingId = trackingId;
            this.cancel = false;
        }

        public TrackingInfo(String trackingId, boolean cancel) {
            this.trackingId = trackingId;
            this.cancel = cancel;
        }

        @Override
        public String toString() {
            return "TrackingInfo{" +
                    "trackingId='" + trackingId + '\'' +
                    ", cancel=" + cancel +
                    '}';
        }
    }

    public static final class UploadInfo implements Serializable {
        public final String trackingId;
        public List<String> hosts;

        public UploadInfo(String trackingId) {
            this.trackingId = trackingId;
        }

        public void setHosts(List<String> hosts) {
            this.hosts = hosts;
        }

        @Override
        public String toString() {
            return "UploadInfo{" +
                    "trackingId='" + trackingId + '\'' +
                    ", hosts=" + hosts +
                    '}';
        }
    }

    public static final class JobSummaryInfo implements Serializable {
    }

    public static final class ServerInfo implements Serializable {

        private ImmutableMap<String, WorkerState> workers;

        public ServerInfo() {
        }

        public ServerInfo(HashMap<String, WorkerState> workers) {
            this.workers = ImmutableMap.copyOf(workers);
        }

        public ImmutableMap<String, WorkerState> getWorkers() {
            return workers == null ? ImmutableMap.of() : workers;
        }

        @Override
        public String toString() {
            return "ServerInfo{" +
                    "workers=" + getWorkers() +
                    '}';
        }
    }

    public static final class Ack implements Serializable {
        final String workId;

        public Ack(String workId) {
            this.workId = workId;
        }

        public String getWorkId() {
            return workId;
        }

        @Override
        public String toString() {
            return "Ack{" + "jobId='" + workId + '\'' + '}';
        }
    }
} public class MasterClientActor extends AbstractActor {
  private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  private ActorRef masterProxy;

  public  MasterClientActor(ActorSystem system,String masterName){
    ClusterSingletonProxySettings proxySettings =
            ClusterSingletonProxySettings.create(system).withRole(masterName);
    masterProxy = system.actorOf(ClusterSingletonProxy.props("/user/master", proxySettings), "masterProxy" + UUID.randomUUID());
  }

  public void receiveHandler(Object message) {
    log.debug("Master client received: {}",message);
    Timeout timeout = new Timeout(120, TimeUnit.SECONDS);
    Future<Object> future = ask(masterProxy, message, timeout);
    final ExecutionContext ec = getContext().system().dispatcher();

    Future<Object> res = future.map(new Mapper<Object, Object>() {
      @Override
      public Object apply(Object msg) {
          return new Ok(msg);
      }
    }, ec).recover(new Recover<Object>() {
      @Override
      public Object recover(Throwable failure) throws Throwable {
        return new NotOk(null);
      }
    }, ec);

    pipe(res, ec).to(getSender());
  }

  @Override
  public Receive createReceive() {
    return receiveBuilder()
            .match(Object.class, cmd -> receiveHandler(cmd))
            .build();
  }

  public  static class Ok implements Serializable {
    private Object msg;
    private Ok(Object msg) {
      this.msg=msg;
    }
    public Object getMsg() {
      return msg;
    }

    @Override
    public String toString() {
      return "Ok";
    }
  }


  public static  class NotOk implements Serializable {
    private Object msg;
    private NotOk(Object msg) {
      this.msg=msg;
    }
    public Object getMsg() {
      return msg;
    }

    @Override
    public String toString() {
      return "NotOk";
    }
  }
} public class ReportExecutor extends WorkExecutor {
    private AgentConfig agentConfig;

    public ReportExecutor(AgentConfig agentConfig){
        this.agentConfig = agentConfig;
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Master.GenerateReport.class, cmd -> runJob(cmd))
                .matchAny(cmd  -> unhandled(cmd))
                .build();
    }

    private void runJob(Master.GenerateReport job) {
        TaskEvent taskEvent = job.reportJob.taskEvent;
        CommandLine cmdLine = new CommandLine(agentConfig.getJob().getCommand());
        Map<String, Object> map = new HashMap<>();

        map.put("path", new File(agentConfig.getJob().getJobArtifact(taskEvent.getJobName())));
        cmdLine.addArgument("${path}");

        //parameters come from the task event
        for (String pair : taskEvent.getParameters()) {
            cmdLine.addArgument(pair);
        }
        String dir = agentConfig.getJob().getLogDirectory()+ "reports/" + job.reportJob.trackingId + "/";
        cmdLine.addArgument(dir);

        cmdLine.setSubstitutionMap(map);
        DefaultExecutor executor = new DefaultExecutor();
        executor.setExitValues(agentConfig.getJob().getExitValues());
        ExecuteWatchdog watchdog = new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT);
        executor.setWatchdog(watchdog);
        executor.setWorkingDirectory(new File(agentConfig.getJob().getPath()));
        FileOutputStream outFile = null;
        FileOutputStream errorFile = null;
        try {
            List<String> resultFiles = new ArrayList<>(job.results.size());
            //download all files adn
            /*int i=0;
            for (Worker.Result result : job.results) {
                String destFile = dir  + i++ + ".log";
                resultFiles.add(destFile);
                DownloadFile.downloadFile(result.metrics,destFile);
            }*/
            AtomicInteger index = new AtomicInteger();
            job.results.parallelStream().forEach(result -> {
                String destFile = dir  + index.incrementAndGet() + ".log";
                resultFiles.add(destFile);
                DownloadFile.downloadFile(result.metrics,destFile);
            });
            String outPath = agentConfig.getJob().getOutPath(taskEvent.getJobName(), job.reportJob.trackingId);
            String errPath = agentConfig.getJob().getErrorPath(taskEvent.getJobName(), job.reportJob.trackingId);
            //create the std and err files
            outFile = FileUtils.openOutputStream(new File(outPath));
            errorFile = FileUtils.openOutputStream(new File(errPath));

            PumpStreamHandler psh = new PumpStreamHandler(new ExecLogHandler(outFile),new ExecLogHandler(errorFile));

            executor.setStreamHandler(psh);
            System.out.println(cmdLine);
            int exitResult = executor.execute(cmdLine);
            ReportResult result ;
            if(executor.isFailure(exitResult)){
                result = new ReportResult(dir,job.reportJob, false);
                log.info("Report Executor Failed, result: " +job.toString());
            }
            else{
                result = new ReportResult(job.reportJob.getHtml() ,job.reportJob, true);
                log.info("Report Executor Completed, result: " +result.toString());
            }
            for (String resultFile : resultFiles) {
                FileUtils.deleteQuietly(new File(resultFile));
            }
            getSender().tell(result, getSelf());

        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }finally {
            IOUtils.closeQuietly(outFile);
            IOUtils.closeQuietly(errorFile);
        }

    }


    class ExecLogHandler extends LogOutputStream {
        private  FileOutputStream file;

        public ExecLogHandler(FileOutputStream file) {
            this.file = file;
        }

        @Override
        protected void processLine(String line, int level) {
            try {
                IOUtils.write(line, file);
            } catch (IOException e) {
                //e.printStackTrace();
            }
        }
    }

    public static final class ReportResult implements Serializable {
        public final Object result;
        public final Master.Report  report;
        public final boolean success;
        public ReportResult(Object result, Master.Report report, boolean success) {
            this.result = result;
            this.report = report;
            this.success = success;
        }

        @Override
        public String toString() {
            return "ReportResult{" +
                    "result=" + result +
                    ", report=" + report +
                    ", Success=" + success +
                    '}';
        }
    }
} public class TaskEvent implements Serializable {

    private List<String> parameters;
    private String jobName;
    private long startTimeStamp;
    private long endTimeStamp;
    private String workerId;
    private String errorLogPath;
    private String stdLogPath;
    private String status;//Master.JobStatusString
    private String taskJobId;
    private JobSummary.JobInfo jobInfo;

    public TaskEvent() {
        parameters = new ArrayList<>();
    }

    public String getTaskJobId() {
        return taskJobId;
    }

    public void setTaskJobId(String taskJobId) {
        this.taskJobId = taskJobId;
    }

    public JobSummary.JobInfo getJobInfo() {
        return jobInfo;
    }

    public void setJobInfo(JobSummary.JobInfo jobInfo) {
        this.jobInfo = jobInfo;
    }

    public List<String> getParameters() {
        return parameters;
    }

    public void setParameters(List<String> parameters) {
        this.parameters = parameters;
    }

    public String getJobName() {
        return jobName;
    }

    public void setJobName(String jobName) {
        this.jobName = jobName;
    }

    public long getStartTimeStamp() {
        return startTimeStamp;
    }

    public void setStartTimeStamp(long startTimeStamp) {
        this.startTimeStamp = startTimeStamp;
    }

    public long getEndTimeStamp() {
        return endTimeStamp;
    }

    public void setEndTimeStamp(long endTimeStamp) {
        this.endTimeStamp = endTimeStamp;
    }

    public String getWorkerId() {
        return workerId;
    }

    public void setWorkerId(String workerId) {
        this.workerId = workerId;
    }

    public String getErrorLogPath() {
        return errorLogPath;
    }

    public void setErrorLogPath(String errorLogPath) {
        this.errorLogPath = errorLogPath;
    }

    public String getStdLogPath() {
        return stdLogPath;
    }

    public void setStdLogPath(String stdLogPath) {
        this.stdLogPath = stdLogPath;
    }

    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }
} @XmlRootElement
public class TrackingResult {

    private final long pendingCount;
    private final long inProgressCount;
    private boolean cancelled;
    private List<TaskTrackingInfo> Completed;
    private List<TaskTrackingInfo> Failed;
    public TrackingResult(long pendingCount, long inProgressCount) {
        this.pendingCount = pendingCount;
        this.inProgressCount = inProgressCount;
    }

    public boolean isCancelled() {
        return cancelled;
    }

    public void setCancelled(boolean cancelled) {
        this.cancelled = cancelled;
    }

    public long getPendingCount() {
        return pendingCount;
    }

    public long getInProgressCount() {
        return inProgressCount;
    }

    public List<TaskTrackingInfo> getCompleted() {
        return Completed;
    }

    public void setCompleted(List<TaskTrackingInfo> completed) {
        Completed = completed;
    }

    public List<TaskTrackingInfo> getFailed() {
        return Failed;
    }

    public void setFailed(List<TaskTrackingInfo> failed) {
        Failed = failed;
    }

    @Override
    public String toString() {
        return "TrackingResult{" +
                "pendingCount=" + pendingCount +
                ", inProgressCount=" + inProgressCount +
                ", Completed=" + Completed +
                ", Failed=" + Failed +
                '}';
    }
} @Data
@XmlRootElement
public class SimulationJobModel implements Serializable {
    private String partitionAccessKey;
    private String user;
    private String roleId; // The unique partition name/id
    private String simulation;
    private String resourcesFile;
    private String jobId;
    private String tag;
    private short count;
    private String fileFullName;
    private String parameterString;

} public class NotFoundException extends Exception {

} public class UnknownResourceException extends Exception {

	public UnknownResourceException(String msg) {
		super(msg);
	}
	
}
import akka.actor.ActorRef;
import akka.util.Timeout;
import com.alh.gatling.commons.AgentConfig;
import com.alh.gatling.commons.JobSummary;
import com.alh.gatling.commons.Master;
import com.alh.gatling.commons.MasterClientActor;
import com.alh.gatling.commons.ReportExecutor;
import com.alh.gatling.commons.TaskEvent;
import com.alh.gatling.commons.TrackingResult;
import com.alh.gatling.domain.SimulationJobModel;
import com.alh.gatling.service.exception.NotFoundException;
import com.alh.gatling.service.exception.UnknownResourceException;
import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.concurrent.Await;
import scala.concurrent.Future;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import static akka.pattern.Patterns.ask;
                .withPartitionName(simulationJobModel.getRoleId())
                .withUser(simulationJobModel.getUser())
                .withTrackingId(trackingId)
                .withHasResourcesFeed(hasResourcesFeed)
                .withParameterString(simulationJobModel.getParameterString())
                .withFileFullName(simulationJobModel.getFileFullName())
                .withResourcesFileName(getResourcesFileName(simulationJobModel, hasResourcesFeed))
                .build();
        Timeout timeout = new Timeout(6, TimeUnit.SECONDS);
        int success = 0;
        for (int i = 0; i < simulationJobModel.getCount(); i++) {
            TaskEvent taskEvent = new TaskEvent();
            taskEvent.setJobName("gatling"); //the gatling.sh script is the gateway for simulation files
            taskEvent.setJobInfo(jobinfo);
            taskEvent.setParameters(new ArrayList<>(parameters));
            Future<Object> future = ask(router, new Master.Job(simulationJobModel.getRoleId(), taskEvent, trackingId,
                    agentConfig.getAbortUrl(),
                    agentConfig.getJobFileUrl(simulationJobModel.getSimulation()),
                    agentConfig.getJobFileUrl(simulationJobModel.getResourcesFile()),
                    false),
                    timeout);
            Object result = Await.result(future, timeout.duration());
            if (result instanceof MasterClientActor.Ok) {
                success++;
                log.debug("Ok message from server just got here, this indicates the job was successfully posted to the master: {}", result);
            }
        }

        if (success == simulationJobModel.getCount()) {
            log.debug("Job Successfully submitted to master");
            return Optional.of(trackingId);
        }
        return Optional.empty();
    }

    private String getResourcesFileName(SimulationJobModel simulationJobModel, boolean hasBodiesFeed) {
        if(!hasBodiesFeed)
            return "";
        String[] splits = simulationJobModel.getResourcesFile().split("/");
        if (splits.length > 0) //return the last name
            return splits[splits.length-1];
        return "";
    }

    public TrackingResult getTrackingInfo(String trackingId) {
        Object result = sendToMaster(new Master.TrackingInfo(trackingId), 60);
        if (result != null && result instanceof TrackingResult) {
            TrackingResult info = (TrackingResult) result;
            return info;
        }
        return new TrackingResult(0, 0);
    }

    public boolean abortJob(String trackingId) {
        Object result = sendToMaster(new Master.TrackingInfo(trackingId, true), 60);
        if (result != null && result instanceof TrackingResult) {
            TrackingResult info = (TrackingResult) result;
            return info.isCancelled();
        }
        return false;
    }

    public Optional<ReportExecutor.ReportResult> generateReport(String trackingId) {

        TaskEvent taskEvent = new TaskEvent();
        taskEvent.setJobName("gatling");
        taskEvent.setParameters(new ArrayList<>(Arrays.asList("-ro")));
        Object result = sendToMaster(new Master.Report(trackingId, taskEvent), 180);

        log.info("Report generated {}", result);
        if (result != null && result instanceof ReportExecutor.ReportResult) {
            log.info("Report generated accurately {}", result);
            return Optional.of((ReportExecutor.ReportResult) result);
        }
        return Optional.empty();
    }

    /**
     * Given the path from the staging area the master instructs workers to pull new files supplied by users
     *
     * @param path
     * @param name
     * @param role
     * @param type
     * @return
     */
    public Optional<String> uploadFile(String path, String name, String role, String type) {
        String trackingId = UUID.randomUUID().toString();
        Master.UploadFile uploadFileRequest = new Master.UploadFile(trackingId, path, name, role, type);
        Object result = sendToMaster(uploadFileRequest, 5);
        log.info("UploadFile request sent {}", result);
        if (result != null && result instanceof Master.Ack) {
            return Optional.of(((Master.Ack) result).getWorkId());
        }
        return Optional.empty();
    }

    /**
     * Tracks file upload status and indicates which workers have successfully pulled the new files
     *
     * @param uploadInfo
     * @return
     */
    public Optional<Master.UploadInfo> getUploadStatus(Master.UploadInfo uploadInfo) {
        Object result = sendToMaster(uploadInfo, 5);
        if (result != null && result instanceof Master.UploadInfo) {
            return Optional.of((Master.UploadInfo) result);
        }
        return Optional.empty();
    }


    public List<JobSummary> getJobSummary() {
        Object result = sendToMaster(new Master.JobSummaryInfo(), 60);

        List<JobSummary> summaries = (List<JobSummary>) result;
        return summaries;

    }

	public String getLogResult(String trackingId, String taskJobId, String logType) throws NotFoundException, UnknownResourceException{