source: class RangeTest {
  
    @SuppressWarnings("unchecked")
    @Test
    public void testNodeSchemaProcessing()
    {
  
target:     Range<Double> oper = new Range<Double>();
      TestSink rangeSink = new TestSink();
      oper.range.setSink(rangeSink);
  
      oper.beginWindow(0); //
  
      int numTuples = 1000;
      for (int i = 0; i < numTuples; i++) {
        Double a = new Double(20.0);
        Double b = new Double(2.0);
        Double c = new Double(1000.0);
  
        oper.data.process(a);
        oper.data.process(b);
        oper.data.process(c);
  
        a = 1.0;
        oper.data.process(a);
        a = 10.0;
        oper.data.process(a);
        b = 5.0;
        oper.data.process(b);
  
        b = 12.0;
        oper.data.process(b);
        c = 22.0;
        oper.data.process(c);
        c = 14.0;
        oper.data.process(c);
  
        a = 46.0;
        oper.data.process(a);
        b = 2.0;
        oper.data.process(b);
        a = 23.0;
        oper.data.process(a);
      }
  
      oper.endWindow(); //
  
      // payload should be 1 bag of tuples with keys "a", "b", "c", "d", "e"
      Assert.assertEquals("number emitted tuples", 1, rangeSink.collectedTuples.size());
      for (Object o : rangeSink.collectedTuples) {
        HighLow<V> hl = (HighLow<V>)o;
        Assert.assertEquals("emitted high value was ", 1000.0,
            hl.getHigh());
        Assert.assertEquals("emitted low value was ", 1.0,
            hl.getLow());
      }
    }
  
  }