package technique.driver;

public class TestDriver {

}

@@ -24,33 +24,2622 @@
 r;%0A%0A
-public class TestDriver %7B%0A
+import java.util.ArrayList;%0Aimport java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map;%0A%0Aimport org.junit.Assert;%0Aimport org.junit.Test;%0A%0Aimport technique.feed.FeedPartition;%0Aimport technique.feed.FixedFeed;%0Aimport technique.model.Operator;%0Aimport technique.model.Tuple;%0Aimport technique.service.CollectorProcessor;%0A%0Apublic class TestDriver %7B%0A%0A  private FeedPartition getPart()%7B%0A    Map%3CString,Object%3E prop = new HashMap%3CString,Object%3E();%0A    int expectedPartitions = 5;%0A    int expectedRows = 1000;%0A    prop.put(FixedFeed.NUMBER_OF_PARTITIONS, expectedPartitions);%0A    prop.put(FixedFeed.NUMBER_OF_ROWS, expectedRows);%0A    FixedFeed pf = new FixedFeed(prop);%0A    List%3CFeedPartition%3E parts = pf.getFeedPartitions();%0A    return parts.get(0);%0A  %7D%0A  %0A  @Test%0A  public void aTest() throws InterruptedException %7B%0A    Driver root = new Driver(getPart(), minus1Operator());%0A    root.initialize();%0A    DriverNode child = new DriverNode(times2Operator(), new CollectorProcessor());%0A    root.getDriverNode().addChild(child);%0A    %0A    Thread t = new Thread(root);%0A    t.run();%0A    t.join();%0A%0A    List%3CTuple%3E expected = new ArrayList%3CTuple%3E();%0A    for (int i = 0; i %3C 9; i++) %7B%0A      Tuple tup = new Tuple();%0A      tup.setField(%22x%22, (i - 1) * 2);%0A      expected.add(tup);%0A    %7D%0A    assertExpectedPairs(child, expected);%0A    %0A  %7D%0A%0A  public void assertExpectedPairs(DriverNode finalNode, List%3CTuple%3E expected) throws InterruptedException %7B%0A    for (int i = 0; i %3C expected.size(); i++) %7B%0A      Assert.assertNotNull(finalNode.getCollectorProcessor().getCollector().peek());%0A      Tuple got = finalNode.getCollectorProcessor().getCollector().take();%0A      Assert.assertTrue(%22element %22+i+%22 comparing %22 + expected.get(i) + %22 %22 + got, expected.get(i).equals(got));%0A    %7D%0A    Assert.assertNull(finalNode.getCollectorProcessor().getCollector().peek());%0A  %7D%0A  %0A  @Test%0A  public void compareTuple()%7B%0A    Tuple t = new Tuple();%0A    t.setField(%22x%22, -2);%0A    Tuple s = new Tuple();%0A    s.setField(%22x%22, -2);%0A    %0A    Assert.assertTrue(t.equals(s));%0A  %7D%0A  %0A  %0A  public static Operator times2Operator()%7B%0A    return new Operator()%7B%0A      @Override%0A      public void handleTuple(Tuple t) %7B%0A        Tuple tnew = new Tuple();%0A        tnew.setField(%22x%22, ((Integer) t.getField(%22x%22)).intValue() * 2);%0A        collector.emit(tnew);%0A      %7D%0A    %7D;%0A%0A  %7D%0A  %0A  public static Operator minus1Operator()%7B%0A    return new Operator()%7B%0A      @Override%0A      public void handleTuple(Tuple t) %7B%0A        Tuple tnew = new Tuple();%0A        tnew.setField(%22x%22, ((Integer) t.getField(%22x%22)).intValue()-1 );%0A        collector.emit(tnew);%0A      %7D%0A    %7D;%0A  %7D%0A  %0A  
 %0A%7D%0A

Test driver