@Test public void testUpdate() { FirstLevelStream stream = new FirstLevelStream(); stream.addView(myView); myView.initLock(); 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); myView.timerCallBack(BATCH_TIME); SupportViewDataCheck.checkOldData(childView, null); SupportViewDataCheck.checkNewData(childView, new IEvent[] {event1, 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); myView.timerCallBack(BATCH_TIME + BATCH_TIME); SupportViewDataCheck.checkOldData(childView, new IEvent[] {event1, event2}); SupportViewDataCheck.checkNewData(childView, new IEvent[] {event3, event4, event5}); }
@Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (newData.length == 0)) { LOG.error("Input Time Batch Window newData is Null!"); return; } try { lock(); for (IEvent event : newData) { curBatch.add(event); } } finally { if (isLocked()) { unlock(); } } }
TimeBatchWindow extends TimeBasedWindow implements IBatch { @Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (newData.length == 0)) { LOG.error("Input Time Batch Window newData is Null!"); return; } try { lock(); for (IEvent event : newData) { curBatch.add(event); } } finally { if (isLocked()) { unlock(); } } } }
TimeBatchWindow extends TimeBasedWindow implements IBatch { @Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (newData.length == 0)) { LOG.error("Input Time Batch Window newData is Null!"); return; } try { lock(); for (IEvent event : newData) { curBatch.add(event); } } finally { if (isLocked()) { unlock(); } } } TimeBatchWindow(long keepTime); }
TimeBatchWindow extends TimeBasedWindow implements IBatch { @Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (newData.length == 0)) { LOG.error("Input Time Batch Window newData is Null!"); return; } try { lock(); for (IEvent event : newData) { curBatch.add(event); } } finally { if (isLocked()) { unlock(); } } } TimeBatchWindow(long keepTime); @Override void update(IEvent[] newData, IEvent[] oldData); @Override void timerCallBack(long currentTime); @Override void sendBatchData(); }
TimeBatchWindow extends TimeBasedWindow implements IBatch { @Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (newData.length == 0)) { LOG.error("Input Time Batch Window newData is Null!"); return; } try { lock(); for (IEvent event : newData) { curBatch.add(event); } } finally { if (isLocked()) { unlock(); } } } TimeBatchWindow(long keepTime); @Override void update(IEvent[] newData, IEvent[] oldData); @Override void timerCallBack(long currentTime); @Override void sendBatchData();  }