benchmark.java

    public void consume_iterator() {
        FastqQueueReader<N> reader = new FastqQueueReader(broker);
        int i = 0;
        long prev = System.currentTimeMillis();
        double accum = 0;
        while(broker.isRunning() || reader.hasNext()) {
            N data = reader.read();
            accum += (((byte[])data).length*1.0d/(1024.0*1024.0));
            if (i % 40000 == 0) {
                long now_st = System.currentTimeMillis();
                double dur = (now_st - prev)*1.0/1000.0;
                // System.out.println("accur: "+accum+" ,dur: "+dur);
                double rate =  (accum / dur);
                System.err.println("Progress > Next .. " + i + ", rate: "+rate +" ("+accum+"/"+dur+")");
                prev = System.currentTimeMillis();
                accum = 0;
            }
            i++;
//            for(byte b : (byte[])data) {
//                System.out.print((char)b);
//            }
        }

    }