%TextKeyValueSink
% Helper class that wraps a java sink class as something that can be used
% by MapTask/ReduceTask.

%   Copyright 2014-2018 The MathWorks, Inc.

classdef (Sealed, Hidden) TextKeyValueSink < parallel.internal.mapreduce.AbstractKeyValueSink
    
    properties (GetAccess = private, SetAccess = immutable)
        FileName; % The filename of the output file.
    end
    
    properties (Access = private)
        FileWriter; % The internal writer object.
    end
    
    methods
        % Construct a Key-Value writer that wraps around the given file writer.
        function this = TextKeyValueSink(filename, ~)
            this.FileName = filename;
        end
        
        % Add the given <key, value> pairs to the underlying data storage.
        % The keys and values input must be vectors of the same length.
        function add(this, keys, values)
            assert(numel(keys) == numel(values));
            if isnumeric(keys)
                keyFormat = '%d';
                keys = num2cell(keys);
            elseif iscellstr(keys)
                keyFormat = '%s';
            else
                assert(false);
            end
            if isnumeric(values) || islogical(values)
                valueFormat = '%d';
                values = num2cell(values);
            elseif iscellstr(values)
                valueFormat = '%s';
            else
                assert(false);
            end
            
            data = [keys, values]';
            output = sprintf([keyFormat, '\t', valueFormat '\n'], data{:});
            
            jStr = java.lang.String(output);
            if isempty(this.FileWriter)
                this.createFileWriter();
            end
            this.FileWriter.write(jStr.getBytes('UTF-8'));
        end
        
        % Commit any output to the final location. This is only called if
        % the task is successful. If a task is not successful, the sink
        % is expected to cleanup all side-effects when destroyed.
        function commit(this)
            if isempty(this.FileWriter)
                this.createFileWriter();
            end
            this.FileWriter.close();
        end
        
        % Provide any data that is to be returned to the driver code for
        % MapReduce. This can be an arbitrary serializable value and will
        % be called once at the end of a Map or Reduce task to obtain the
        % return value of the run method.
        function results = getTaskOutputMetadata(~)
            results = [];
        end
        
        % Close the underlying java object.
        function delete(this)
            if ~isempty(this.FileWriter)
                this.FileWriter.close();
            end
        end
        
        % Create a key/value processor that ensures keys and values have the
        % correct type and size for this given sink.
        function processor = createKeyValueProcessor(~)
            import matlab.mapreduce.internal.TextKeyValueProcessor;
            processor = TextKeyValueProcessor();
        end
    end
    
    methods (Access = private)
        % Create the internal writer object
        function createFileWriter(this)
            import matlab.io.datastore.internal.ContextClassLoaderGuard;
            import matlab.io.datastore.internal.HadoopConfiguration;
            guard = ContextClassLoaderGuard(); %#ok<NASGU>
            
            hdfsLoader = com.mathworks.storage.hdfsloader.GlobalHdfsLoader.get();
            path = hdfsLoader.newPath(this.FileName);
            fileSystem = HadoopConfiguration.getGlobalFileSystem(path.toUri());
            for retry = 1:3
                try
                    this.FileWriter = fileSystem.create(path);
                catch err
                    continue;
                end
                return;
            end
            % Failed after retry attempts.
            newErr = MException(message('parallel:mapreduce:ParallelOutputError'));
            newErr = addCause(newErr, err);
            throw(newErr);
        end
    end
end
