%AsyncCommandRunner
% A helper class that executes a command in asynchronous nature.

%   Copyright 2014-2016 The MathWorks, Inc.

classdef (Sealed, Hidden) AsyncCommandRunner < handle
    
    properties (Access = private)
        ProcessWrapper; % An instance of the process
        
        OutputLogFile; % Path to the output log file.
        OutputLogFilePosition = 0; % A position in the output log file that has already been displayed.
        OutputLogTempFolder; % A temporary folder to hold the output log if no path has been given.
        
        InputStream; % A stream to send data to the input pipe of the child process.
    end
    
    methods
        % Construct a HadoopCommandRunner from the given installation path
        % and version of Hadoop.
        function obj = AsyncCommandRunner(varargin)
            import com.mathworks.toolbox.parallel.mapreduce.ProcessWrapper;
            import java.io.File;
            import java.lang.Process;
            import java.lang.ProcessBuilder;
            import parallel.internal.mapreduce.TempLocalFolder;
            
            p = inputParser;
            p.addParameter('Command', {});
            p.addParameter('Environment', containers.Map());
            p.addParameter('OutputLogFile', []);
            p.addParameter('CurrentWorkingFolder', []);
            p.parse(varargin{:});
            
            results = p.Results;
            
            pb = ProcessBuilder(results.Command);
            pb.redirectErrorStream(true);
            
            % Output log file
            if isempty(results.OutputLogFile)
                obj.OutputLogTempFolder = TempLocalFolder();
                results.OutputLogFile = fullfile(obj.OutputLogTempFolder.Path, 'localhadoop.log');
            end
            obj.OutputLogFile = results.OutputLogFile;
            pb.redirectOutput(File(obj.OutputLogFile));
            
            % Working directory
            if isempty(results.CurrentWorkingFolder)
                results.CurrentWorkingFolder = pwd;
            end
            pb.directory(File(results.CurrentWorkingFolder));
            
            % Environment
            pbEnvMap = pb.environment();
            iAddMatlabSetEnvironmentVariables(pbEnvMap);
            names = results.Environment.keys;
            if ~isempty(names)
                values = results.Environment.values;
                for ii = 1:numel(names)
                    if isempty(values{ii})
                        pbEnvMap.put(names{ii}, '');
                    else
                        pbEnvMap.put(names{ii}, values{ii});
                    end
                end
            end
            
            process = pb.start();
            obj.ProcessWrapper = ProcessWrapper(process);
            obj.InputStream = process.getOutputStream();
        end
        
        % Wait the given number of milliseconds and return if the process
        % is still running.
        function isStillRunning = wait(obj, numMilliseconds)
            import java.util.concurrent.TimeUnit;
            if ~isempty(obj.ProcessWrapper)
                isStillRunning = obj.ProcessWrapper.waitFor(numMilliseconds, TimeUnit.valueOf('MILLISECONDS'));
            end
        end
        
        % Get the exit value of a process that has terminated without
        % cancel.
        function exitValue = getExitValue(obj)
            exitValue = obj.ProcessWrapper.getExitValue();
        end
        
        % Terminate the process by a SIGINT
        function cancel(obj)
            assert(~isempty(obj.ProcessWrapper));
            obj.ProcessWrapper.cancel();
        end
        
        % Get the full output from the output log.
        function data = getAllOutput(obj)
            import parallel.internal.hadoop.FileSystemTools;
            [fid, fidCleanup] = FileSystemTools.openLocalTextFile(obj.OutputLogFile, 'r'); %#ok<ASGLU>
            
            data = fread(fid, '*char')';
        end
        
        % Get any output that has not been returned as yet.
        function data = getNextOutput(obj)
            import parallel.internal.hadoop.FileSystemTools;
            [fid, fidCleanup] = FileSystemTools.openLocalTextFile(obj.OutputLogFile, 'r'); %#ok<ASGLU>
            
            fseek(fid, obj.OutputLogFilePosition, 'bof');
            data = fread(fid, '*char')';
            obj.OutputLogFilePosition = ftell(fid);
        end
        
        % Send binary data to the input pipe of the process.
        function sendData(obj, data)
            assert(isa(data, 'int8'));
            obj.InputStream.write(data);
            obj.InputStream.flush();
        end
        
        % Attempt to prevent ctrl-c drooling the process.
        function delete(obj)
            obj.cancel();
        end
    end
    
end

% Add any environment set by MATLAB that is not visible to Java.
function iAddMatlabSetEnvironmentVariables(pbEnvMap)
import parallel.internal.apishared.EnvVars;
[names, values] = EnvVars.createLocalClusterEnvironment({});
for ii = 1:numel(names)
    pbEnvMap.put(names{ii}, values{ii});
end
end
