function jobStatusData = getJobStatusData(jobs)
% Allocate the job corruption-related cell array

%   Copyright 2018-2020 The MathWorks, Inc.

corruptionInfos = cell(numel(jobs), 1);

% Properties of jobs which we require to fill in the table.
props = {'Username', 'CreateDateTime', 'SubmitDateTime', 'FinishDateTime', ...
    'ApiTag', 'Tag'};
data  = cell(numel(jobs), numel(props));
% Get State separately since it requires using the hGetStateEnumForDisplay
% method to get it in a vectorized manner.
states = cell(numel(jobs), 1);
% Get the job ids separately as it does not require a remote call and must 
% be filled in even for corrupt jobs.
jobIDs = arrayfun(@(j) j.ID, jobs, 'UniformOutput', false);
jobIDs = reshape(jobIDs, numel(jobs), 1);

try
    statesFcn = @(x) hGetStateEnumForDisplay(x);
    stateEnums = parallel.internal.cluster.hetfun(statesFcn, jobs, 'stable');
    stateEnums = vertcat(stateEnums{:});
    states = cellfun(@(s) s.Name, stateEnums, 'UniformOutput', false);
    data = get(jobs, props);
catch E
    dctSchedulerMessage(1, 'Exception thrown while fetching job states', E)
    % If this fails, attempt to get properties one by one
    % to find out which one was actually failing
    for index = 1:numel(jobs)
        try
            states{index} = get(jobs(index), 'State');
            data(index, :) = get(jobs(index), props);
        catch E
            corruptionInfos{index}{end+1} = parallel.internal.cluster.JobCorruptionInfo(...
                parallel.internal.cluster.types.JobCorruptionLevel.CannotProvideProperties, ...
                E.message);
        end
    end
end

userNames    = data(:,1);
createTimes  = data(:,2);
submitTimes  = data(:,3);
finishTimes  = data(:,4);
apitags      = data(:,5);
tags         = data(:,6);

[jobTypes, corruptionInfos] = iGetJobTypes(jobs, corruptionInfos);

[tempJobStatus, corruptionInfos] = parallel.internal.cluster.getStatusFromJob(jobs, corruptionInfos, states);
jobStatusCell = num2cell(tempJobStatus);

[firstTaskFunctionNames, corruptionInfos] = iGetFirstTaskFunctionNames(jobs, corruptionInfos);

% Create the actual job table
allDataStructure = struct(...
    'Id', jobIDs, ...
    'Tag', tags, ...
    'UserName', userNames, ...
    'CreateDateTime', createTimes, ...
    'SubmitDateTime', submitTimes, ...
    'FinishDateTime', finishTimes, ...
    'Status', jobStatusCell, ...
    'Type', jobTypes, ...
    'ApiTag', apitags, ...
    'FirstTaskFunction', firstTaskFunctionNames, ...
    'CorruptionInfo', corruptionInfos);

% Build JobStatusData
jobStatusData = arrayfun(@iGetJobStatusData, allDataStructure);
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function jobStatusData = iGetJobStatusData(allDataStructure)
jobStatesStructure = allDataStructure.Status;
corruptionInfo = allDataStructure.CorruptionInfo;
if ~isempty(corruptionInfo)
    corruptionInfo = cellfun(@(x) x, corruptionInfo); % Convert to array preserving size
end

jobStatusData = parallel.internal.cluster.JobStatusData(allDataStructure.Id, ...
    allDataStructure.Type,...
    allDataStructure.UserName,...
    allDataStructure.CreateDateTime,...
    allDataStructure.SubmitDateTime,...
    allDataStructure.FinishDateTime,...
    jobStatesStructure.State,...
    jobStatesStructure.PendingTasks,...
    jobStatesStructure.RunningTasks,...
    jobStatesStructure.FinishedTasks,...
    jobStatesStructure.TotalTasks,...
    allDataStructure.Tag,...
    allDataStructure.ApiTag,...
    jobStatesStructure.FinishedTasksNoError,...
    jobStatesStructure.ErroredTaskIDs,...
    jobStatesStructure.TaskIDsWithWarnings,...
    allDataStructure.FirstTaskFunction, ...
    corruptionInfo);
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function [firstTaskFunctionNames, corruptionInfos] = iGetFirstTaskFunctionNames(jobs, corruptionInfos)
firstTaskFunctionNames = repmat({''}, numel(jobs), 1);
try
    firstTaskFunctionNames = iGetFirstTaskFunctionNamesImpl(jobs);
catch E
    dctSchedulerMessage(1, 'Exception thrown while fetching first task function names', E)
    % If this fails, attempt to get properties one by one
    % to find out which one was actually failing
    for index = 1:numel(jobs)
        try
            firstTaskFunctionNames(index) = iGetFirstTaskFunctionNamesImpl(jobs(index));
        catch E
            corruptionInfos{index}{end+1} = parallel.internal.cluster.JobCorruptionInfo(...
                parallel.internal.cluster.types.JobCorruptionLevel.CannotProvideDescription, ...
                E.message);
        end
    end
end
end

function firstTaskFunctionNames = iGetFirstTaskFunctionNamesImpl(jobs)
    firstTaskFunctionNames = repmat({''}, numel(jobs), 1);
    jobTasks = iGetHetJobTasks(jobs);
    emptyIds = cellfun(@isempty, jobTasks);
    firstTasks = cellfun(@(tasks) tasks(1), jobTasks(~emptyIds));
    functionNames = get(firstTasks, 'FunctionName');
    if ~iscell(functionNames)
        functionNames = {functionNames};
    end
    firstTaskFunctionNames(~emptyIds) = functionNames;
end

function jobTasks = iGetHetJobTasks(jobs)
    if isempty(jobs)
        jobTasks = {};
        return
    end
    jobTasks = parallel.internal.cluster.hetfun(@hGetTasks, jobs, 'stable');
    jobTasks = [jobTasks{:}];
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function [jobTypes, corruptionInfos] = iGetJobTypes(jobs, corruptionInfos)
jobTypes = cell(numel(jobs), 1);
for idx = 1:numel(jobs)
    try
        jobTypes{idx} = iGetJobType(jobs(idx));
    catch E
        dctSchedulerMessage(1, 'Exception thrown while fetching java job types', E)
        corruptionInfos{idx}{end+1} = parallel.internal.cluster.JobCorruptionInfo(...
            parallel.internal.cluster.types.JobCorruptionLevel.CannotProvideType, ...
            E.message);
    end
end
end

function jobType = iGetJobType(job)
import parallel.internal.cluster.types.JobType
import parallel.internal.types.Variant

jobType = job.Type;
if isequal(jobType, Variant.IndependentJob.Name)
    jobType = JobType.IndependentJob;
elseif isequal(jobType, Variant.CommunicatingSPMDJob.Name)
    jobType = JobType.CommunicatingSPMDJob;
elseif isequal(jobType, Variant.CommunicatingPoolJob.Name)
    jobType = JobType.CommunicatingPoolJob;
elseif isequal(jobType, Variant.ConcurrentJob.Name)
    % NB: Should never get here as there is no documented non-parpool concurrent job.
    jobType = JobType.ConcurrentJob;
else
    error(message('parallel:ui:JobMonitorUnsupportedJobType', jobType));
end
end
