package com.microsoft.wake.test.examples;

import java.util.Random;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;


import org.junit.Test;

import com.microsoft.wake.examples.accumulate.CombinerStage;
import com.microsoft.wake.examples.accumulate.CombinerStage.Combiner;
import com.microsoft.wake.rx.Observer;

public class TestCombiner {
  
  int bucketCount = 1000;
  int bucketSleepMillis = 100;
  int threadCount = 1;
  int eventsCount = 2*1000000;
  int eventsPerThread = eventsCount / threadCount;
  
  Observer<Entry<Integer, Integer>> o;

  volatile boolean done = false;
  
  private class WorkerThread extends Thread {
    @Override
    public void run() {
      Random rand = new Random();
      for (int i = 0; i < eventsPerThread; i++) {
        int r = rand.nextInt(bucketCount);
        o.onNext(new CombinerStage.Pair<Integer, Integer>(r, 1));
      }

    }
  };

  @Test
  public void test() throws Exception {
    CombinerStage<Integer, Integer> stage = new CombinerStage<Integer, Integer>(
        new Combiner<Integer, Integer>() {

          @Override
          public Integer combine(Integer key, Integer old, Integer cur) {
            if (old != null) {
              return old.intValue() + cur.intValue();
            } else {
              return cur;
            }
          }

        }, new Observer<Entry<Integer, Integer>>() {
          private AtomicInteger x = new AtomicInteger(0);
          
          @Override
          public void onNext(Entry<Integer, Integer> value) {
            System.out.println(value.getKey() + "=" + value.getValue());
            x.incrementAndGet();
//            try {
//              if(!done) 
//                Thread.sleep(100);
//            } catch (InterruptedException e) {
//              throw new IllegalStateException(e);
//            }
          }

          @Override
          public void onError(Exception error) {
            System.err.println("onError called!");
            error.printStackTrace();
          }

          @Override
          public void onCompleted() {
            System.out.println("onCompleted " + x);
          }

        });

    o = stage.wireIn();

    WorkerThread[] workers = new WorkerThread[threadCount];

    for (int i = 0; i < threadCount; i++) {
      workers[i] = new WorkerThread();
    }
    long start = System.currentTimeMillis();
    for (int i = 0; i < threadCount; i++) {
      workers[i].start();
    }
    for (int i = 0; i < threadCount; i++) {
      workers[i].join();
    }
    o.onCompleted();

    long inStop = System.currentTimeMillis();
    done = true;

    stage.close();
    long outStop = System.currentTimeMillis();

    long eventCount = threadCount * eventsPerThread;
    double inelapsed = ((double)(inStop-start))/1000.0;
    double inoutelapsed = ((double)(outStop-start))/1000.0;
    System.out.println("Emitted " + eventCount + " events in " + inelapsed
        + " seconds (" + ((double) eventCount) / (inelapsed*1000.0*1000.0) + " million events/sec)");
    System.out.println("Emitted/output " + eventCount + " events in " + inoutelapsed
        + " seconds (" + ((double) eventCount) / (inoutelapsed*1000.0*1000.0) + " million events/sec)");

  }
}

@@ -1642,18 +1642,16 @@
 dGet();%0A
-//
         
@@ -1660,18 +1660,16 @@
   try %7B%0A
-//
         
@@ -1685,18 +1685,16 @@
 !done) %0A
-//
         
@@ -1720,14 +1720,11 @@
 p(10
-0
 );%0A
-//
     
@@ -1766,18 +1766,16 @@
 on e) %7B%0A
-//
         
@@ -1820,10 +1820,8 @@
 e);%0A
-//
     

Add a thread sleep