khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 1 | /* |
| 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 | */ |
khenaidoo | b6238b3 | 2020-04-07 12:07:36 -0400 | [diff] [blame] | 16 | package kafka |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 17 | |
| 18 | import ( |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 19 | "context" |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 20 | "fmt" |
serkant.uluderya | b38671c | 2019-11-01 09:35:38 -0700 | [diff] [blame] | 21 | "sync" |
Scott Baker | 84a55ce | 2020-04-17 10:11:30 -0700 | [diff] [blame] | 22 | "time" |
serkant.uluderya | b38671c | 2019-11-01 09:35:38 -0700 | [diff] [blame] | 23 | |
Himani Chawla | 6c38e57 | 2021-03-23 19:44:25 +0530 | [diff] [blame] | 24 | "github.com/golang/protobuf/proto" |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 25 | "github.com/opencord/voltha-lib-go/v7/pkg/kafka" |
| 26 | "github.com/opencord/voltha-lib-go/v7/pkg/log" |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 27 | "google.golang.org/grpc/codes" |
| 28 | "google.golang.org/grpc/status" |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 29 | ) |
| 30 | |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 31 | const ( |
| 32 | maxConcurrentMessage = 100 |
| 33 | ) |
| 34 | |
Kent Hagerman | ccfa213 | 2019-12-17 13:29:34 -0500 | [diff] [blame] | 35 | // static check to ensure KafkaClient implements kafka.Client |
| 36 | var _ kafka.Client = &KafkaClient{} |
| 37 | |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 38 | type KafkaClient struct { |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 39 | topicsChannelMap map[string][]chan proto.Message |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 40 | lock sync.RWMutex |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 41 | alive bool |
| 42 | livenessMutex sync.Mutex |
| 43 | liveness chan bool |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 44 | } |
| 45 | |
| 46 | func NewKafkaClient() *KafkaClient { |
| 47 | return &KafkaClient{ |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 48 | topicsChannelMap: make(map[string][]chan proto.Message), |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 49 | lock: sync.RWMutex{}, |
| 50 | } |
| 51 | } |
| 52 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 53 | func (kc *KafkaClient) Start(ctx context.Context) error { |
| 54 | logger.Debug(ctx, "kafka-client-started") |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 55 | return nil |
| 56 | } |
| 57 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 58 | func (kc *KafkaClient) Stop(ctx context.Context) { |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 59 | kc.lock.Lock() |
| 60 | defer kc.lock.Unlock() |
| 61 | for topic, chnls := range kc.topicsChannelMap { |
| 62 | for _, c := range chnls { |
| 63 | close(c) |
| 64 | } |
| 65 | delete(kc.topicsChannelMap, topic) |
| 66 | } |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 67 | logger.Debug(ctx, "kafka-client-stopped") |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 68 | } |
| 69 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 70 | func (kc *KafkaClient) CreateTopic(ctx context.Context, topic *kafka.Topic, numPartition int, repFactor int) error { |
| 71 | logger.Debugw(ctx, "CreatingTopic", log.Fields{"topic": topic.Name, "numPartition": numPartition, "replicationFactor": repFactor}) |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 72 | kc.lock.Lock() |
| 73 | defer kc.lock.Unlock() |
| 74 | if _, ok := kc.topicsChannelMap[topic.Name]; ok { |
| 75 | return fmt.Errorf("Topic %s already exist", topic.Name) |
| 76 | } |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 77 | ch := make(chan proto.Message) |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 78 | kc.topicsChannelMap[topic.Name] = append(kc.topicsChannelMap[topic.Name], ch) |
| 79 | return nil |
| 80 | } |
| 81 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 82 | func (kc *KafkaClient) DeleteTopic(ctx context.Context, topic *kafka.Topic) error { |
| 83 | logger.Debugw(ctx, "DeleteTopic", log.Fields{"topic": topic.Name}) |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 84 | kc.lock.Lock() |
| 85 | defer kc.lock.Unlock() |
| 86 | delete(kc.topicsChannelMap, topic.Name) |
| 87 | return nil |
| 88 | } |
| 89 | |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 90 | func (kc *KafkaClient) Subscribe(ctx context.Context, topic *kafka.Topic, kvArgs ...*kafka.KVArg) (<-chan proto.Message, error) { |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 91 | logger.Debugw(ctx, "Subscribe", log.Fields{"topic": topic.Name, "args": kvArgs}) |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 92 | kc.lock.Lock() |
| 93 | defer kc.lock.Unlock() |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 94 | ch := make(chan proto.Message, maxConcurrentMessage) |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 95 | kc.topicsChannelMap[topic.Name] = append(kc.topicsChannelMap[topic.Name], ch) |
| 96 | return ch, nil |
| 97 | } |
| 98 | |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 99 | func removeChannel(s []chan proto.Message, i int) []chan proto.Message { |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 100 | s[i] = s[len(s)-1] |
| 101 | return s[:len(s)-1] |
| 102 | } |
| 103 | |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 104 | func (kc *KafkaClient) UnSubscribe(ctx context.Context, topic *kafka.Topic, ch <-chan proto.Message) error { |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 105 | logger.Debugw(ctx, "UnSubscribe", log.Fields{"topic": topic.Name}) |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 106 | kc.lock.Lock() |
| 107 | defer kc.lock.Unlock() |
| 108 | if chnls, ok := kc.topicsChannelMap[topic.Name]; ok { |
| 109 | idx := -1 |
| 110 | for i, c := range chnls { |
| 111 | if c == ch { |
| 112 | close(c) |
| 113 | idx = i |
| 114 | } |
| 115 | } |
| 116 | if idx >= 0 { |
| 117 | kc.topicsChannelMap[topic.Name] = removeChannel(kc.topicsChannelMap[topic.Name], idx) |
| 118 | } |
| 119 | } |
| 120 | return nil |
| 121 | } |
| 122 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 123 | func (kc *KafkaClient) SubscribeForMetadata(ctx context.Context, _ func(fromTopic string, timestamp time.Time)) { |
| 124 | logger.Debug(ctx, "SubscribeForMetadata - unimplemented") |
Kent Hagerman | ccfa213 | 2019-12-17 13:29:34 -0500 | [diff] [blame] | 125 | } |
| 126 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 127 | func (kc *KafkaClient) Send(ctx context.Context, msg interface{}, topic *kafka.Topic, keys ...string) error { |
Himani Chawla | 6c38e57 | 2021-03-23 19:44:25 +0530 | [diff] [blame] | 128 | // Assert message is a proto message |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 129 | protoMsg, ok := msg.(proto.Message) |
| 130 | if !ok { |
Himani Chawla | 6c38e57 | 2021-03-23 19:44:25 +0530 | [diff] [blame] | 131 | logger.Warnw(ctx, "message-not-a-proto-message", log.Fields{"msg": msg}) |
| 132 | return status.Error(codes.InvalidArgument, "msg-not-a-proto-msg") |
| 133 | } |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 134 | kc.lock.RLock() |
| 135 | defer kc.lock.RUnlock() |
| 136 | for _, ch := range kc.topicsChannelMap[topic.Name] { |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 137 | select { |
| 138 | case ch <- protoMsg: |
| 139 | logger.Debugw(ctx, "publishing", log.Fields{"toTopic": topic.Name, "msg": protoMsg}) |
| 140 | default: |
| 141 | logger.Debugw(ctx, "ignoring-event-channel-busy", log.Fields{"toTopic": topic.Name, "msg": protoMsg}) |
| 142 | } |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 143 | } |
| 144 | return nil |
| 145 | } |
| 146 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 147 | func (kc *KafkaClient) SendLiveness(ctx context.Context) error { |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 148 | kc.livenessMutex.Lock() |
| 149 | defer kc.livenessMutex.Unlock() |
| 150 | if kc.liveness != nil { |
| 151 | kc.liveness <- true // I am a mock |
| 152 | } |
| 153 | return nil |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 154 | } |
| 155 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 156 | func (kc *KafkaClient) EnableLivenessChannel(ctx context.Context, enable bool) chan bool { |
khenaidoo | 2672188 | 2021-08-11 17:42:52 -0400 | [diff] [blame] | 157 | logger.Infow(ctx, "kafka-enable-liveness-channel", log.Fields{"enable": enable}) |
| 158 | if enable { |
| 159 | kc.livenessMutex.Lock() |
| 160 | defer kc.livenessMutex.Unlock() |
| 161 | if kc.liveness == nil { |
| 162 | logger.Info(ctx, "kafka-create-liveness-channel") |
| 163 | kc.liveness = make(chan bool, 10) |
| 164 | // post intial state to the channel |
| 165 | kc.liveness <- kc.alive |
| 166 | } |
| 167 | } else { |
| 168 | panic("Turning off liveness reporting is not supported") |
| 169 | } |
| 170 | return kc.liveness |
khenaidoo | 59ce9dd | 2019-11-11 13:05:32 -0500 | [diff] [blame] | 171 | } |
Scott Baker | 0fef698 | 2019-12-12 09:49:42 -0800 | [diff] [blame] | 172 | |
Neha Sharma | 94f16a9 | 2020-06-26 04:17:55 +0000 | [diff] [blame] | 173 | func (kc *KafkaClient) EnableHealthinessChannel(ctx context.Context, enable bool) chan bool { |
| 174 | logger.Debug(ctx, "EnableHealthinessChannel - unimplemented") |
Scott Baker | 0fef698 | 2019-12-12 09:49:42 -0800 | [diff] [blame] | 175 | return nil |
| 176 | } |