Distributed topic primitive
Change-Id: Ia3ccd84c33075f297d7e6b9bc205efe92aec9bea
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StorageManager.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StorageManager.java
index bc699d5..7658793 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StorageManager.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StorageManager.java
@@ -41,6 +41,7 @@
import org.onosproject.store.primitives.PartitionService;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.AsyncAtomicValue;
import org.onosproject.store.service.AsyncConsistentMap;
import org.onosproject.store.service.AtomicCounterBuilder;
import org.onosproject.store.service.AtomicValueBuilder;
@@ -54,6 +55,7 @@
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.StorageAdminService;
import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Topic;
import org.onosproject.store.service.TransactionContextBuilder;
import org.onosproject.store.service.WorkQueue;
import org.onosproject.store.service.WorkQueueStats;
@@ -217,4 +219,13 @@
return new MapInfo(name, map.size());
}).collect(Collectors.toList());
}
+
+ @Override
+ public <T> Topic<T> getTopic(String name, Serializer serializer) {
+ AsyncAtomicValue<T> atomicValue = this.<T>atomicValueBuilder()
+ .withName("topic-" + name)
+ .withSerializer(serializer)
+ .build();
+ return new DefaultDistributedTopic<>(atomicValue);
+ }
}