%HadoopCommandRunner
% A helper class that executes a call to 'hadoop jar <our job jar file>'
% with the given arguments, start-in path and environment.

%   Copyright 2014-2018 The MathWorks, Inc.

classdef (Sealed, Hidden) HadoopCommandRunner < handle
    
    properties (SetAccess = immutable)
        HadoopInstallFolder; % Installation path of Hadoop.
        Version; % The major Version of Hadoop.
    end
    
    properties
        Display = false; % Flag to display if we let Hadoop logging reach the command line.
        DisplayFcn = @(x)fprintf('%s', x); % The function to call at each poll interval to display output.
        StartInPath; % The path to call Hadoop from.
        OutputLogFile; % A path to a log file for storing the output from Hadoop.
    end
    
    properties (SetAccess=private)
        Environment; % Environment variables that are set during execution.
    end
    
    properties (Constant, Access = private)
        DISPLAY_POLL_TIME = 1000; % Poll time to call display callback in milliseconds
    end
    
    methods
        % Construct a HadoopCommandRunner from the given installation path
        % and version of Hadoop.
        function obj = HadoopCommandRunner(hadoopInstallFolder, version)
            obj.HadoopInstallFolder = hadoopInstallFolder;
            obj.Version = version;
            obj.Environment = containers.Map('KeyType', 'char', 'ValueType', 'any');
            
            % Hadoop requires JAVA_HOME to be set to a valid location.
            if isempty(getenv('JAVA_HOME'))
                obj.Environment('JAVA_HOME') = fullfile(matlabroot, 'sys', 'java', 'jre', computer('arch'), 'jre');
            else
                obj.Environment('JAVA_HOME') = getenv('JAVA_HOME');
            end
            
            % We will get warnings from Hadoop 1.x if we do not ensure
            % HADOOP_HOME is empty.
            obj.Environment('HADOOP_HOME') = '';
        end
        
        % Execute the tool of the given name with the given input
        % arguments.
        function [status, message] = runTool(obj, name, varargin)
            % We use matlabroot as opposed to toolboxdir here because in a
            % deployed environment, the hadoopjob jar will be shipped as
            % part of the MCR and not as part of the deployed application.
            jobFile = fullfile(matlabroot, 'toolbox', 'parallel', 'bigdata', 'jar', sprintf('hadoopjob.%i.jar', obj.Version));
            toolName = sprintf('com.mathworks.toolbox.parallel.hadoop.%s', name);
            
            args = [...
                {'jar'}, ...
                {jobFile}, ...
                {toolName}, ...
                varargin(:)'];
            
            if nargout
                [status, message] = run(obj, args{:});
            else
                run(obj, args{:});
            end
        end
        
        % Execute a general command on hadoop
        function [status, msg] = run(obj, varargin)
            import parallel.internal.hadoop.AsyncCommandRunner;

            if ispc
                hadoopBinFile = 'hadoop.cmd';
            else
                hadoopBinFile = 'hadoop';
            end
            hadoopBinPath = fullfile(obj.HadoopInstallFolder, 'bin', hadoopBinFile);
            asyncRunner = AsyncCommandRunner(...
                'Command', [{hadoopBinPath}, varargin], ...
                'CurrentWorkingFolder', obj.StartInPath, ...
                'Environment', obj.Environment, ...
                'OutputLogFile', obj.OutputLogFile);
                

            shouldContinue = true;
            while shouldContinue
                shouldContinue = ~asyncRunner.wait(obj.DISPLAY_POLL_TIME);

                if obj.Display
                    obj.DisplayFcn(asyncRunner.getNextOutput());
                end
            end
            
            status = asyncRunner.getExitValue();
            msg = asyncRunner.getAllOutput();
            
            if ~nargout && status ~= 0
                error(message('parallel:mapreduce:HadoopJobFailure'));
            end
        end
    end
    
end
