%BufferedKeyValueSink
% Class that buffers input in order to vectorize operations that happen
% further in the sink chain.

%   Copyright 2014-2016 The MathWorks, Inc.

classdef (Sealed, Hidden) BufferedKeyValueSink < parallel.internal.mapreduce.AbstractKeyValueSink
    properties (GetAccess = private, SetAccess = immutable)
        Sink; % The actual sink that will receive vectorized adds.
        Buffer; % A buffer that collects data to be vectorized.
    end
    
    methods
        % Construct a BufferedKeyValueSink that passes data to the given
        % sink.
        function obj = BufferedKeyValueSink(sink)
            import parallel.internal.mapreduce.KeyValueBuffer;
            
            assert(isa(sink, 'parallel.internal.mapreduce.AbstractKeyValueSink'));
            obj.Sink = sink;
            
            obj.Buffer = KeyValueBuffer();
        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(obj, keys, values)
            if isempty(keys)
                return;
            end
            
            bufferNeedsFlush = obj.Buffer.add(keys, values);
            if bufferNeedsFlush
                obj.flush();
            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(obj)
            if obj.Buffer.NumAdds > 0
                obj.flush();
            end
            obj.Sink.commit();
        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 result = getTaskOutputMetadata(obj)
            result = obj.Sink.getTaskOutputMetadata();
        end
        
        % Create a key/value processor that ensures keys and values have the
        % correct type and size for this given sink.
        function processor = createKeyValueProcessor(obj)
            processor = obj.Sink.createKeyValueProcessor();
        end
    end
    
    methods (Access = private)
        % Flush all buffered data.
        function flush(obj)
            [keys, values] = obj.Buffer.get();
            obj.Buffer.reset();
            
            obj.Sink.add(keys, values);
        end
    end
end
