Update the device discovery protos to include the core POD name.
Change-Id: I0d562487f5ea5ba5e5ccfbb89d276a254edc612c
diff --git a/kafka/kafka_inter_container_library.go b/kafka/kafka_inter_container_library.go
index 4a3ec92..2d2714f 100644
--- a/kafka/kafka_inter_container_library.go
+++ b/kafka/kafka_inter_container_library.go
@@ -26,6 +26,7 @@
"github.com/opencord/voltha-go/common/log"
ic "github.com/opencord/voltha-go/protos/inter_container"
"reflect"
+ "strings"
"sync"
"time"
)
@@ -178,7 +179,7 @@
}
// DeviceDiscovered publish the discovered device onto the kafka messaging bus
-func (kp *InterContainerProxy) DeviceDiscovered(deviceId string, deviceType string, parentId string) error {
+func (kp *InterContainerProxy) DeviceDiscovered(deviceId string, deviceType string, parentId string, publisher string) error {
log.Debugw("sending-device-discovery-msg", log.Fields{"deviceId": deviceId})
// Simple validation
if deviceId == "" || deviceType == "" {
@@ -197,6 +198,7 @@
Id: deviceId,
DeviceType: deviceType,
ParentId: parentId,
+ Publisher:publisher,
}
var marshalledData *any.Any
@@ -560,6 +562,9 @@
func CallFuncByName(myClass interface{}, funcName string, params ...interface{}) (out []reflect.Value, err error) {
myClassValue := reflect.ValueOf(myClass)
+ // Capitalize the first letter in the funcName to workaround the first capital letters required to
+ // invoke a function from a different package
+ funcName = strings.Title(funcName)
m := myClassValue.MethodByName(funcName)
if !m.IsValid() {
return make([]reflect.Value, 0), fmt.Errorf("method-not-found \"%s\"", funcName)
diff --git a/protos/inter_container.proto b/protos/inter_container.proto
index 931ddb9..ac2af4f 100644
--- a/protos/inter_container.proto
+++ b/protos/inter_container.proto
@@ -86,6 +86,7 @@
string id = 1;
string parent_id = 2;
string device_type = 3;
+ string publisher = 4;
}
message InterAdapterMessageType {
diff --git a/rw_core/core/core.go b/rw_core/core/core.go
index 298c3ce..c42a64e 100644
--- a/rw_core/core/core.go
+++ b/rw_core/core/core.go
@@ -75,28 +75,28 @@
}
func (core *Core) Start(ctx context.Context) {
- log.Info("starting-core", log.Fields{"coreId": core.instanceId})
+ log.Info("starting-adaptercore", log.Fields{"coreId": core.instanceId})
core.startKafkaMessagingProxy(ctx)
log.Info("values", log.Fields{"kmp": core.kmp})
- core.deviceMgr = newDeviceManager(core.kmp, core.clusterDataProxy)
+ core.deviceMgr = newDeviceManager(core.kmp, core.clusterDataProxy, core.instanceId)
core.logicalDeviceMgr = newLogicalDeviceManager(core.deviceMgr, core.kmp, core.clusterDataProxy)
core.registerAdapterRequestHandler(ctx, core.instanceId, core.deviceMgr, core.logicalDeviceMgr, core.clusterDataProxy, core.localDataProxy)
go core.startDeviceManager(ctx)
go core.startLogicalDeviceManager(ctx)
go core.startGRPCService(ctx)
- log.Info("core-started")
+ log.Info("adaptercore-started")
}
func (core *Core) Stop(ctx context.Context) {
- log.Info("stopping-core")
+ log.Info("stopping-adaptercore")
core.exitChannel <- 1
// Stop all the started services
core.grpcServer.Stop()
core.logicalDeviceMgr.stop(ctx)
core.deviceMgr.stop(ctx)
core.kmp.Stop()
- log.Info("core-stopped")
+ log.Info("adaptercore-stopped")
}
//startGRPCService creates the grpc service handlers, registers it to the grpc server
diff --git a/rw_core/core/device_manager.go b/rw_core/core/device_manager.go
index 682de48..34fc956 100644
--- a/rw_core/core/device_manager.go
+++ b/rw_core/core/device_manager.go
@@ -38,16 +38,18 @@
kafkaICProxy *kafka.InterContainerProxy
stateTransitions *TransitionMap
clusterDataProxy *model.Proxy
+ coreInstanceId string
exitChannel chan int
lockDeviceAgentsMap sync.RWMutex
}
-func newDeviceManager(kafkaICProxy *kafka.InterContainerProxy, cdProxy *model.Proxy) *DeviceManager {
+func newDeviceManager(kafkaICProxy *kafka.InterContainerProxy, cdProxy *model.Proxy, coreInstanceId string) *DeviceManager {
var deviceMgr DeviceManager
deviceMgr.exitChannel = make(chan int, 1)
deviceMgr.deviceAgents = make(map[string]*DeviceAgent)
deviceMgr.adapterProxy = NewAdapterProxy(kafkaICProxy)
deviceMgr.kafkaICProxy = kafkaICProxy
+ deviceMgr.coreInstanceId = coreInstanceId
deviceMgr.clusterDataProxy = cdProxy
deviceMgr.lockDeviceAgentsMap = sync.RWMutex{}
return &deviceMgr
@@ -360,7 +362,7 @@
}
// Publish on the messaging bus that we have discovered new devices
- go dMgr.kafkaICProxy.DeviceDiscovered(agent.deviceId, deviceType, parentDeviceId)
+ go dMgr.kafkaICProxy.DeviceDiscovered(agent.deviceId, deviceType, parentDeviceId, dMgr.coreInstanceId)
return nil
}
diff --git a/tests/kafka/kafka_inter_container_messaging_test.go b/tests/kafka/kafka_inter_container_messaging_test.go
index 57e7ab9..de97558 100644
--- a/tests/kafka/kafka_inter_container_messaging_test.go
+++ b/tests/kafka/kafka_inter_container_messaging_test.go
@@ -524,7 +524,7 @@
time.Sleep(time.Millisecond * 100)
// Send the message
- go testProxy.DeviceDiscovered("TestDeviceId", "TestDevicetype", "TestParentId")
+ go testProxy.DeviceDiscovered("TestDeviceId", "TestDevicetype", "TestParentId", "myPODName")
msg := <-waitingChannel
totalTime := (time.Now().UnixNano() - msg.Header.Timestamp) / int64(time.Millisecond)
@@ -536,6 +536,7 @@
assert.Equal(t, dd.Id, "TestDeviceId")
assert.Equal(t, dd.DeviceType, "TestDevicetype")
assert.Equal(t, dd.ParentId, "TestParentId")
+ assert.Equal(t, dd.Publisher, "myPODName")
log.Debugw("TotalTime", log.Fields{"time": totalTime})
}