blob: 97bb1357318e412910646e33fca226cc0bf286a6 [file] [log] [blame]
khenaidoo59ce9dd2019-11-11 13:05:32 -05001/*
2 * Copyright 2019-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 */
khenaidoob6238b32020-04-07 12:07:36 -040016package kafka
khenaidoo59ce9dd2019-11-11 13:05:32 -050017
18import (
Neha Sharma94f16a92020-06-26 04:17:55 +000019 "context"
khenaidoo59ce9dd2019-11-11 13:05:32 -050020 "fmt"
Himani Chawla6c38e572021-03-23 19:44:25 +053021 "github.com/golang/protobuf/ptypes"
serkant.uluderyab38671c2019-11-01 09:35:38 -070022 "sync"
Scott Baker84a55ce2020-04-17 10:11:30 -070023 "time"
serkant.uluderyab38671c2019-11-01 09:35:38 -070024
Himani Chawla6c38e572021-03-23 19:44:25 +053025 "github.com/golang/protobuf/proto"
Girish Gowdra89c985b2020-10-14 15:02:09 -070026 "github.com/opencord/voltha-lib-go/v4/pkg/kafka"
27 "github.com/opencord/voltha-lib-go/v4/pkg/log"
28 ic "github.com/opencord/voltha-protos/v4/go/inter_container"
Himani Chawla6c38e572021-03-23 19:44:25 +053029 "github.com/opencord/voltha-protos/v4/go/voltha"
khenaidoo59ce9dd2019-11-11 13:05:32 -050030 "google.golang.org/grpc/codes"
31 "google.golang.org/grpc/status"
khenaidoo59ce9dd2019-11-11 13:05:32 -050032)
33
Kent Hagermanccfa2132019-12-17 13:29:34 -050034// static check to ensure KafkaClient implements kafka.Client
35var _ kafka.Client = &KafkaClient{}
36
khenaidoo59ce9dd2019-11-11 13:05:32 -050037type KafkaClient struct {
38 topicsChannelMap map[string][]chan *ic.InterContainerMessage
39 lock sync.RWMutex
40}
41
42func NewKafkaClient() *KafkaClient {
43 return &KafkaClient{
44 topicsChannelMap: make(map[string][]chan *ic.InterContainerMessage),
45 lock: sync.RWMutex{},
46 }
47}
48
Neha Sharma94f16a92020-06-26 04:17:55 +000049func (kc *KafkaClient) Start(ctx context.Context) error {
50 logger.Debug(ctx, "kafka-client-started")
khenaidoo59ce9dd2019-11-11 13:05:32 -050051 return nil
52}
53
Neha Sharma94f16a92020-06-26 04:17:55 +000054func (kc *KafkaClient) Stop(ctx context.Context) {
khenaidoo59ce9dd2019-11-11 13:05:32 -050055 kc.lock.Lock()
56 defer kc.lock.Unlock()
57 for topic, chnls := range kc.topicsChannelMap {
58 for _, c := range chnls {
59 close(c)
60 }
61 delete(kc.topicsChannelMap, topic)
62 }
Neha Sharma94f16a92020-06-26 04:17:55 +000063 logger.Debug(ctx, "kafka-client-stopped")
khenaidoo59ce9dd2019-11-11 13:05:32 -050064}
65
Neha Sharma94f16a92020-06-26 04:17:55 +000066func (kc *KafkaClient) CreateTopic(ctx context.Context, topic *kafka.Topic, numPartition int, repFactor int) error {
67 logger.Debugw(ctx, "CreatingTopic", log.Fields{"topic": topic.Name, "numPartition": numPartition, "replicationFactor": repFactor})
khenaidoo59ce9dd2019-11-11 13:05:32 -050068 kc.lock.Lock()
69 defer kc.lock.Unlock()
70 if _, ok := kc.topicsChannelMap[topic.Name]; ok {
71 return fmt.Errorf("Topic %s already exist", topic.Name)
72 }
73 ch := make(chan *ic.InterContainerMessage)
74 kc.topicsChannelMap[topic.Name] = append(kc.topicsChannelMap[topic.Name], ch)
75 return nil
76}
77
Neha Sharma94f16a92020-06-26 04:17:55 +000078func (kc *KafkaClient) DeleteTopic(ctx context.Context, topic *kafka.Topic) error {
79 logger.Debugw(ctx, "DeleteTopic", log.Fields{"topic": topic.Name})
khenaidoo59ce9dd2019-11-11 13:05:32 -050080 kc.lock.Lock()
81 defer kc.lock.Unlock()
82 delete(kc.topicsChannelMap, topic.Name)
83 return nil
84}
85
Neha Sharma94f16a92020-06-26 04:17:55 +000086func (kc *KafkaClient) Subscribe(ctx context.Context, topic *kafka.Topic, kvArgs ...*kafka.KVArg) (<-chan *ic.InterContainerMessage, error) {
87 logger.Debugw(ctx, "Subscribe", log.Fields{"topic": topic.Name, "args": kvArgs})
khenaidoo59ce9dd2019-11-11 13:05:32 -050088 kc.lock.Lock()
89 defer kc.lock.Unlock()
90 ch := make(chan *ic.InterContainerMessage)
91 kc.topicsChannelMap[topic.Name] = append(kc.topicsChannelMap[topic.Name], ch)
92 return ch, nil
93}
94
95func removeChannel(s []chan *ic.InterContainerMessage, i int) []chan *ic.InterContainerMessage {
96 s[i] = s[len(s)-1]
97 return s[:len(s)-1]
98}
99
Neha Sharma94f16a92020-06-26 04:17:55 +0000100func (kc *KafkaClient) UnSubscribe(ctx context.Context, topic *kafka.Topic, ch <-chan *ic.InterContainerMessage) error {
101 logger.Debugw(ctx, "UnSubscribe", log.Fields{"topic": topic.Name})
khenaidoo59ce9dd2019-11-11 13:05:32 -0500102 kc.lock.Lock()
103 defer kc.lock.Unlock()
104 if chnls, ok := kc.topicsChannelMap[topic.Name]; ok {
105 idx := -1
106 for i, c := range chnls {
107 if c == ch {
108 close(c)
109 idx = i
110 }
111 }
112 if idx >= 0 {
113 kc.topicsChannelMap[topic.Name] = removeChannel(kc.topicsChannelMap[topic.Name], idx)
114 }
115 }
116 return nil
117}
118
Neha Sharma94f16a92020-06-26 04:17:55 +0000119func (kc *KafkaClient) SubscribeForMetadata(ctx context.Context, _ func(fromTopic string, timestamp time.Time)) {
120 logger.Debug(ctx, "SubscribeForMetadata - unimplemented")
Kent Hagermanccfa2132019-12-17 13:29:34 -0500121}
122
Himani Chawla6c38e572021-03-23 19:44:25 +0530123func toIntercontainerMessage(event *voltha.Event) *ic.InterContainerMessage {
124 msg := &ic.InterContainerMessage{
125 Header: &ic.Header{
126 Id: event.Header.Id,
127 Type: ic.MessageType_REQUEST,
128 Timestamp: event.Header.RaisedTs,
129 },
130 }
131 // Marshal event
132 if eventBody, err := ptypes.MarshalAny(event); err == nil {
133 msg.Body = eventBody
134 }
135 return msg
136}
137
Neha Sharma94f16a92020-06-26 04:17:55 +0000138func (kc *KafkaClient) Send(ctx context.Context, msg interface{}, topic *kafka.Topic, keys ...string) error {
Himani Chawla6c38e572021-03-23 19:44:25 +0530139 // Assert message is a proto message
140 // ascertain the value interface type is a proto.Message
141 if _, ok := msg.(proto.Message); !ok {
142 logger.Warnw(ctx, "message-not-a-proto-message", log.Fields{"msg": msg})
143 return status.Error(codes.InvalidArgument, "msg-not-a-proto-msg")
144 }
khenaidoo59ce9dd2019-11-11 13:05:32 -0500145 req, ok := msg.(*ic.InterContainerMessage)
146 if !ok {
Himani Chawla6c38e572021-03-23 19:44:25 +0530147 event, ok := msg.(*voltha.Event) //This is required as event message will be of type voltha.Event
148 if !ok {
149 return status.Error(codes.InvalidArgument, "unexpected-message-type")
150 }
151 req = toIntercontainerMessage(event)
khenaidoo59ce9dd2019-11-11 13:05:32 -0500152 }
153 if req == nil {
154 return status.Error(codes.InvalidArgument, "msg-nil")
155 }
156 kc.lock.RLock()
157 defer kc.lock.RUnlock()
158 for _, ch := range kc.topicsChannelMap[topic.Name] {
Neha Sharma94f16a92020-06-26 04:17:55 +0000159 logger.Debugw(ctx, "Publishing", log.Fields{"fromTopic": req.Header.FromTopic, "toTopic": topic.Name, "id": req.Header.Id})
khenaidoo59ce9dd2019-11-11 13:05:32 -0500160 ch <- req
161 }
162 return nil
163}
164
Neha Sharma94f16a92020-06-26 04:17:55 +0000165func (kc *KafkaClient) SendLiveness(ctx context.Context) error {
khenaidoo59ce9dd2019-11-11 13:05:32 -0500166 return status.Error(codes.Unimplemented, "SendLiveness")
167}
168
Neha Sharma94f16a92020-06-26 04:17:55 +0000169func (kc *KafkaClient) EnableLivenessChannel(ctx context.Context, enable bool) chan bool {
170 logger.Debug(ctx, "EnableLivenessChannel - unimplemented")
khenaidoo59ce9dd2019-11-11 13:05:32 -0500171 return nil
172}
Scott Baker0fef6982019-12-12 09:49:42 -0800173
Neha Sharma94f16a92020-06-26 04:17:55 +0000174func (kc *KafkaClient) EnableHealthinessChannel(ctx context.Context, enable bool) chan bool {
175 logger.Debug(ctx, "EnableHealthinessChannel - unimplemented")
Scott Baker0fef6982019-12-12 09:49:42 -0800176 return nil
177}