%DataQueueKeyValueSink
% Helper class that wraps a DataQueue as something that can be used
% by MapTask/ReduceTask.

% Copyright 2015-2020 The MathWorks, Inc.

classdef (Sealed, Hidden) DataQueueKeyValueSink < parallel.internal.mapreduce.AbstractKeyValueSink

properties (GetAccess = private, SetAccess = immutable)
DataQueue; % The internal DataQueue
Processor; % The processor instance that is used to clean user input
% intended to be added to the output. This is not used
% directly by this class, but passed onto the instance
% of OutputKeyValueStore given to the reduce function.
end

properties (Access = private)
KeyType = []; % The key type that has been written to this sink.
ValueType = []; % The value type that has been written to this sink.
end

methods
% Construct a KeyValueSink that wraps around the given DataQueue object.
function obj = DataQueueKeyValueSink(dataQueue, processor)
obj.DataQueue = dataQueue;
if nargin >= 2
obj.Processor = processor;
end
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)
import parallel.internal.mapreduce.ByteBufferHandleWrapper;

if isempty(obj.KeyType) && ~isempty(keys)
exampleKey = keys(1);
if iscellstr(exampleKey) %#ok<ISCLSTR>
exampleKey = exampleKey{1};
end

% We want to store the key/value class type for future error
% checking. We are guaranteed that all data is the same type by
% the checking that has happened in KeyValueOutputStore. This
% is for comparing the data of two different workers.
obj.KeyType = class(exampleKey);
obj.ValueType = class(values);
end

obj.DataQueue.send({keys, values});
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(~)
% We're reliant on the fact communication in PCT is ordered and
% so the destination will only receive the parfeval complete
% message after all data messages.
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(obj)
import parallel.internal.mapreduce.ReduceTaskResult;
results = ReduceTaskResult({}, obj.KeyType, obj.ValueType);
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)
import parallel.internal.mapreduce.AbstractKeyValueSink
if isempty(obj.Processor)
processor = createKeyValueProcessor@AbstractKeyValueSink(obj);
else
processor = obj.Processor;
end
end
end
end