홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Thomas Vachuska
Committed by Ray Milkey
ecb63c5a 1 parent 8546348b

Updated accumulator documentation and refactored names to remove the event heritage. 

Change-Id: I2238ab1215281702e670a406fb901ba8a4ef85ce
Inline Side-by-side
Showing 8 changed files with 88 additions and 156 deletions
1 -/*
2 - * Copyright 2014 Open Networking Laboratory
3 - *
4 - * Licensed under the Apache License, Version 2.0 (the "License");
5 - * you may not use this file except in compliance with the License.
6 - * You may obtain a copy of the License at
7 - *
8 - * http://www.apache.org/licenses/LICENSE-2.0
9 - *
10 - * Unless required by applicable law or agreed to in writing, software
11 - * distributed under the License is distributed on an "AS IS" BASIS,
12 - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 - * See the License for the specific language governing permissions and
14 - * limitations under the License.
15 - */
16 -package org.onosproject.event;
17 -
18 -import org.onlab.util.AbstractAccumulator;
19 -
20 -import java.util.Timer;
21 -
22 -/**
23 - * Base implementation of an event accumulator. It allows triggering based on
24 - * event inter-arrival time threshold, maximum batch life threshold and maximum
25 - * batch size.
26 - */
27 -public abstract class AbstractEventAccumulator
28 - extends AbstractAccumulator<Event>
29 - implements EventAccumulator {
30 -
31 - /**
32 - * Creates an event accumulator capable of triggering on the specified
33 - * thresholds.
34 - *
35 - * @param timer timer to use for scheduling check-points
36 - * @param maxEvents maximum number of events to accumulate before
37 - * processing is triggered
38 - * @param maxBatchMillis maximum number of millis allowed since the first
39 - * event before processing is triggered
40 - * @param maxIdleMillis maximum number millis between events before
41 - * processing is triggered
42 - */
43 - protected AbstractEventAccumulator(Timer timer, int maxEvents,
44 - int maxBatchMillis, int maxIdleMillis) {
45 - super(timer, maxEvents, maxBatchMillis, maxIdleMillis);
46 - }
47 -}
1 -/*
2 - * Copyright 2014 Open Networking Laboratory
3 - *
4 - * Licensed under the Apache License, Version 2.0 (the "License");
5 - * you may not use this file except in compliance with the License.
6 - * You may obtain a copy of the License at
7 - *
8 - * http://www.apache.org/licenses/LICENSE-2.0
9 - *
10 - * Unless required by applicable law or agreed to in writing, software
11 - * distributed under the License is distributed on an "AS IS" BASIS,
12 - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 - * See the License for the specific language governing permissions and
14 - * limitations under the License.
15 - */
16 -package org.onosproject.event;
17 -
18 -import org.onlab.util.Accumulator;
19 -
20 -/**
21 - * Abstraction of an accumulator capable of collecting events and at some
22 - * point in time triggers processing of all previously accumulated events.
23 - */
24 -public interface EventAccumulator extends Accumulator<Event> {
25 -}
...@@ -53,8 +53,8 @@ public class IntentAccumulator extends AbstractAccumulator<IntentData> { ...@@ -53,8 +53,8 @@ public class IntentAccumulator extends AbstractAccumulator<IntentData> {
53 } 53 }
54 54
55 @Override 55 @Override
56 - public void processEvents(List<IntentData> ops) { 56 + public void processItems(List<IntentData> items) {
57 - delegate.execute(reduce(ops)); 57 + delegate.execute(reduce(items));
58 // FIXME kick off the work 58 // FIXME kick off the work
59 //for (IntentData data : opMap.values()) {} 59 //for (IntentData data : opMap.values()) {}
60 } 60 }
......
...@@ -16,7 +16,6 @@ ...@@ -16,7 +16,6 @@
16 package org.onosproject.net.topology.impl; 16 package org.onosproject.net.topology.impl;
17 17
18 import com.google.common.collect.ImmutableList; 18 import com.google.common.collect.ImmutableList;
19 -
20 import org.apache.felix.scr.annotations.Activate; 19 import org.apache.felix.scr.annotations.Activate;
21 import org.apache.felix.scr.annotations.Component; 20 import org.apache.felix.scr.annotations.Component;
22 import org.apache.felix.scr.annotations.Deactivate; 21 import org.apache.felix.scr.annotations.Deactivate;
...@@ -25,9 +24,9 @@ import org.apache.felix.scr.annotations.Property; ...@@ -25,9 +24,9 @@ import org.apache.felix.scr.annotations.Property;
25 import org.apache.felix.scr.annotations.Reference; 24 import org.apache.felix.scr.annotations.Reference;
26 import org.apache.felix.scr.annotations.ReferenceCardinality; 25 import org.apache.felix.scr.annotations.ReferenceCardinality;
27 import org.apache.felix.scr.annotations.Service; 26 import org.apache.felix.scr.annotations.Service;
28 -import org.onosproject.event.AbstractEventAccumulator; 27 +import org.onlab.util.AbstractAccumulator;
28 +import org.onlab.util.Accumulator;
29 import org.onosproject.event.Event; 29 import org.onosproject.event.Event;
30 -import org.onosproject.event.EventAccumulator;
31 import org.onosproject.net.device.DeviceEvent; 30 import org.onosproject.net.device.DeviceEvent;
32 import org.onosproject.net.device.DeviceListener; 31 import org.onosproject.net.device.DeviceListener;
33 import org.onosproject.net.device.DeviceService; 32 import org.onosproject.net.device.DeviceService;
...@@ -51,9 +50,9 @@ import java.util.concurrent.ExecutorService; ...@@ -51,9 +50,9 @@ import java.util.concurrent.ExecutorService;
51 50
52 import static com.google.common.base.Strings.isNullOrEmpty; 51 import static com.google.common.base.Strings.isNullOrEmpty;
53 import static java.util.concurrent.Executors.newFixedThreadPool; 52 import static java.util.concurrent.Executors.newFixedThreadPool;
53 +import static org.onlab.util.Tools.namedThreads;
54 import static org.onosproject.core.CoreService.CORE_PROVIDER_ID; 54 import static org.onosproject.core.CoreService.CORE_PROVIDER_ID;
55 import static org.onosproject.net.device.DeviceEvent.Type.*; 55 import static org.onosproject.net.device.DeviceEvent.Type.*;
56 -import static org.onlab.util.Tools.namedThreads;
57 import static org.slf4j.LoggerFactory.getLogger; 56 import static org.slf4j.LoggerFactory.getLogger;
58 57
59 /** 58 /**
...@@ -104,7 +103,7 @@ public class DefaultTopologyProvider extends AbstractProvider ...@@ -104,7 +103,7 @@ public class DefaultTopologyProvider extends AbstractProvider
104 private DeviceListener deviceListener = new InternalDeviceListener(); 103 private DeviceListener deviceListener = new InternalDeviceListener();
105 private LinkListener linkListener = new InternalLinkListener(); 104 private LinkListener linkListener = new InternalLinkListener();
106 105
107 - private EventAccumulator accumulator; 106 + private Accumulator<Event> accumulator;
108 private ExecutorService executor; 107 private ExecutorService executor;
109 108
110 /** 109 /**
...@@ -245,18 +244,15 @@ public class DefaultTopologyProvider extends AbstractProvider ...@@ -245,18 +244,15 @@ public class DefaultTopologyProvider extends AbstractProvider
245 } 244 }
246 245
247 // Event accumulator for paced triggering of topology assembly. 246 // Event accumulator for paced triggering of topology assembly.
248 - private class TopologyChangeAccumulator 247 + private class TopologyChangeAccumulator extends AbstractAccumulator<Event> {
249 - extends AbstractEventAccumulator implements EventAccumulator {
250 -
251 TopologyChangeAccumulator() { 248 TopologyChangeAccumulator() {
252 super(TIMER, maxEvents, maxBatchMs, maxIdleMs); 249 super(TIMER, maxEvents, maxBatchMs, maxIdleMs);
253 } 250 }
254 251
255 @Override 252 @Override
256 - public void processEvents(List<Event> events) { 253 + public void processItems(List<Event> items) {
257 - triggerTopologyBuild(events); 254 + triggerTopologyBuild(items);
258 } 255 }
259 -
260 } 256 }
261 257
262 // Task for building topology data in a separate thread. 258 // Task for building topology data in a separate thread.
......
...@@ -27,64 +27,63 @@ import static com.google.common.base.Preconditions.checkArgument; ...@@ -27,64 +27,63 @@ import static com.google.common.base.Preconditions.checkArgument;
27 import static com.google.common.base.Preconditions.checkNotNull; 27 import static com.google.common.base.Preconditions.checkNotNull;
28 28
29 /** 29 /**
30 - * Base implementation of an event accumulator. It allows triggering based on 30 + * Base implementation of an item accumulator. It allows triggering based on
31 - * event inter-arrival time threshold, maximum batch life threshold and maximum 31 + * item inter-arrival time threshold, maximum batch life threshold and maximum
32 * batch size. 32 * batch size.
33 */ 33 */
34 -// FIXME refactor the names here
35 public abstract class AbstractAccumulator<T> implements Accumulator<T> { 34 public abstract class AbstractAccumulator<T> implements Accumulator<T> {
36 35
37 private Logger log = LoggerFactory.getLogger(AbstractAccumulator.class); 36 private Logger log = LoggerFactory.getLogger(AbstractAccumulator.class);
38 37
39 private final Timer timer; 38 private final Timer timer;
40 - private final int maxEvents; 39 + private final int maxItems;
41 private final int maxBatchMillis; 40 private final int maxBatchMillis;
42 private final int maxIdleMillis; 41 private final int maxIdleMillis;
43 42
44 private TimerTask idleTask = new ProcessorTask(); 43 private TimerTask idleTask = new ProcessorTask();
45 private TimerTask maxTask = new ProcessorTask(); 44 private TimerTask maxTask = new ProcessorTask();
46 45
47 - private List<T> events = Lists.newArrayList(); 46 + private List<T> items = Lists.newArrayList();
48 47
49 /** 48 /**
50 - * Creates an event accumulator capable of triggering on the specified 49 + * Creates an item accumulator capable of triggering on the specified
51 * thresholds. 50 * thresholds.
52 * 51 *
53 - * @param timer timer to use for scheduling check-points 52 + * @param timer timer to use for scheduling check-points
54 - * @param maxEvents maximum number of events to accumulate before 53 + * @param maxItems maximum number of items to accumulate before
55 - * processing is triggered 54 + * processing is triggered
56 - * @param maxBatchMillis maximum number of millis allowed since the first 55 + * @param maxBatchMillis maximum number of millis allowed since the first
57 - * event before processing is triggered 56 + * item before processing is triggered
58 - * @param maxIdleMillis maximum number millis between events before 57 + * @param maxIdleMillis maximum number millis between items before
59 - * processing is triggered 58 + * processing is triggered
60 */ 59 */
61 - protected AbstractAccumulator(Timer timer, int maxEvents, 60 + protected AbstractAccumulator(Timer timer, int maxItems,
62 int maxBatchMillis, int maxIdleMillis) { 61 int maxBatchMillis, int maxIdleMillis) {
63 this.timer = checkNotNull(timer, "Timer cannot be null"); 62 this.timer = checkNotNull(timer, "Timer cannot be null");
64 63
65 - checkArgument(maxEvents > 1, "Maximum number of events must be > 1"); 64 + checkArgument(maxItems > 1, "Maximum number of items must be > 1");
66 checkArgument(maxBatchMillis > 0, "Maximum millis must be positive"); 65 checkArgument(maxBatchMillis > 0, "Maximum millis must be positive");
67 checkArgument(maxIdleMillis > 0, "Maximum idle millis must be positive"); 66 checkArgument(maxIdleMillis > 0, "Maximum idle millis must be positive");
68 67
69 - this.maxEvents = maxEvents; 68 + this.maxItems = maxItems;
70 this.maxBatchMillis = maxBatchMillis; 69 this.maxBatchMillis = maxBatchMillis;
71 this.maxIdleMillis = maxIdleMillis; 70 this.maxIdleMillis = maxIdleMillis;
72 } 71 }
73 72
74 @Override 73 @Override
75 - public synchronized void add(T event) { 74 + public synchronized void add(T item) {
76 idleTask = cancelIfActive(idleTask); 75 idleTask = cancelIfActive(idleTask);
77 - events.add(checkNotNull(event, "Event cannot be null")); 76 + items.add(checkNotNull(item, "Item cannot be null"));
78 77
79 - // Did we hit the max event threshold? 78 + // Did we hit the max item threshold?
80 - if (events.size() == maxEvents) { 79 + if (items.size() == maxItems) {
81 maxTask = cancelIfActive(maxTask); 80 maxTask = cancelIfActive(maxTask);
82 schedule(1); 81 schedule(1);
83 } else { 82 } else {
84 - // Otherwise, schedule idle task and if this is a first event 83 + // Otherwise, schedule idle task and if this is a first item
85 // also schedule the max batch age task. 84 // also schedule the max batch age task.
86 idleTask = schedule(maxIdleMillis); 85 idleTask = schedule(maxIdleMillis);
87 - if (events.size() == 1) { 86 + if (items.size() == 1) {
88 maxTask = schedule(maxBatchMillis); 87 maxTask = schedule(maxBatchMillis);
89 } 88 }
90 } 89 }
...@@ -105,24 +104,24 @@ public abstract class AbstractAccumulator<T> implements Accumulator<T> { ...@@ -105,24 +104,24 @@ public abstract class AbstractAccumulator<T> implements Accumulator<T> {
105 return task; 104 return task;
106 } 105 }
107 106
108 - // Task for triggering processing of accumulated events 107 + // Task for triggering processing of accumulated items
109 private class ProcessorTask extends TimerTask { 108 private class ProcessorTask extends TimerTask {
110 @Override 109 @Override
111 public void run() { 110 public void run() {
112 try { 111 try {
113 idleTask = cancelIfActive(idleTask); 112 idleTask = cancelIfActive(idleTask);
114 maxTask = cancelIfActive(maxTask); 113 maxTask = cancelIfActive(maxTask);
115 - processEvents(finalizeCurrentBatch()); 114 + processItems(finalizeCurrentBatch());
116 } catch (Exception e) { 115 } catch (Exception e) {
117 log.warn("Unable to process batch due to {}", e); 116 log.warn("Unable to process batch due to {}", e);
118 } 117 }
119 } 118 }
120 } 119 }
121 120
122 - // Demotes and returns the current batch of events and promotes a new one. 121 + // Demotes and returns the current batch of items and promotes a new one.
123 private synchronized List<T> finalizeCurrentBatch() { 122 private synchronized List<T> finalizeCurrentBatch() {
124 - List<T> toBeProcessed = events; 123 + List<T> toBeProcessed = items;
125 - events = Lists.newArrayList(); 124 + items = Lists.newArrayList();
126 return toBeProcessed; 125 return toBeProcessed;
127 } 126 }
128 127
...@@ -136,18 +135,18 @@ public abstract class AbstractAccumulator<T> implements Accumulator<T> { ...@@ -136,18 +135,18 @@ public abstract class AbstractAccumulator<T> implements Accumulator<T> {
136 } 135 }
137 136
138 /** 137 /**
139 - * Returns the maximum number of events allowed to accumulate before 138 + * Returns the maximum number of items allowed to accumulate before
140 * processing is triggered. 139 * processing is triggered.
141 * 140 *
142 - * @return max number of events 141 + * @return max number of items
143 */ 142 */
144 - public int maxEvents() { 143 + public int maxItems() {
145 - return maxEvents; 144 + return maxItems;
146 } 145 }
147 146
148 /** 147 /**
149 * Returns the maximum number of millis allowed to expire since the first 148 * Returns the maximum number of millis allowed to expire since the first
150 - * event before processing is triggered. 149 + * item before processing is triggered.
151 * 150 *
152 * @return max number of millis a batch is allowed to last 151 * @return max number of millis a batch is allowed to last
153 */ 152 */
...@@ -157,9 +156,9 @@ public abstract class AbstractAccumulator<T> implements Accumulator<T> { ...@@ -157,9 +156,9 @@ public abstract class AbstractAccumulator<T> implements Accumulator<T> {
157 156
158 /** 157 /**
159 * Returns the maximum number of millis allowed to expire since the last 158 * Returns the maximum number of millis allowed to expire since the last
160 - * event arrival before processing is triggered. 159 + * item arrival before processing is triggered.
161 * 160 *
162 - * @return max number of millis since the last event 161 + * @return max number of millis since the last item
163 */ 162 */
164 public int maxIdleMillis() { 163 public int maxIdleMillis() {
165 return maxIdleMillis; 164 return maxIdleMillis;
......
...@@ -18,25 +18,28 @@ package org.onlab.util; ...@@ -18,25 +18,28 @@ package org.onlab.util;
18 import java.util.List; 18 import java.util.List;
19 19
20 /** 20 /**
21 - * Abstraction of an accumulator capable of collecting events and at some 21 + * Abstraction of an accumulator capable of collecting items and at some
22 - * point in time triggers processing of all previously accumulated events. 22 + * point in time triggers processing of all previously accumulated items.
23 + *
24 + * @param <T> item type
23 */ 25 */
24 public interface Accumulator<T> { 26 public interface Accumulator<T> {
25 27
26 /** 28 /**
27 - * Adds an event to the current batch. This operation may, or may not 29 + * Adds an item to the current batch. This operation may, or may not
28 - * trigger processing of the current batch of events. 30 + * trigger processing of the current batch of items.
29 * 31 *
30 - * @param event event to be added to the current batch 32 + * @param item item to be added to the current batch
31 */ 33 */
32 - void add(T event); 34 + void add(T item);
33 35
34 /** 36 /**
35 - * Processes the specified list of accumulated events. 37 + * Processes the specified list of accumulated items.
36 * 38 *
37 - * @param events list of accumulated events 39 + * @param items list of accumulated items
38 */ 40 */
39 - void processEvents(List<T> events); 41 + void processItems(List<T> items);
40 42
41 //TODO consider a blocking version that required consumer participation 43 //TODO consider a blocking version that required consumer participation
44 +
42 } 45 }
......
1 /* 1 /*
2 - * Copyright 2014 Open Networking Laboratory 2 + * Copyright 2015 Open Networking Laboratory
3 * 3 *
4 * Licensed under the Apache License, Version 2.0 (the "License"); 4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License. 5 * you may not use this file except in compliance with the License.
...@@ -13,7 +13,7 @@ ...@@ -13,7 +13,7 @@
13 * See the License for the specific language governing permissions and 13 * See the License for the specific language governing permissions and
14 * limitations under the License. 14 * limitations under the License.
15 */ 15 */
16 -package org.onosproject.event; 16 +package org.onlab.util;
17 17
18 import org.junit.Ignore; 18 import org.junit.Ignore;
19 import org.junit.Test; 19 import org.junit.Test;
...@@ -21,16 +21,13 @@ import org.junit.Test; ...@@ -21,16 +21,13 @@ import org.junit.Test;
21 import java.util.List; 21 import java.util.List;
22 import java.util.Timer; 22 import java.util.Timer;
23 23
24 -import static org.junit.Assert.assertEquals; 24 +import static org.junit.Assert.*;
25 -import static org.junit.Assert.assertFalse;
26 -import static org.junit.Assert.assertTrue;
27 import static org.onlab.junit.TestTools.delay; 25 import static org.onlab.junit.TestTools.delay;
28 -import static org.onosproject.event.TestEvent.Type.FOO;
29 26
30 /** 27 /**
31 * Tests the operation of the accumulator. 28 * Tests the operation of the accumulator.
32 */ 29 */
33 -public class AbstractEventAccumulatorTest { 30 +public class AbstractAccumulatorTest {
34 31
35 private final Timer timer = new Timer(); 32 private final Timer timer = new Timer();
36 33
...@@ -38,7 +35,7 @@ public class AbstractEventAccumulatorTest { ...@@ -38,7 +35,7 @@ public class AbstractEventAccumulatorTest {
38 public void basics() throws Exception { 35 public void basics() throws Exception {
39 TestAccumulator accumulator = new TestAccumulator(); 36 TestAccumulator accumulator = new TestAccumulator();
40 assertEquals("incorrect timer", timer, accumulator.timer()); 37 assertEquals("incorrect timer", timer, accumulator.timer());
41 - assertEquals("incorrect max events", 5, accumulator.maxEvents()); 38 + assertEquals("incorrect max events", 5, accumulator.maxItems());
42 assertEquals("incorrect max ms", 100, accumulator.maxBatchMillis()); 39 assertEquals("incorrect max ms", 100, accumulator.maxBatchMillis());
43 assertEquals("incorrect idle ms", 50, accumulator.maxIdleMillis()); 40 assertEquals("incorrect idle ms", 50, accumulator.maxIdleMillis());
44 } 41 }
...@@ -46,12 +43,12 @@ public class AbstractEventAccumulatorTest { ...@@ -46,12 +43,12 @@ public class AbstractEventAccumulatorTest {
46 @Test 43 @Test
47 public void eventTrigger() { 44 public void eventTrigger() {
48 TestAccumulator accumulator = new TestAccumulator(); 45 TestAccumulator accumulator = new TestAccumulator();
49 - accumulator.add(new TestEvent(FOO, "a")); 46 + accumulator.add(new TestItem("a"));
50 - accumulator.add(new TestEvent(FOO, "b")); 47 + accumulator.add(new TestItem("b"));
51 - accumulator.add(new TestEvent(FOO, "c")); 48 + accumulator.add(new TestItem("c"));
52 - accumulator.add(new TestEvent(FOO, "d")); 49 + accumulator.add(new TestItem("d"));
53 assertTrue("should not have fired yet", accumulator.batch.isEmpty()); 50 assertTrue("should not have fired yet", accumulator.batch.isEmpty());
54 - accumulator.add(new TestEvent(FOO, "e")); 51 + accumulator.add(new TestItem("e"));
55 delay(20); 52 delay(20);
56 assertFalse("should have fired", accumulator.batch.isEmpty()); 53 assertFalse("should have fired", accumulator.batch.isEmpty());
57 assertEquals("incorrect batch", "abcde", accumulator.batch); 54 assertEquals("incorrect batch", "abcde", accumulator.batch);
...@@ -61,16 +58,16 @@ public class AbstractEventAccumulatorTest { ...@@ -61,16 +58,16 @@ public class AbstractEventAccumulatorTest {
61 @Test 58 @Test
62 public void timeTrigger() { 59 public void timeTrigger() {
63 TestAccumulator accumulator = new TestAccumulator(); 60 TestAccumulator accumulator = new TestAccumulator();
64 - accumulator.add(new TestEvent(FOO, "a")); 61 + accumulator.add(new TestItem("a"));
65 delay(30); 62 delay(30);
66 assertTrue("should not have fired yet", accumulator.batch.isEmpty()); 63 assertTrue("should not have fired yet", accumulator.batch.isEmpty());
67 - accumulator.add(new TestEvent(FOO, "b")); 64 + accumulator.add(new TestItem("b"));
68 delay(30); 65 delay(30);
69 assertTrue("should not have fired yet", accumulator.batch.isEmpty()); 66 assertTrue("should not have fired yet", accumulator.batch.isEmpty());
70 - accumulator.add(new TestEvent(FOO, "c")); 67 + accumulator.add(new TestItem("c"));
71 delay(30); 68 delay(30);
72 assertTrue("should not have fired yet", accumulator.batch.isEmpty()); 69 assertTrue("should not have fired yet", accumulator.batch.isEmpty());
73 - accumulator.add(new TestEvent(FOO, "d")); 70 + accumulator.add(new TestItem("d"));
74 delay(30); 71 delay(30);
75 assertFalse("should have fired", accumulator.batch.isEmpty()); 72 assertFalse("should have fired", accumulator.batch.isEmpty());
76 assertEquals("incorrect batch", "abcd", accumulator.batch); 73 assertEquals("incorrect batch", "abcd", accumulator.batch);
...@@ -79,15 +76,23 @@ public class AbstractEventAccumulatorTest { ...@@ -79,15 +76,23 @@ public class AbstractEventAccumulatorTest {
79 @Test 76 @Test
80 public void idleTrigger() { 77 public void idleTrigger() {
81 TestAccumulator accumulator = new TestAccumulator(); 78 TestAccumulator accumulator = new TestAccumulator();
82 - accumulator.add(new TestEvent(FOO, "a")); 79 + accumulator.add(new TestItem("a"));
83 assertTrue("should not have fired yet", accumulator.batch.isEmpty()); 80 assertTrue("should not have fired yet", accumulator.batch.isEmpty());
84 - accumulator.add(new TestEvent(FOO, "b")); 81 + accumulator.add(new TestItem("b"));
85 delay(80); 82 delay(80);
86 assertFalse("should have fired", accumulator.batch.isEmpty()); 83 assertFalse("should have fired", accumulator.batch.isEmpty());
87 assertEquals("incorrect batch", "ab", accumulator.batch); 84 assertEquals("incorrect batch", "ab", accumulator.batch);
88 } 85 }
89 86
90 - private class TestAccumulator extends AbstractEventAccumulator { 87 + private class TestItem {
88 + private final String s;
89 +
90 + public TestItem(String s) {
91 + this.s = s;
92 + }
93 + }
94 +
95 + private class TestAccumulator extends AbstractAccumulator<TestItem> {
91 96
92 String batch = ""; 97 String batch = "";
93 98
...@@ -96,10 +101,11 @@ public class AbstractEventAccumulatorTest { ...@@ -96,10 +101,11 @@ public class AbstractEventAccumulatorTest {
96 } 101 }
97 102
98 @Override 103 @Override
99 - public void processEvents(List<Event> events) { 104 + public void processItems(List<TestItem> items) {
100 - for (Event event : events) { 105 + for (TestItem item : items) {
101 - batch += event.subject(); 106 + batch += item.s;
102 } 107 }
103 } 108 }
104 } 109 }
110 +
105 } 111 }
......
...@@ -20,14 +20,14 @@ import com.fasterxml.jackson.databind.node.ArrayNode; ...@@ -20,14 +20,14 @@ import com.fasterxml.jackson.databind.node.ArrayNode;
20 import com.fasterxml.jackson.databind.node.ObjectNode; 20 import com.fasterxml.jackson.databind.node.ObjectNode;
21 import org.eclipse.jetty.websocket.WebSocket; 21 import org.eclipse.jetty.websocket.WebSocket;
22 import org.onlab.osgi.ServiceDirectory; 22 import org.onlab.osgi.ServiceDirectory;
23 +import org.onlab.util.AbstractAccumulator;
24 +import org.onlab.util.Accumulator;
23 import org.onosproject.cluster.ClusterEvent; 25 import org.onosproject.cluster.ClusterEvent;
24 import org.onosproject.cluster.ClusterEventListener; 26 import org.onosproject.cluster.ClusterEventListener;
25 import org.onosproject.cluster.ControllerNode; 27 import org.onosproject.cluster.ControllerNode;
26 import org.onosproject.core.ApplicationId; 28 import org.onosproject.core.ApplicationId;
27 import org.onosproject.core.CoreService; 29 import org.onosproject.core.CoreService;
28 -import org.onosproject.event.AbstractEventAccumulator;
29 import org.onosproject.event.Event; 30 import org.onosproject.event.Event;
30 -import org.onosproject.event.EventAccumulator;
31 import org.onosproject.mastership.MastershipAdminService; 31 import org.onosproject.mastership.MastershipAdminService;
32 import org.onosproject.mastership.MastershipEvent; 32 import org.onosproject.mastership.MastershipEvent;
33 import org.onosproject.mastership.MastershipListener; 33 import org.onosproject.mastership.MastershipListener;
...@@ -120,7 +120,7 @@ public class TopologyViewWebSocket ...@@ -120,7 +120,7 @@ public class TopologyViewWebSocket
120 private final IntentListener intentListener = new InternalIntentListener(); 120 private final IntentListener intentListener = new InternalIntentListener();
121 private final FlowRuleListener flowListener = new InternalFlowListener(); 121 private final FlowRuleListener flowListener = new InternalFlowListener();
122 122
123 - private final EventAccumulator eventAccummulator = new InternalEventAccummulator(); 123 + private final Accumulator<Event> eventAccummulator = new InternalEventAccummulator();
124 124
125 private TimerTask trafficTask; 125 private TimerTask trafficTask;
126 private ObjectNode trafficEvent; 126 private ObjectNode trafficEvent;
...@@ -721,13 +721,13 @@ public class TopologyViewWebSocket ...@@ -721,13 +721,13 @@ public class TopologyViewWebSocket
721 } 721 }
722 722
723 // Accumulates events to drive methodic update of the summary pane. 723 // Accumulates events to drive methodic update of the summary pane.
724 - private class InternalEventAccummulator extends AbstractEventAccumulator { 724 + private class InternalEventAccummulator extends AbstractAccumulator<Event> {
725 protected InternalEventAccummulator() { 725 protected InternalEventAccummulator() {
726 super(new Timer("topo-summary"), MAX_EVENTS, MAX_BATCH_MS, MAX_IDLE_MS); 726 super(new Timer("topo-summary"), MAX_EVENTS, MAX_BATCH_MS, MAX_IDLE_MS);
727 } 727 }
728 728
729 @Override 729 @Override
730 - public void processEvents(List<Event> events) { 730 + public void processItems(List<Event> items) {
731 try { 731 try {
732 if (summaryEvent != null) { 732 if (summaryEvent != null) {
733 sendMessage(summmaryMessage(0)); 733 sendMessage(summmaryMessage(0));
......