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

%   Copyright 2014-2016 The MathWorks, Inc.

classdef (Sealed, Hidden) KeyValueSinkAdapter < parallel.internal.mapreduce.AbstractKeyValueSink
    
    properties (GetAccess = private, SetAccess = immutable)
        KeyValueSink; % The internal (java) key-value sink.
    end
    
    methods
        % Construct a KeyValueSink that wraps around the given java object.
        function this = KeyValueSinkAdapter(keyValueSink)
            this.KeyValueSink = keyValueSink;
        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)
            import java.util.concurrent.TimeUnit;
            if isnumeric(keys)
                keys = num2cell(keys);
            end
            
            sink = this.KeyValueSink;
            for ii = 1:numel(keys)
                % Due to memory and buffer limitations in Hadoop, each key
                % and value effectively is limited to 32-bit size.
                sink.write(distcompserialize(keys{ii}), distcompserialize(values(ii)));
                while ~sink.flush(1000, TimeUnit.MILLISECONDS)
                end
            end
        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)
            import java.util.concurrent.TimeUnit;
            sink = this.KeyValueSink;
            while ~sink.commit(1000, TimeUnit.MILLISECONDS)
            end
        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)
            this.KeyValueSink.close();
        end
    end
end
