LengthSlideWindow extends LengthBasedWindow { @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 event : newData) { events.add(event); } int expireCount = events.size() - getKeepLength(); IEvent[] expireData = null; if (expireCount > 0) { expireData = new IEvent[expireCount]; for (int i = 0; i < expireCount; i++) { expireData[i] = events.removeFirst(); } } IDataCollection dataCollection = getDataCollection(); if (dataCollection != null) { dataCollection.update(newData, expireData); } if (this.hasViews()) { updateChild(newData, expireData); } } LengthSlideWindow(int keepLength); @Override void update(IEvent[] newData, IEvent[] oldData); @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, 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}); 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); SupportViewDataCheck.checkOldData(childView, new IEvent[] {event1}); SupportViewDataCheck.checkNewData(childView, new IEvent[] {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); SupportViewDataCheck.checkOldData(childView, new IEvent[] {event2}); SupportViewDataCheck.checkNewData(childView, new IEvent[] {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); SupportViewDataCheck.checkOldData(childView, new IEvent[] {event3}); SupportViewDataCheck.checkNewData(childView, new IEvent[] {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); SupportViewDataCheck.checkOldData(childView, new IEvent[] {event4}); SupportViewDataCheck.checkNewData(childView, new IEvent[] {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[] {event5}); SupportViewDataCheck.checkNewData(childView, new IEvent[] {event10}); }