blob: 02fa3de92e822cac44f223e635e21f67696f9cfe [file] [log] [blame]
Scott Baker2c1c4822019-10-16 11:02:41 -07001/*
2 * Copyright 2018-present Open Networking Foundation
3
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7
8 * http://www.apache.org/licenses/LICENSE-2.0
9
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16package common
17
18import (
19 "context"
serkant.uluderyab38671c2019-11-01 09:35:38 -070020 "time"
21
Scott Baker2c1c4822019-10-16 11:02:41 -070022 "github.com/golang/protobuf/proto"
23 "github.com/golang/protobuf/ptypes"
24 "github.com/golang/protobuf/ptypes/any"
25 "github.com/google/uuid"
serkant.uluderyab38671c2019-11-01 09:35:38 -070026 "github.com/opencord/voltha-lib-go/v3/pkg/kafka"
27 "github.com/opencord/voltha-lib-go/v3/pkg/log"
28 ic "github.com/opencord/voltha-protos/v3/go/inter_container"
Scott Baker2c1c4822019-10-16 11:02:41 -070029)
30
31type AdapterProxy struct {
Matteo Scandolo2ba00d32020-01-16 17:33:03 -080032 kafkaICProxy kafka.InterContainerProxy
Scott Baker2c1c4822019-10-16 11:02:41 -070033 adapterTopic string
34 coreTopic string
35}
36
Matteo Scandolo2ba00d32020-01-16 17:33:03 -080037func NewAdapterProxy(kafkaProxy kafka.InterContainerProxy, adapterTopic string, coreTopic string) *AdapterProxy {
Scott Baker2c1c4822019-10-16 11:02:41 -070038 var proxy AdapterProxy
39 proxy.kafkaICProxy = kafkaProxy
40 proxy.adapterTopic = adapterTopic
41 proxy.coreTopic = coreTopic
khenaidoob332f9b2020-01-16 16:25:26 -050042 logger.Debugw("TOPICS", log.Fields{"core": proxy.coreTopic, "adapter": proxy.adapterTopic})
Scott Baker2c1c4822019-10-16 11:02:41 -070043 return &proxy
44}
45
46func (ap *AdapterProxy) SendInterAdapterMessage(ctx context.Context,
47 msg proto.Message,
48 msgType ic.InterAdapterMessageType_Types,
49 fromAdapter string,
50 toAdapter string,
51 toDeviceId string,
52 proxyDeviceId string,
53 messageId string) error {
khenaidoob332f9b2020-01-16 16:25:26 -050054 logger.Debugw("sending-inter-adapter-message", log.Fields{"type": msgType, "from": fromAdapter,
Scott Baker2c1c4822019-10-16 11:02:41 -070055 "to": toAdapter, "toDevice": toDeviceId, "proxyDevice": proxyDeviceId})
56
57 //Marshal the message
58 var marshalledMsg *any.Any
59 var err error
60 if marshalledMsg, err = ptypes.MarshalAny(msg); err != nil {
khenaidoob332f9b2020-01-16 16:25:26 -050061 logger.Warnw("cannot-marshal-msg", log.Fields{"error": err})
Scott Baker2c1c4822019-10-16 11:02:41 -070062 return err
63 }
64
65 //Build the inter adapter message
66 header := &ic.InterAdapterHeader{
67 Type: msgType,
68 FromTopic: fromAdapter,
69 ToTopic: toAdapter,
70 ToDeviceId: toDeviceId,
71 ProxyDeviceId: proxyDeviceId,
72 }
73 if messageId != "" {
74 header.Id = messageId
75 } else {
76 header.Id = uuid.New().String()
77 }
78 header.Timestamp = time.Now().Unix()
79 iaMsg := &ic.InterAdapterMessage{
80 Header: header,
81 Body: marshalledMsg,
82 }
83 args := make([]*kafka.KVArg, 1)
84 args[0] = &kafka.KVArg{
85 Key: "msg",
86 Value: iaMsg,
87 }
88
89 // Set up the required rpc arguments
90 topic := kafka.Topic{Name: toAdapter}
91 replyToTopic := kafka.Topic{Name: fromAdapter}
92 rpc := "process_inter_adapter_message"
93
94 success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &topic, &replyToTopic, true, proxyDeviceId, args...)
khenaidoob332f9b2020-01-16 16:25:26 -050095 logger.Debugw("inter-adapter-msg-response", log.Fields{"replyTopic": replyToTopic, "success": success})
Scott Baker2c1c4822019-10-16 11:02:41 -070096 return unPackResponse(rpc, "", success, result)
97}