Update TestTopic to confirm to the new Topic API
Change-Id: Ieade9340ac32d50f4e8304f3a347687a77ef49e3
diff --git a/core/api/src/test/java/org/onosproject/store/service/TestTopic.java b/core/api/src/test/java/org/onosproject/store/service/TestTopic.java
index 2155255..72c05ce 100644
--- a/core/api/src/test/java/org/onosproject/store/service/TestTopic.java
+++ b/core/api/src/test/java/org/onosproject/store/service/TestTopic.java
@@ -16,10 +16,11 @@
package org.onosproject.store.service;
-import com.google.common.collect.Sets;
+import com.google.common.collect.Maps;
-import java.util.Set;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
import java.util.function.Consumer;
/**
@@ -27,7 +28,7 @@
*/
public class TestTopic<T> implements Topic<T> {
private final String name;
- private final Set<Consumer<T>> callbacks = Sets.newConcurrentHashSet();
+ private final Map<Consumer<T>, Executor> callbacks = Maps.newIdentityHashMap();
public TestTopic(String name) {
this.name = name;
@@ -35,13 +36,17 @@
@Override
public CompletableFuture<Void> publish(T message) {
- callbacks.forEach(c -> c.accept(message));
+ callbacks.forEach((k, v) -> {
+ v.execute(() -> {
+ k.accept(message);
+ });
+ });
return CompletableFuture.completedFuture(null);
}
@Override
- public CompletableFuture<Void> subscribe(Consumer<T> callback) {
- callbacks.add(callback);
+ public CompletableFuture<Void> subscribe(Consumer<T> callback, Executor executor) {
+ callbacks.put(callback, executor);
return CompletableFuture.completedFuture(null);
}