VOL-2788 update voltha-go to latest voltha-lib-go and release
Change-Id: I853955d5ad3d84bae8b7b52e676872171c9ffe67
diff --git a/VERSION b/VERSION
index ca2d639..3f684d2 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-2.3.3-dev
+2.3.4
diff --git a/go.mod b/go.mod
index 9f3a469..d696801 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@
github.com/gogo/protobuf v1.3.0
github.com/golang/protobuf v1.3.2
github.com/google/uuid v1.1.1
- github.com/opencord/voltha-lib-go/v3 v3.0.15
+ github.com/opencord/voltha-lib-go/v3 v3.0.20
github.com/opencord/voltha-protos/v3 v3.2.6
github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
github.com/stretchr/testify v1.4.0
diff --git a/go.sum b/go.sum
index 67ecde6..57d22d2 100644
--- a/go.sum
+++ b/go.sum
@@ -190,8 +190,8 @@
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.4.2 h1:3mYCb7aPxS/RU7TI1y4rkEn1oKmPRjNJLNEXgw7MH2I=
github.com/onsi/gomega v1.4.2/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
-github.com/opencord/voltha-lib-go/v3 v3.0.15 h1:CA+L1iqzugusANjUYI5bneU4dY657mSzlSdzyHalMoU=
-github.com/opencord/voltha-lib-go/v3 v3.0.15/go.mod h1:69Y+rVd25Nq2SUeoY7Q1BXtwrcUPllG0erhq+aK8Qec=
+github.com/opencord/voltha-lib-go/v3 v3.0.20 h1:b3r4tKOREX63jOxDDioipPP/hkCcVwmRu8MXLx/LNMk=
+github.com/opencord/voltha-lib-go/v3 v3.0.20/go.mod h1:QuAohPQ+InSw+8XgCFxnp4cpHWcxO2efVTtiBFUmuOY=
github.com/opencord/voltha-protos/v3 v3.2.3 h1:Wv73mw1Ye0bCfyhOk5svgrlE2tLizHq6tQluoDq9Vg8=
github.com/opencord/voltha-protos/v3 v3.2.3/go.mod h1:RIGHt7b80BHpHh3ceodknh0DxUjUHCWSbYbZqRx7Og0=
github.com/opencord/voltha-protos/v3 v3.2.6 h1:Wf7PJ1Ekv9WYcbKBa9bgLNig1Y9krmy3zr6QRCug1LU=
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/adapters/adapterif/core_proxy_if.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/adapters/adapterif/core_proxy_if.go
index dbf3418..9636a7d 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/adapters/adapterif/core_proxy_if.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/adapters/adapterif/core_proxy_if.go
@@ -18,7 +18,6 @@
import (
"context"
-
"github.com/opencord/voltha-protos/v3/go/voltha"
)
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/config/configmanager.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/config/configmanager.go
index 8f96b22..462b743 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/config/configmanager.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/config/configmanager.go
@@ -24,6 +24,13 @@
"strings"
)
+func init() {
+ _, err := log.AddPackage(log.JSON, log.FatalLevel, nil)
+ if err != nil {
+ log.Errorw("unable-to-register-package-to-the-log-map", log.Fields{"error": err})
+ }
+}
+
const (
defaultkvStoreConfigPath = "config"
kvStoreDataPathPrefix = "/service/voltha"
@@ -102,6 +109,35 @@
}
}
+// RetrieveComponentList list the component Names for which loglevel is stored in kvstore
+func (c *ConfigManager) RetrieveComponentList(ctx context.Context, configType ConfigType) ([]string, error) {
+ data, err := c.backend.List(ctx, c.KvStoreConfigPrefix)
+ if err != nil {
+ return nil, err
+ }
+
+ // Looping through the data recieved from the backend for config
+ // Trimming and Splitting the required key and value from data and storing as componentName,PackageName and Level
+ // For Example, recieved key would be <Backend Prefix Path>/<Config Prefix>/<Component Name>/<Config Type>/default and value \"DEBUG\"
+ // Then in default will be stored as PackageName,componentName as <Component Name> and DEBUG will be stored as value in List struct
+ ccPathPrefix := kvStorePathSeparator + configType.String() + kvStorePathSeparator
+ pathPrefix := kvStoreDataPathPrefix + kvStorePathSeparator + c.KvStoreConfigPrefix + kvStorePathSeparator
+ var list []string
+ keys := make(map[string]interface{})
+ for attr := range data {
+ cname := strings.TrimPrefix(attr, pathPrefix)
+ cName := strings.SplitN(cname, ccPathPrefix, 2)
+ if len(cName) != 2 {
+ continue
+ }
+ if _, exist := keys[cName[0]]; !exist {
+ keys[cName[0]] = nil
+ list = append(list, cName[0])
+ }
+ }
+ return list, nil
+}
+
// Initialize the component config
func (cm *ConfigManager) InitComponentConfig(componentLabel string, configType ConfigType) *ComponentConfig {
@@ -147,8 +183,11 @@
func (c *ComponentConfig) processKVStoreWatchEvents() {
ccKeyPrefix := c.makeConfigPath()
+
log.Debugw("processing-kvstore-event-change", log.Fields{"key-prefix": ccKeyPrefix})
+
ccPathPrefix := c.cManager.backend.PathPrefix + ccKeyPrefix + kvStorePathSeparator
+
for watchResp := range c.kvStoreEventChan {
if watchResp.EventType == kvstore.CONNECTIONDOWN || watchResp.EventType == kvstore.UNKNOWN {
@@ -172,6 +211,7 @@
key := c.makeConfigPath()
log.Debugw("retreiving-list", log.Fields{"key": key})
+
data, err := c.cManager.backend.List(ctx, key)
if err != nil {
return nil, err
@@ -190,7 +230,7 @@
return res, nil
}
-func (c *ComponentConfig) Save(configKey string, configValue string, ctx context.Context) error {
+func (c *ComponentConfig) Save(ctx context.Context, configKey string, configValue string) error {
key := c.makeConfigPath() + "/" + configKey
log.Debugw("saving-key", log.Fields{"key": key, "value": configValue})
@@ -202,7 +242,7 @@
return nil
}
-func (c *ComponentConfig) Delete(configKey string, ctx context.Context) error {
+func (c *ComponentConfig) Delete(ctx context.Context, configKey string) error {
//construct key using makeConfigPath
key := c.makeConfigPath() + "/" + configKey
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go
index 04fe35d..faa86ed 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go
@@ -222,7 +222,7 @@
defer b.Unlock()
formattedPath := b.makePath(key)
- logger.Debugw("putting-key", log.Fields{"key": key, "value": string(value.([]byte)), "path": formattedPath})
+ logger.Debugw("putting-key", log.Fields{"key": key, "value": value, "path": formattedPath})
err := b.Client.Put(ctx, formattedPath, value)
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go
index 1014ada..d38f0f6 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go
@@ -29,13 +29,13 @@
// EtcdClient represents the Etcd KV store client
type EtcdClient struct {
- ectdAPI *v3Client.Client
- keyReservations map[string]*v3Client.LeaseID
- watchedChannels sync.Map
- writeLock sync.Mutex
- lockToMutexMap map[string]*v3Concurrency.Mutex
- lockToSessionMap map[string]*v3Concurrency.Session
- lockToMutexLock sync.Mutex
+ ectdAPI *v3Client.Client
+ keyReservations map[string]*v3Client.LeaseID
+ watchedChannels sync.Map
+ keyReservationsLock sync.RWMutex
+ lockToMutexMap map[string]*v3Concurrency.Mutex
+ lockToSessionMap map[string]*v3Concurrency.Session
+ lockToMutexLock sync.Mutex
}
// NewEtcdClient returns a new client for the Etcd KV store
@@ -114,13 +114,13 @@
return fmt.Errorf("unexpected-type-%T", value)
}
- c.writeLock.Lock()
- defer c.writeLock.Unlock()
-
var err error
// Check if there is already a lease for this key - if there is then use it, otherwise a PUT will make
// that KV key permanent instead of automatically removing it after a lease expiration
- if leaseID, ok := c.keyReservations[key]; ok {
+ c.keyReservationsLock.RLock()
+ leaseID, ok := c.keyReservations[key]
+ c.keyReservationsLock.RUnlock()
+ if ok {
_, err = c.ectdAPI.Put(ctx, key, val, v3Client.WithLease(*leaseID))
} else {
_, err = c.ectdAPI.Put(ctx, key, val)
@@ -146,9 +146,6 @@
// wait for a response
func (c *EtcdClient) Delete(ctx context.Context, key string) error {
- c.writeLock.Lock()
- defer c.writeLock.Unlock()
-
// delete the key
if _, err := c.ectdAPI.Delete(ctx, key); err != nil {
logger.Errorw("failed-to-delete-key", log.Fields{"key": key, "error": err})
@@ -177,9 +174,9 @@
return nil, err
}
// Register the lease id
- c.writeLock.Lock()
+ c.keyReservationsLock.Lock()
c.keyReservations[key] = &resp.ID
- c.writeLock.Unlock()
+ c.keyReservationsLock.Unlock()
// Revoke lease if reservation is not successful
reservationSuccessful := false
@@ -235,8 +232,8 @@
// ReleaseAllReservations releases all key reservations previously made (using Reserve API)
func (c *EtcdClient) ReleaseAllReservations(ctx context.Context) error {
- c.writeLock.Lock()
- defer c.writeLock.Unlock()
+ c.keyReservationsLock.Lock()
+ defer c.keyReservationsLock.Unlock()
for key, leaseID := range c.keyReservations {
_, err := c.ectdAPI.Revoke(ctx, *leaseID)
@@ -255,8 +252,8 @@
logger.Debugw("Release-reservation", log.Fields{"key": key})
var ok bool
var leaseID *v3Client.LeaseID
- c.writeLock.Lock()
- defer c.writeLock.Unlock()
+ c.keyReservationsLock.Lock()
+ defer c.keyReservationsLock.Unlock()
if leaseID, ok = c.keyReservations[key]; !ok {
return nil
}
@@ -278,9 +275,11 @@
// Get the leaseid using the key
var ok bool
var leaseID *v3Client.LeaseID
- c.writeLock.Lock()
- defer c.writeLock.Unlock()
- if leaseID, ok = c.keyReservations[key]; !ok {
+ c.keyReservationsLock.RLock()
+ leaseID, ok = c.keyReservations[key]
+ c.keyReservationsLock.RUnlock()
+
+ if !ok {
return errors.New("key-not-reserved")
}
@@ -372,8 +371,6 @@
// Get the array of channels mapping
var watchedChannels []map[chan *Event]v3Client.Watcher
var ok bool
- c.writeLock.Lock()
- defer c.writeLock.Unlock()
if watchedChannels, ok = c.getChannelMaps(key); !ok {
logger.Warnw("key-has-no-watched-channels", log.Fields{"key": key})
@@ -425,8 +422,6 @@
// Close closes the KV store client
func (c *EtcdClient) Close() {
- c.writeLock.Lock()
- defer c.writeLock.Unlock()
if err := c.ectdAPI.Close(); err != nil {
logger.Errorw("error-closing-client", log.Fields{"error": err})
}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/kafka/kafka_inter_container_library.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/kafka/kafka_inter_container_library.go
index 91b2143..5dbde9c 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/kafka/kafka_inter_container_library.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/kafka/kafka_inter_container_library.go
@@ -19,6 +19,8 @@
"context"
"errors"
"fmt"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
"reflect"
"strings"
"sync"
@@ -34,7 +36,7 @@
const (
DefaultMaxRetries = 3
- DefaultRequestTimeout = 10000 // 10000 milliseconds - to handle a wider latency range
+ DefaultRequestTimeout = 60000 // 60000 milliseconds - to handle a wider latency range
)
const (
@@ -66,6 +68,7 @@
GetDefaultTopic() *Topic
DeviceDiscovered(deviceId string, deviceType string, parentId string, publisher string) error
InvokeRPC(ctx context.Context, rpc string, toTopic *Topic, replyToTopic *Topic, waitForResponse bool, key string, kvArgs ...*KVArg) (bool, *any.Any)
+ InvokeAsyncRPC(ctx context.Context, rpc string, toTopic *Topic, replyToTopic *Topic, waitForResponse bool, key string, kvArgs ...*KVArg) chan *RpcResponse
SubscribeWithRequestHandlerInterface(topic Topic, handler interface{}) error
SubscribeWithDefaultRequestHandler(topic Topic, initialOffset int64) error
UnSubscribeFromRequestHandler(topic Topic) error
@@ -246,6 +249,104 @@
return nil
}
+// InvokeAsyncRPC is used to make an RPC request asynchronously
+func (kp *interContainerProxy) InvokeAsyncRPC(ctx context.Context, rpc string, toTopic *Topic, replyToTopic *Topic,
+ waitForResponse bool, key string, kvArgs ...*KVArg) chan *RpcResponse {
+
+ logger.Debugw("InvokeAsyncRPC", log.Fields{"rpc": rpc, "key": key})
+ // If a replyToTopic is provided then we use it, otherwise just use the default toTopic. The replyToTopic is
+ // typically the device ID.
+ responseTopic := replyToTopic
+ if responseTopic == nil {
+ responseTopic = kp.GetDefaultTopic()
+ }
+
+ chnl := make(chan *RpcResponse)
+
+ go func() {
+
+ // once we're done,
+ // close the response channel
+ defer close(chnl)
+
+ var err error
+ var protoRequest *ic.InterContainerMessage
+
+ // Encode the request
+ protoRequest, err = encodeRequest(rpc, toTopic, responseTopic, key, kvArgs...)
+ if err != nil {
+ logger.Warnw("cannot-format-request", log.Fields{"rpc": rpc, "error": err})
+ chnl <- NewResponse(RpcFormattingError, err, nil)
+ return
+ }
+
+ // Subscribe for response, if needed, before sending request
+ var ch <-chan *ic.InterContainerMessage
+ if ch, err = kp.subscribeForResponse(*responseTopic, protoRequest.Header.Id); err != nil {
+ logger.Errorw("failed-to-subscribe-for-response", log.Fields{"error": err, "toTopic": toTopic.Name})
+ chnl <- NewResponse(RpcTransportError, err, nil)
+ return
+ }
+
+ // Send request - if the topic is formatted with a device Id then we will send the request using a
+ // specific key, hence ensuring a single partition is used to publish the request. This ensures that the
+ // subscriber on that topic will receive the request in the order it was sent. The key used is the deviceId.
+ logger.Debugw("sending-msg", log.Fields{"rpc": rpc, "toTopic": toTopic, "replyTopic": responseTopic, "key": key, "xId": protoRequest.Header.Id})
+
+ // if the message is not sent on kafka publish an event an close the channel
+ if err = kp.kafkaClient.Send(protoRequest, toTopic, key); err != nil {
+ chnl <- NewResponse(RpcTransportError, err, nil)
+ return
+ }
+
+ // if the client is not waiting for a response send the ack and close the channel
+ chnl <- NewResponse(RpcSent, nil, nil)
+ if !waitForResponse {
+ return
+ }
+
+ defer func() {
+ // Remove the subscription for a response on return
+ if err := kp.unSubscribeForResponse(protoRequest.Header.Id); err != nil {
+ logger.Warnw("invoke-async-rpc-unsubscriber-for-response-failed", log.Fields{"err": err})
+ }
+ }()
+
+ // Wait for response as well as timeout or cancellation
+ select {
+ case msg, ok := <-ch:
+ if !ok {
+ logger.Warnw("channel-closed", log.Fields{"rpc": rpc, "replyTopic": replyToTopic.Name})
+ chnl <- NewResponse(RpcTransportError, status.Error(codes.Aborted, "channel closed"), nil)
+ }
+ logger.Debugw("received-response", log.Fields{"rpc": rpc, "msgHeader": msg.Header})
+ if responseBody, err := decodeResponse(msg); err != nil {
+ chnl <- NewResponse(RpcReply, err, nil)
+ } else {
+ if responseBody.Success {
+ chnl <- NewResponse(RpcReply, nil, responseBody.Result)
+ } else {
+ // response body contains an error
+ unpackErr := &ic.Error{}
+ if err := ptypes.UnmarshalAny(responseBody.Result, unpackErr); err != nil {
+ chnl <- NewResponse(RpcReply, err, nil)
+ } else {
+ chnl <- NewResponse(RpcReply, status.Error(codes.Internal, unpackErr.Reason), nil)
+ }
+ }
+ }
+ case <-ctx.Done():
+ logger.Errorw("context-cancelled", log.Fields{"rpc": rpc, "ctx": ctx.Err()})
+ err := status.Error(codes.DeadlineExceeded, ctx.Err().Error())
+ chnl <- NewResponse(RpcTimeout, err, nil)
+ case <-kp.doneCh:
+ chnl <- NewResponse(RpcSystemClosing, nil, nil)
+ logger.Warnw("received-exit-signal", log.Fields{"toTopic": toTopic.Name, "rpc": rpc})
+ }
+ }()
+ return chnl
+}
+
// InvokeRPC is used to send a request to a given topic
func (kp *interContainerProxy) InvokeRPC(ctx context.Context, rpc string, toTopic *Topic, replyToTopic *Topic,
waitForResponse bool, key string, kvArgs ...*KVArg) (bool, *any.Any) {
@@ -799,7 +900,8 @@
// Create a specific channel for this consumers. We cannot use the channel from the kafkaclient as it will
// broadcast any message for this topic to all channels waiting on it.
- ch := make(chan *ic.InterContainerMessage)
+ // Set channel size to 1 to prevent deadlock, see VOL-2708
+ ch := make(chan *ic.InterContainerMessage, 1)
kp.addToTransactionIdToChannelMap(trnsId, &topic, ch)
return ch, nil
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/kafka/utils.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/kafka/utils.go
index 0cb9535..bdc615f 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/kafka/utils.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/kafka/utils.go
@@ -15,7 +15,10 @@
*/
package kafka
-import "strings"
+import (
+ "github.com/golang/protobuf/ptypes/any"
+ "strings"
+)
const (
TopicSeparator = "_"
@@ -36,6 +39,31 @@
Value interface{}
}
+type RpcMType int
+
+const (
+ RpcFormattingError RpcMType = iota
+ RpcSent
+ RpcReply
+ RpcTimeout
+ RpcTransportError
+ RpcSystemClosing
+)
+
+type RpcResponse struct {
+ MType RpcMType
+ Err error
+ Reply *any.Any
+}
+
+func NewResponse(messageType RpcMType, err error, body *any.Any) *RpcResponse {
+ return &RpcResponse{
+ MType: messageType,
+ Err: err,
+ Reply: body,
+ }
+}
+
// TODO: Remove and provide better may to get the device id
// GetDeviceIdFromTopic extract the deviceId from the topic name. The topic name is formatted either as:
// <any string> or <any string>_<deviceId>. The device Id is 24 characters long.
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/log/log.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/log/log.go
index 3ebdd3a..47fa3fb 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/log/log.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/log/log.go
@@ -221,6 +221,7 @@
LevelKey: "level",
MessageKey: "msg",
TimeKey: "ts",
+ CallerKey: "caller",
StacktraceKey: "stacktrace",
LineEnding: zc.DefaultLineEnding,
EncodeLevel: zc.LowercaseLevelEncoder,
@@ -237,7 +238,7 @@
// Build a custom config using zap
cfg = getDefaultConfig(outputType, level, defaultFields)
- l, err := cfg.Build()
+ l, err := cfg.Build(zp.AddCallerSkip(1))
if err != nil {
return nil, err
}
@@ -282,7 +283,7 @@
cfgs[pkgName] = getDefaultConfig(outputType, level, defaultFields)
- l, err := cfgs[pkgName].Build()
+ l, err := cfgs[pkgName].Build(zp.AddCallerSkip(1))
if err != nil {
return nil, err
}
@@ -304,16 +305,14 @@
}
cfg.InitialFields[k] = v
}
- l, err := cfg.Build()
+ l, err := cfg.Build(zp.AddCallerSkip(1))
if err != nil {
return err
}
- loggers[pkgName] = &logger{
- log: l.Sugar(),
- parent: l,
- packageName: pkgName,
- }
+ // Update the existing zap logger instance
+ loggers[pkgName].log = l.Sugar()
+ loggers[pkgName].parent = l
}
return nil
}
@@ -329,19 +328,16 @@
return keys
}
-// UpdateLogger deletes the logger associated with a caller's package and creates a new logger with the
-// defaultFields. If a calling package is holding on to a Logger reference obtained from AddPackage invocation, then
-// that package needs to invoke UpdateLogger if it needs to make changes to the default fields and obtain a new logger
-// reference
-func UpdateLogger(defaultFields Fields) (Logger, error) {
+// UpdateLogger updates the logger associated with a caller's package with supplied defaultFields
+func UpdateLogger(defaultFields Fields) error {
pkgName, _, _, _ := getCallerInfo()
if _, exist := loggers[pkgName]; !exist {
- return nil, fmt.Errorf("package-%s-not-registered", pkgName)
+ return fmt.Errorf("package-%s-not-registered", pkgName)
}
// Build a new logger
if _, exist := cfgs[pkgName]; !exist {
- return nil, fmt.Errorf("config-%s-not-registered", pkgName)
+ return fmt.Errorf("config-%s-not-registered", pkgName)
}
cfg := cfgs[pkgName]
@@ -351,18 +347,16 @@
}
cfg.InitialFields[k] = v
}
- l, err := cfg.Build()
+ l, err := cfg.Build(zp.AddCallerSkip(1))
if err != nil {
- return nil, err
+ return err
}
- // Set the logger
- loggers[pkgName] = &logger{
- log: l.Sugar(),
- parent: l,
- packageName: pkgName,
- }
- return loggers[pkgName], nil
+ // Update the existing zap logger instance
+ loggers[pkgName].log = l.Sugar()
+ loggers[pkgName].parent = l
+
+ return nil
}
func setLevel(cfg zp.Config, level LogLevel) {
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/mocks/kafka_inter_container_proxy.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/mocks/kafka_inter_container_proxy.go
index 405fbe7..9879830 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/mocks/kafka_inter_container_proxy.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/mocks/kafka_inter_container_proxy.go
@@ -42,6 +42,13 @@
Response proto.Message
}
+type InvokeAsyncRpcSpy struct {
+ CallCount int
+ Calls map[int]InvokeRpcArgs
+ Timeout bool
+ Response proto.Message
+}
+
type MockKafkaICProxy struct {
InvokeRpcSpy InvokeRpcSpy
}
@@ -58,6 +65,29 @@
return nil
}
func (s *MockKafkaICProxy) Stop() {}
+
+func (s *MockKafkaICProxy) InvokeAsyncRPC(ctx context.Context, rpc string, toTopic *kafka.Topic, replyToTopic *kafka.Topic,
+ waitForResponse bool, key string, kvArgs ...*kafka.KVArg) chan *kafka.RpcResponse {
+
+ args := make(map[int]interface{}, 4)
+ for k, v := range kvArgs {
+ args[k] = v
+ }
+
+ s.InvokeRpcSpy.Calls[s.InvokeRpcSpy.CallCount] = InvokeRpcArgs{
+ Rpc: rpc,
+ ToTopic: toTopic,
+ ReplyToTopic: replyToTopic,
+ WaitForResponse: waitForResponse,
+ Key: key,
+ KvArgs: args,
+ }
+
+ chnl := make(chan *kafka.RpcResponse)
+
+ return chnl
+}
+
func (s *MockKafkaICProxy) InvokeRPC(ctx context.Context, rpc string, toTopic *kafka.Topic, replyToTopic *kafka.Topic, waitForResponse bool, key string, kvArgs ...*kafka.KVArg) (bool, *any.Any) {
s.InvokeRpcSpy.CallCount++
diff --git a/vendor/modules.txt b/vendor/modules.txt
index d5d84a2..3c4fa8c 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -96,7 +96,7 @@
github.com/modern-go/concurrent
# github.com/modern-go/reflect2 v1.0.1
github.com/modern-go/reflect2
-# github.com/opencord/voltha-lib-go/v3 v3.0.15
+# github.com/opencord/voltha-lib-go/v3 v3.0.20
github.com/opencord/voltha-lib-go/v3/pkg/adapters
github.com/opencord/voltha-lib-go/v3/pkg/adapters/adapterif
github.com/opencord/voltha-lib-go/v3/pkg/adapters/common