%PoolOutputDispatcher
% Dispatcher that handles messages about streaming output. This is
% effectively an adaptor around the OutputHandler interface that allows
% MATLAB workers to stream chunks of output to the output handlers.
%

%   Copyright 2017 The MathWorks, Inc.

classdef PoolOutputDispatcher < handle
    properties (SetAccess = immutable)
        % A unique ID used to identify this dispatcher.
        Id;
    end
    
    properties (GetAccess = private, SetAccess = immutable)
        
        % The underlying output handlers on the client.
        OutputHandlers;
    end
    
    properties (Access = private)
        % The number of messages received by this dispatcher.
        NumReceivedMessages = 0;
        
        % The number of messages that workers report to have sent.
        NumSentMessages = 0;
    end
    
    properties (Constant)
        % A factory for generating unique ID for PoolOutputDispatcher
        % instances.
        IdFactory = matlab.bigdata.internal.util.UniqueIdFactory('PoolOutputDispatcher');
    end
    
    methods
        function obj = PoolOutputDispatcher(outputHandlers)
            % Construct a dispatcher that redirects to the given output
            % handlers.
            obj.Id = obj.IdFactory.nextId();
            obj.OutputHandlers = outputHandlers;
        end
        
        function streamWriterFactory = createStreamFactory(obj, taskId, dataQueueSession)
            % Create a factory for generating Writer objects that send data
            % back to this object. The factory can be serialized and sent to
            % workers.
            import parallel.internal.bigdata.PoolOutputWriter;
            id = obj.Id;
            streamWriterFactory = PoolOutputWriter.createFactory(taskId, @(varargin) dataQueueSession.send(id, varargin{:}));
        end
        
        function dispatch(obj, taskId, partitionIndex, numPartitions, isLastChunk, outputs, numSentMessages)
            % Dispatch a received message. This will contain one chunk of
            % data to be passed to the output handlers.
            obj.OutputHandlers.handleOutput(...
                taskId, partitionIndex, numPartitions, ...
                isLastChunk, outputs);
            
            obj.NumReceivedMessages = obj.NumReceivedMessages + 1;
            if isLastChunk
                obj.NumSentMessages = obj.NumSentMessages + numSentMessages;
            end
        end
        
        function reset(obj)
            % Reset all state for the next parallel operation.
            obj.NumReceivedMessages = 0;
            obj.NumSentMessages = 0;
        end
        
        function check(obj)
            % Check that this dispatcher has seen all messages it expects.
            assert(obj.NumReceivedMessages == obj.NumSentMessages, ...
                'Client MATLAB has not received all of the output of a gather.');
        end
    end
end
