LengthBatchWindow extends LengthBasedWindow implements IBatch { @Override public void update(IEvent[] newData, IEvent[] oldData) { if (null == newData || 0 == newData.length) { LOG.error("Input Length Batch Window newData is Null!"); return; } for (IEvent newEvent : newData) { curBatch.add(newEvent); } if (curBatch.size() < getKeepLength()) { return; } sendBatchData(); } LengthBatchWindow(int keepLength); @Override void update(IEvent[] newData, IEvent[] oldData); void sendBatchData(); @Override IView renewView();  }
@Test public void testUpdate() { FirstLevelStream stream = new FirstLevelStream(); stream.addView(myView); Map<String, Object> values = new HashMap<String, Object>(); values.put("a", 1); values.put("b", 1); values.put("c", "c1"); IEvent event1 = new TupleEvent("stream", eventType, values); stream.add(event1); SupportViewDataCheck.checkOldData(childView, null); SupportViewDataCheck.checkNewData(childView, null); values.put("a", SupportConst.I_TWO); values.put("b", SupportConst.I_TWO); values.put("c", "c2"); IEvent event2 = new TupleEvent("stream", eventType, values); stream.add(event2); values.put("a", SupportConst.I_THREE); values.put("b", SupportConst.I_THREE); values.put("c", "c3"); IEvent event3 = new TupleEvent("stream", eventType, values); stream.add(event3); values.put("a", SupportConst.I_FOUR); values.put("b", SupportConst.I_FOUR); values.put("c", "c4"); IEvent event4 = new TupleEvent("stream", eventType, values); stream.add(event4); values.put("a", SupportConst.I_FIVE); values.put("b", SupportConst.I_FIVE); values.put("c", "c5"); IEvent event5 = new TupleEvent("stream", eventType, values); stream.add(event5); SupportViewDataCheck.checkOldData(childView, null); SupportViewDataCheck.checkNewData(childView, new IEvent[] {event1, event2, event3, event4, event5}); values.put("a", SupportConst.I_SIX); values.put("b", SupportConst.I_SIX); values.put("c", "c6"); IEvent event6 = new TupleEvent("stream", eventType, values); stream.add(event6); values.put("a", SupportConst.I_SEVEN); values.put("b", SupportConst.I_SEVEN); values.put("c", "c7"); IEvent event7 = new TupleEvent("stream", eventType, values); stream.add(event7); values.put("a", SupportConst.I_EIGHT); values.put("b", SupportConst.I_EIGHT); values.put("c", "c8"); IEvent event8 = new TupleEvent("stream", eventType, values); stream.add(event8); values.put("a", SupportConst.I_NINE); values.put("b", SupportConst.I_NINE); values.put("c", "c9"); IEvent event9 = new TupleEvent("stream", eventType, values); stream.add(event9); values.put("a", SupportConst.I_TEN); values.put("b", SupportConst.I_TEN); values.put("c", "c10"); IEvent event10 = new TupleEvent("stream", eventType, values); stream.add(event10); SupportViewDataCheck.checkOldData(childView, new IEvent[] {event1, event2, event3, event4, event5}); SupportViewDataCheck.checkNewData(childView, new IEvent[] {event6, event7, event8, event9, event10}); }