package com.fasterxml.sort.util;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.concurrent.BlockingQueue;

import com.fasterxml.sort.DataReader;

/**
 * Base implementation for {@link DataReader} that uses a
 * {@link BlockingQueue} for getting input.
 * The only missing part is implementation for
 * {@link #estimateSizeInBytes(Object)}, since there is no way
 * to provide a meaningful estimate without knowing object type.
 */
public abstract class BlockingQueueReader<E>
    extends DataReader<E>
{
    protected final BlockingQueue<E> _queue;
    
    public BlockingQueueReader(BlockingQueue<E> q) {
        _queue = q;
    }
    
    @Override
    public void close() throws IOException {
        // no-op
    }

    @Override
    public abstract int estimateSizeInBytes(E item);

    @Override
    public E readNext() throws IOException {
        try {
            return _queue.take();
        } catch (InterruptedException e) {
            InterruptedIOException ie = new InterruptedIOException();
            ie.initCause(e);
            throw ie;
        }
    }

}

@@ -594,74 +594,573 @@
    p
-ublic BlockingQueueReader(BlockingQueue%3CE%3E q) %7B%0A        _queue = q
+rotected final E _endMarker;%0A%0A    protected boolean _closed;%0A%0A    @Deprecated%0A    public BlockingQueueReader(BlockingQueue%3CE%3E q) %7B%0A        this(q, null);%0A    %7D%0A    %0A    /**%0A     * @param q Queue to read entries from%0A     * @param endMarker Value that is used to signal end-of-input; when this value%0A     *   is gotten from queue, reader assumes that no more input is coming and%0A     *   will return %3Ccode%3Enull%3C/code%3E from %7B@link #readNext%7D.%0A     */%0A    public BlockingQueueReader(BlockingQueue%3CE%3E q, E endMarker) %7B%0A        _queue = q;%0A        _endMarker = endMarker
 ;%0A  
@@ -1239,16 +1239,23 @@
     
-// no-op
+_closed = true;
 %0A   
@@ -1397,11 +1397,20 @@
     
-try
+if (_closed)
  %7B%0A 
@@ -1431,21 +1431,213 @@
 urn 
-_queue.take()
+null;%0A        %7D%0A        try %7B%0A            E value = _queue.take();%0A            if (value == _endMarker) %7B%0A                _closed = true;%0A                return null;%0A            %7D%0A            return value
 ;%0A  
@@ -1814,11 +1814,10 @@
 %7D%0A    %7D%0A
-%0A
 %7D%0A

Add support for end-marker token