classdef (Sealed) OrderedDataQueue < handle
% This object implments the same interface as a DataQueue, except that when
% you send data from the worker to the client you need to specify a
% sequence number (a double integer starting at 1). This queue object will
% use that sequence number to order the values delivered to the callbacks 
% such that they are in sequence order. It is a programming error to NOT
% deliver a value with a particular sequence number, or to return the same
% sequence number multiple times. In the case where a particular sequence
% number is never returned the queue will block forever awaiting that
% sequence number.

    properties (Access = private)
        % The DataQueue used to return data from the worker to the client
        Queue
    end
        
    properties (Transient, Access = private)
        % The index of the next item to be dispatched on the client side
        NextSeqNum = 1
        % Data structures to hold items in the buffer. A cell array for the
        % values and a logical array to indicate that we have received a
        % value for  that item in the buffer. The sequence number
        % associated with an item in the buffer starts at NextSeqNum (for
        % buffer position 1) and increments 
        Buffer = cell.empty
        BufferFilled = logical.empty
        % Listeners to be called when the results are available
        AfterEachListeners = event.listener.empty
    end
    
    events (ListenAccess = 'protected', NotifyAccess = 'protected')
        % This event is used to trigger all the listener callbacks supplied
        % to afterEach. 
        AfterEachTrigger
    end
    
    methods                
        function obj = OrderedDataQueue()
            obj.Queue = parallel.pool.DataQueue;
            % Define the callback to execute when ever we receive data from
            % the worker. This will either queue the data in the buffer
            % and/or execute the callback appropriately
            afterEach(obj.Queue, @(data) obj.receiveOne(data));
        end
        
        function send(obj, seqNum, value)
        % Return a single out of order value with sequence number seqNum
            validateattributes(seqNum, {'double'}, ...
                {'scalar', 'integer', 'positive'}, 'send');
            % Return data format is a cell array of sequence number and
            % value that will be unpacked in the receiveOne method
            obj.Queue.send({seqNum value});
        end
        
        function newListener = afterEach(obj, functionHandle)
            validateattributes(functionHandle, {'function_handle'}, ...
                {'scalar'}, 'afterEach', 'functionHandle');
            newListener = obj.createAfterEachListener(functionHandle);
        end
        
        % TODO - might consider adding a method here that indicates that
        % some particular sequence numbers are now never going to show up
        % (for example the code that was going to send them errored). Since
        % it might be difficult to deduce which sequence numbers have been
        % sent we should only ignore those numbers which have not (so far)
        % arrived. This will require us to consider what to do if something
        % shows up after it has been labelled as never arriving.
    end
    
    methods (Access = private) 
        function receiveOne(obj, data)
            % Unpack data from the worker - see send for the data packing 
            % strategy used.
            seqNum = data{1};
            value = data{2};
            % Assert that this sequence number is at least as large as the
            % next one to dispatch. It is a programming error to send the
            % same sequence number twice to the client.
            assert(seqNum >= obj.NextSeqNum);
            % Deduce if this is the next item to be passed to the callback
            isNextSeqNum = seqNum == obj.NextSeqNum;
            
            if isNextSeqNum && ~obj.hasBuffer
                % If this item is the next to be received and there is
                % nothing in the buffer - just dispatch to the callback
                obj.dispatchToListeners(seqNum, value)                
            else
                % Otherwise stick this into the buffer at the right place
                obj.insertInBuffer(seqNum, value);
                if isNextSeqNum
                    % If this one is the next to be received then we should
                    % dispatch this and any others that are ready to dispatch
                    % so loop until there isn't something to remove from
                    % the buffer.
                    while true
                        [OK, seqNum, value] = obj.pollFromBuffer;
                        if ~OK
                            break
                        end
                        obj.dispatchToListeners(seqNum, value);
                    end
                end
            end
        end

        function dispatchToListeners(obj, seqNum, value)
        % Dispatch the next item to all afterEach listeners
            % Check that this really is the correct one to send to the
            % listeners - would be bad if somehow we were dispatching the
            % wrong sequence number!
            assert(seqNum == obj.NextSeqNum);
            % Following the same pattern as the afterEach system in
            % DataQueue we create some DataReceivedEventData and trigger
            % the AfterEachTrigger. 
            evtData = parallel.pool.DataReceivedEventData(value);
            obj.notify('AfterEachTrigger', evtData);
            % Increment the next sequence number that we expect to dispatch
            % next
            obj.NextSeqNum = obj.NextSeqNum + 1;
        end
        
        function insertInBuffer(obj, seqNum, value)
        % Rather than dispatching an item lets put it into the current
        % buffer - the buffer starts at NextSeqNum so we need to compute
        % the bufferIndex based on this items sequence number
            bufferIndex = seqNum - (obj.NextSeqNum - 1);
            % Check that this bufferIndex has NOT been used before - this
            % checks that the program is NOT sending the same value twice
            assert(bufferIndex > numel(obj.BufferFilled) || ...
                ~obj.BufferFilled(bufferIndex));
            % Put the value in the buffer and indicate that that element is
            % now filled
            obj.Buffer{bufferIndex} = value;
            obj.BufferFilled(bufferIndex) = true;
        end
        
        function [OK, seqNum, value] = pollFromBuffer(obj)
        % Remove first item from the buffer if it exists. OK will be true
        % if the item was removed, false if there was no item to return. If
        % OK is true then seqNum and value will be removed from the buffer
            OK = ~isempty(obj.BufferFilled) && obj.BufferFilled(1);
            if OK 
                value = obj.Buffer{1};
                seqNum = obj.NextSeqNum;
                % Remove the first item from the buffer
                obj.Buffer(1) = [];
                obj.BufferFilled(1) = [];
            else
                % Default returns if no item is available at the front of
                % the queue
                seqNum = 0;
                value = [];
            end
        end
        
        function OK = hasBuffer(obj)
            OK = ~isempty(obj.BufferFilled);
        end
        
        function listener = createAfterEachListener(obj, func)
        % Follow the same pattern for afterEach as used for DataQueue using
        % listeners to trigger all the multiple callbacks. 
            listener = event.listener(obj, 'AfterEachTrigger', ...
                @(src, event) iDispatchDataReceived(func, src, event));
            listener.Recursive = true;
            obj.removeInvalidAfterEachListeners();
            obj.AfterEachListeners(end+1) = listener;
        end
        
        function removeInvalidAfterEachListeners(obj)
            obj.AfterEachListeners(~isvalid(obj.AfterEachListeners)) = [];
        end                
    end
end

function iDispatchDataReceived(func, ~, evtData)
try
    feval(func, evtData.ReceivedData);
catch err
    parallel.internal.warningNoBackTrace(err.identifier, err.message);
    dctSchedulerMessage(2, ...
        'Error caught while executing afterEach continuation: %s', err.message);
end
end