@Override
    public void nextTuple() {
//        Utils.sleep(10);
        String message = null;
        int i = 0;
        while ((message = queue.poll()) != null) {
            if (++i % 2000 == 0)
                Utils.sleep(10);
            String[] columns = message.split(",");
            Values tuple = new Values(columns[0], columns[1], Long.parseLong(columns[2]), columns[4], columns[5]);
            if (logger.isDebugEnabled()) {
                logger.debug(format("[%s]:%s", SIGNALLING, tuple.toString()));
            }
            spoutOutputCollector.emit(SIGNALLING, tuple);
            if (logger.isDebugEnabled()) {
                logger.info(String.format("spout sent:%s,%s,%s", tuple.get(0), tuple.get(1), tuple.get(2)));
            }
        }
    }