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

%   Copyright 2014-2016 The MathWorks, Inc.

classdef (Sealed, Hidden) KeyValuesSourceAdapter < parallel.internal.mapreduce.AbstractKeyValuesSource
    
    properties (GetAccess = private, SetAccess = immutable)
        KeyValuesSource; % The internal (java) key-values source.
    end
    
    
    methods
        % Construct an KeyValuesSource that reads from the given java object.
        function this = KeyValuesSourceAdapter(source)
            this.KeyValuesSource = source;
            
            % Have to cache first key to ensure hasData returns a correct
            % result.
            this.nextKeyOrValue();
        end
        
        % Returns true if and only if there is more data overall to read.
        function result = hasData(this)
            result = ~this.KeyValuesSource.isFinished();
        end
        
        % Returns true if and only if there is more data associated with the
        % current key.
        function result = hasDataForCurrentKey(this)
            result = this.KeyValuesSource.isLastMessageAValue();
        end
        
        % Indicate that this source should start reading values from the
        % next key. This cannot be called when there is still data for the
        % current key.
        function key = nextKey(this)
            import java.util.concurrent.TimeUnit;
            assert(~this.hasDataForCurrentKey());
            key = this.KeyValuesSource.getCurrentKeyAsByteArray();
            source = this.KeyValuesSource;
            while ~source.isFinished()
                if source.nextKeyOrValue(1000, TimeUnit.MILLISECONDS)
                    break;
                end
            end
            if ~isempty(key)
                key = distcompdeserialize(key);
            end
        end
        
        % Discard any remaining data associated with the current key.
        function discardRemainingDataForCurrentKey(this)
            source = this.KeyValuesSource;
            while source.isLastMessageAValue()
                this.nextKeyOrValue();
            end
        end
        
        % Return a chunk of values up to maxNumEntries number of entries
        % associated with the current key.
        function values = read(this, maxNumEntries)
            source = this.KeyValuesSource;
            values = cell(maxNumEntries, 1);
            for ii = 1:numel(values)
                if ~source.isLastMessageAValue()
                    values(ii:end) = [];
                    break;
                end
                values{ii} = source.getCurrentValueAsByteArray();
                this.nextKeyOrValue();
            end
            [values{:}] = distcompdeserialize(values{:});
            values = vertcat(values{:});
        end
        
        % Close the underlying java object.
        function delete(this)
            this.KeyValuesSource.close();
        end
    end
    
    methods (Access = private)
        % Move to the next key or value, blocking until it does. We do
        % timed wait so that we can escape on cancel/ctrl-c
        function nextKeyOrValue(this)
            import java.util.concurrent.TimeUnit;
            source = this.KeyValuesSource;
            while ~source.isFinished()
                if source.nextKeyOrValue(1000, TimeUnit.MILLISECONDS)
                    break;
                end
            end
        end
    end
end
