[VOL-3117] Read KV store data path prefix from an environment variable
- voltha-lib-go reads the KVStoreDataPrefix from env variable, so removed the command-line flag in voltha-go
- also removed unused corepair topic flag
Change-Id: Ibe8403bf187126b587a92cd9c58aa6d923f84cd0
diff --git a/rw_core/core/device/agent_test.go b/rw_core/core/device/agent_test.go
index 29d8062..cd49681 100755
--- a/rw_core/core/device/agent_test.go
+++ b/rw_core/core/device/agent_test.go
@@ -112,7 +112,7 @@
func (dat *DATest) startCore(inCompeteMode bool) {
cfg := config.NewRWCoreFlags()
- cfg.CorePairTopic = "rw_core"
+ cfg.CoreTopic = "rw_core"
cfg.DefaultRequestTimeout = dat.defaultTimeout
cfg.KVStorePort = dat.kvClientPort
cfg.InCompetingMode = inCompeteMode
@@ -129,8 +129,7 @@
Host: cfg.KVStoreHost,
Port: cfg.KVStorePort,
Timeout: cfg.KVStoreTimeout,
- LivenessChannelInterval: cfg.LiveProbeInterval / 2,
- PathPrefix: cfg.KVStoreDataPrefix}
+ LivenessChannelInterval: cfg.LiveProbeInterval / 2}
dat.kmp = kafka.NewInterContainerProxy(
kafka.InterContainerHost(cfg.KafkaAdapterHost),
kafka.InterContainerPort(cfg.KafkaAdapterPort),
@@ -142,13 +141,13 @@
proxy := model.NewDBPath(backend)
dat.adapterMgr = adapter.NewAdapterManager(proxy, dat.coreInstanceID, dat.kClient)
- dat.deviceMgr, dat.logicalDeviceMgr = NewManagers(proxy, dat.adapterMgr, dat.kmp, endpointMgr, cfg.CorePairTopic, dat.coreInstanceID, cfg.DefaultCoreTimeout)
+ dat.deviceMgr, dat.logicalDeviceMgr = NewManagers(proxy, dat.adapterMgr, dat.kmp, endpointMgr, cfg.CoreTopic, dat.coreInstanceID, cfg.DefaultCoreTimeout)
dat.adapterMgr.Start(context.Background())
if err = dat.kmp.Start(); err != nil {
logger.Fatal("Cannot start InterContainerProxy")
}
- if err := dat.kmp.SubscribeWithDefaultRequestHandler(kafka.Topic{Name: cfg.CorePairTopic}, kafka.OffsetNewest); err != nil {
+ if err := dat.kmp.SubscribeWithDefaultRequestHandler(kafka.Topic{Name: cfg.CoreTopic}, kafka.OffsetNewest); err != nil {
logger.Fatalf("Cannot add default request handler: %s", err)
}
diff --git a/rw_core/core/device/logical_agent_test.go b/rw_core/core/device/logical_agent_test.go
index 77c4b7c..2649ee4 100644
--- a/rw_core/core/device/logical_agent_test.go
+++ b/rw_core/core/device/logical_agent_test.go
@@ -454,7 +454,7 @@
func (lda *LDATest) startCore(inCompeteMode bool) {
cfg := config.NewRWCoreFlags()
- cfg.CorePairTopic = "rw_core"
+ cfg.CoreTopic = "rw_core"
cfg.DefaultRequestTimeout = lda.defaultTimeout
cfg.KVStorePort = lda.kvClientPort
cfg.InCompetingMode = inCompeteMode
@@ -471,8 +471,7 @@
Host: cfg.KVStoreHost,
Port: cfg.KVStorePort,
Timeout: cfg.KVStoreTimeout,
- LivenessChannelInterval: cfg.LiveProbeInterval / 2,
- PathPrefix: cfg.KVStoreDataPrefix}
+ LivenessChannelInterval: cfg.LiveProbeInterval / 2}
lda.kmp = kafka.NewInterContainerProxy(
kafka.InterContainerHost(cfg.KafkaAdapterHost),
kafka.InterContainerPort(cfg.KafkaAdapterPort),
@@ -484,7 +483,7 @@
proxy := model.NewDBPath(backend)
adapterMgr := adapter.NewAdapterManager(proxy, lda.coreInstanceID, lda.kClient)
- lda.deviceMgr, lda.logicalDeviceMgr = NewManagers(proxy, adapterMgr, lda.kmp, endpointMgr, cfg.CorePairTopic, lda.coreInstanceID, cfg.DefaultCoreTimeout)
+ lda.deviceMgr, lda.logicalDeviceMgr = NewManagers(proxy, adapterMgr, lda.kmp, endpointMgr, cfg.CoreTopic, lda.coreInstanceID, cfg.DefaultCoreTimeout)
if err = lda.kmp.Start(); err != nil {
logger.Fatal("Cannot start InterContainerProxy")
}
diff --git a/rw_core/core/device/manager.go b/rw_core/core/device/manager.go
index 17ac266..c7af54a 100755
--- a/rw_core/core/device/manager.go
+++ b/rw_core/core/device/manager.go
@@ -60,11 +60,11 @@
}
//NewManagers creates the Manager and the Logical Manager.
-func NewManagers(dbPath *model.Path, adapterMgr *adapter.Manager, kmp kafka.InterContainerProxy, endpointMgr kafka.EndpointManager, corePairTopic, coreInstanceID string, defaultCoreTimeout time.Duration) (*Manager, *LogicalManager) {
+func NewManagers(dbPath *model.Path, adapterMgr *adapter.Manager, kmp kafka.InterContainerProxy, endpointMgr kafka.EndpointManager, coreTopic, coreInstanceID string, defaultCoreTimeout time.Duration) (*Manager, *LogicalManager) {
deviceMgr := &Manager{
rootDevices: make(map[string]bool),
kafkaICProxy: kmp,
- adapterProxy: remote.NewAdapterProxy(kmp, corePairTopic, endpointMgr),
+ adapterProxy: remote.NewAdapterProxy(kmp, coreTopic, endpointMgr),
coreInstanceID: coreInstanceID,
dbPath: dbPath,
dProxy: dbPath.Proxy("devices"),
diff --git a/rw_core/core/device/remote/adapter_proxy.go b/rw_core/core/device/remote/adapter_proxy.go
index f4579ef..4cbb363 100755
--- a/rw_core/core/device/remote/adapter_proxy.go
+++ b/rw_core/core/device/remote/adapter_proxy.go
@@ -18,6 +18,7 @@
import (
"context"
+
"github.com/opencord/voltha-lib-go/v3/pkg/kafka"
"github.com/opencord/voltha-lib-go/v3/pkg/log"
ic "github.com/opencord/voltha-protos/v3/go/inter_container"
@@ -29,22 +30,22 @@
type AdapterProxy struct {
kafka.EndpointManager
deviceTopicRegistered bool
- corePairTopic string
+ coreTopic string
kafkaICProxy kafka.InterContainerProxy
}
// NewAdapterProxy will return adapter proxy instance
-func NewAdapterProxy(kafkaProxy kafka.InterContainerProxy, corePairTopic string, endpointManager kafka.EndpointManager) *AdapterProxy {
+func NewAdapterProxy(kafkaProxy kafka.InterContainerProxy, coreTopic string, endpointManager kafka.EndpointManager) *AdapterProxy {
return &AdapterProxy{
EndpointManager: endpointManager,
kafkaICProxy: kafkaProxy,
- corePairTopic: corePairTopic,
+ coreTopic: coreTopic,
deviceTopicRegistered: false,
}
}
func (ap *AdapterProxy) getCoreTopic() kafka.Topic {
- return kafka.Topic{Name: ap.corePairTopic}
+ return kafka.Topic{Name: ap.coreTopic}
}
func (ap *AdapterProxy) getAdapterTopic(deviceID string, adapterType string) (*kafka.Topic, error) {