Add NETCONF notification for ONU activation and Kafka client to receive events, update dependencies
Change-Id: I5f768fa8077ef7c64e00a534744ca47492344935
diff --git a/internal/clients/kafka.go b/internal/clients/kafka.go
new file mode 100644
index 0000000..9350895
--- /dev/null
+++ b/internal/clients/kafka.go
@@ -0,0 +1,155 @@
+/*
+* Copyright 2022-present Open Networking Foundation
+
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+
+* http://www.apache.org/licenses/LICENSE-2.0
+
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+ */
+
+package clients
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "github.com/Shopify/sarama"
+ "github.com/golang/protobuf/proto"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
+)
+
+const (
+ volthaEventsTopic = "voltha.events"
+ kafkaBackoffInterval = time.Second * 10
+)
+
+//Used to listen for events coming from VOLTHA
+type KafkaConsumer struct {
+ address string
+ config *sarama.Config
+ client sarama.Client
+ consumer sarama.Consumer
+ partitionConsumer sarama.PartitionConsumer
+ highwater int64
+}
+
+//Creates a sarama client with the specified address
+func NewKafkaConsumer(clusterAddress string) *KafkaConsumer {
+ c := KafkaConsumer{address: clusterAddress}
+ c.config = sarama.NewConfig()
+ c.config.ClientID = "bbf-adapter-consumer"
+ c.config.Consumer.Return.Errors = true
+ c.config.Consumer.Offsets.Initial = sarama.OffsetNewest
+ c.config.Version = sarama.V1_0_0_0
+
+ return &c
+}
+
+//Starts consuming new messages on the voltha events topic, executing the provided callback on each event
+func (c *KafkaConsumer) Start(ctx context.Context, eventCallback func(context.Context, *voltha.Event)) error {
+ var err error
+
+ for {
+ if c.client, err = sarama.NewClient([]string{c.address}, c.config); err == nil {
+ logger.Debug(ctx, "kafka-client-created")
+ break
+ } else {
+ logger.Warnw(ctx, "kafka-not-reachable", log.Fields{
+ "err": err,
+ })
+ }
+
+ //Wait a bit before trying again
+ select {
+ case <-ctx.Done():
+ return fmt.Errorf("kafka-client-creation-stopped-due-to-context-done")
+ case <-time.After(kafkaBackoffInterval):
+ continue
+ }
+ }
+
+ c.consumer, err = sarama.NewConsumerFromClient(c.client)
+ if err != nil {
+ return err
+ }
+
+ partitions, _ := c.consumer.Partitions(volthaEventsTopic)
+
+ // TODO: Add support for multiple partitions
+ if len(partitions) > 1 {
+ logger.Warnw(ctx, "only-listening-one-partition", log.Fields{
+ "topic": volthaEventsTopic,
+ "partitionsNum": len(partitions),
+ })
+ }
+
+ hw, err := c.client.GetOffset(volthaEventsTopic, partitions[0], sarama.OffsetNewest)
+ if err != nil {
+ return fmt.Errorf("cannot-get-highwater: %v", err)
+ }
+ c.highwater = hw
+
+ c.partitionConsumer, err = c.consumer.ConsumePartition(volthaEventsTopic, partitions[0], sarama.OffsetOldest)
+ if nil != err {
+ return fmt.Errorf("Error in consume(): Topic %v Partitions: %v", volthaEventsTopic, partitions)
+ }
+
+ //Start consuming the event topic in a goroutine
+ logger.Debugw(ctx, "start-consuming-kafka-topic", log.Fields{"topic": volthaEventsTopic})
+ go func(topic string, pConsumer sarama.PartitionConsumer) {
+ for {
+ select {
+ case <-ctx.Done():
+ logger.Info(ctx, "stopped-listening-for-events-due-to-context-done")
+ return
+ case err := <-pConsumer.Errors():
+ logger.Errorw(ctx, "kafka-consumer-error", log.Fields{
+ "err": err.Error(),
+ "topic": err.Topic,
+ "partition": err.Partition,
+ })
+ case msg := <-pConsumer.Messages():
+ if msg.Offset <= c.highwater {
+ continue
+ }
+
+ //Unmarshal the content of the message to a voltha Event protobuf message
+ event := &voltha.Event{}
+ if err := proto.Unmarshal(msg.Value, event); err != nil {
+ logger.Errorw(ctx, "error-unmarshalling-kafka-event", log.Fields{"err": err})
+ continue
+ }
+
+ eventCallback(ctx, event)
+ }
+ }
+ }(volthaEventsTopic, c.partitionConsumer)
+
+ return nil
+}
+
+//Closes the sarama client and all consumers
+func (c *KafkaConsumer) Stop() error {
+ if err := c.partitionConsumer.Close(); err != nil {
+ return err
+ }
+
+ if err := c.consumer.Close(); err != nil {
+ return err
+ }
+
+ if err := c.client.Close(); err != nil {
+ return err
+ }
+
+ return nil
+}