@Override
    public Object take(int forThreadId) throws InterruptedException
    {
        if (isInTerminalPhase(forThreadId))
        {
            return terminalPhaseTake(forThreadId);
        }

        while (routerQueueHasData(forThreadId))
        {
            Object nameValues = takeFromRouter(forThreadId);

            if (nameValues == terminalNameValues)
            {
                deadRouterProducers[forThreadId]++;
            }
            else
            {
                return nameValues;
            }
        }

        String[] line = Import.measureTime(queueTakes, new Callable<String[]>()
        {
            @Override
            public String[] call() throws InterruptedException
            {
                return queue.take();
            }
        });

        if (!isTerminalLine(line))
        {
            return line;
        }

        // Let other consumers know that producer has finished reading lines,
        // and there won't be new records in the shared queue
        producerDone();

        if (isRouterEnabled())
        {
            deadQueueConsumers[forThreadId]++;
            deadRouterProducers[forThreadId]++;

            // Notify other consumers that this thread has done processing shared queue,
            // and is waiting for confirmation from other consumers that they don't have any
            // messages for it
            for (int i = 0; i < routerQueues.length; i++)
            {
                if (i != forThreadId)
                {
                    dispatch(terminalNameValues, i);
                }
            }

            return terminalPhaseTake(forThreadId);
        }

        // Empty array is a terminal line
        return new String[0];
    }