@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, new IEvent[] {event1}); 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); SupportViewDataCheck.checkOldData(childView, null); SupportViewDataCheck.checkNewData(childView, new IEvent[] {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); SupportViewDataCheck.checkOldData(childView, null); SupportViewDataCheck.checkNewData(childView, new IEvent[] {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); SupportViewDataCheck.checkOldData(childView, null); SupportViewDataCheck.checkNewData(childView, new IEvent[] {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[] {event5}); myView.timerCallBack(System.currentTimeMillis() + OFFSET); SupportViewDataCheck.checkOldData(childView, new IEvent[] {event1, event2, event3, event4, event5}); SupportViewDataCheck.checkNewData(childView, null); }
@Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (0 == newData.length)) { LOG.error("Input Time Slide Window newData is Null!"); return; } try { lock(); long timestamp = System.currentTimeMillis(); for (int i = 0; i < newData.length; i++) { events.add(timestamp + getKeepTime(), newData[i]); } IDataCollection dataCollection = getDataCollection(); if (dataCollection != null) { dataCollection.update(newData, null); } if (this.hasViews()) { updateChild(newData, null); } } finally { if (isLocked()) { unlock(); } } }
TimeSlideWindow extends TimeBasedWindow { @Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (0 == newData.length)) { LOG.error("Input Time Slide Window newData is Null!"); return; } try { lock(); long timestamp = System.currentTimeMillis(); for (int i = 0; i < newData.length; i++) { events.add(timestamp + getKeepTime(), newData[i]); } IDataCollection dataCollection = getDataCollection(); if (dataCollection != null) { dataCollection.update(newData, null); } if (this.hasViews()) { updateChild(newData, null); } } finally { if (isLocked()) { unlock(); } } } }
TimeSlideWindow extends TimeBasedWindow { @Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (0 == newData.length)) { LOG.error("Input Time Slide Window newData is Null!"); return; } try { lock(); long timestamp = System.currentTimeMillis(); for (int i = 0; i < newData.length; i++) { events.add(timestamp + getKeepTime(), newData[i]); } IDataCollection dataCollection = getDataCollection(); if (dataCollection != null) { dataCollection.update(newData, null); } if (this.hasViews()) { updateChild(newData, null); } } finally { if (isLocked()) { unlock(); } } } TimeSlideWindow(long keepTime, long slideInterval); }
TimeSlideWindow extends TimeBasedWindow { @Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (0 == newData.length)) { LOG.error("Input Time Slide Window newData is Null!"); return; } try { lock(); long timestamp = System.currentTimeMillis(); for (int i = 0; i < newData.length; i++) { events.add(timestamp + getKeepTime(), newData[i]); } IDataCollection dataCollection = getDataCollection(); if (dataCollection != null) { dataCollection.update(newData, null); } if (this.hasViews()) { updateChild(newData, null); } } finally { if (isLocked()) { unlock(); } } } TimeSlideWindow(long keepTime, long slideInterval); @Override void update(IEvent[] newData, IEvent[] oldData); @Override void timerCallBack(long currentTime); }
TimeSlideWindow extends TimeBasedWindow { @Override public void update(IEvent[] newData, IEvent[] oldData) { if ((null == newData) || (0 == newData.length)) { LOG.error("Input Time Slide Window newData is Null!"); return; } try { lock(); long timestamp = System.currentTimeMillis(); for (int i = 0; i < newData.length; i++) { events.add(timestamp + getKeepTime(), newData[i]); } IDataCollection dataCollection = getDataCollection(); if (dataCollection != null) { dataCollection.update(newData, null); } if (this.hasViews()) { updateChild(newData, null); } } finally { if (isLocked()) { unlock(); } } } TimeSlideWindow(long keepTime, long slideInterval); @Override void update(IEvent[] newData, IEvent[] oldData); @Override void timerCallBack(long currentTime);  }