VOL-2017 voltha-lib moved from voltha-go;
release version 2.2.1
Based on voltha-go commit 5259f8e52b3e3f5c7ad422a4b0e506e1d07f6b36
Change-Id: I8bbecdf456e420714a4016120eafc0d237c80565
diff --git a/pkg/adapters/common/events_proxy.go b/pkg/adapters/common/events_proxy.go
new file mode 100644
index 0000000..59e7a00
--- /dev/null
+++ b/pkg/adapters/common/events_proxy.go
@@ -0,0 +1,114 @@
+/*
+ * Copyright 2018-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 common
+
+import (
+ "errors"
+ "fmt"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/opencord/voltha-lib-go/pkg/adapters/adapterif"
+ "github.com/opencord/voltha-lib-go/pkg/common/log"
+ "github.com/opencord/voltha-lib-go/pkg/kafka"
+ "github.com/opencord/voltha-protos/go/voltha"
+)
+
+type EventProxy struct {
+ kafkaClient kafka.Client
+ eventTopic kafka.Topic
+}
+
+func NewEventProxy(opts ...EventProxyOption) *EventProxy {
+ var proxy EventProxy
+ for _, option := range opts {
+ option(&proxy)
+ }
+ return &proxy
+}
+
+type EventProxyOption func(*EventProxy)
+
+func MsgClient(client kafka.Client) EventProxyOption {
+ return func(args *EventProxy) {
+ args.kafkaClient = client
+ }
+}
+
+func MsgTopic(topic kafka.Topic) EventProxyOption {
+ return func(args *EventProxy) {
+ args.eventTopic = topic
+ }
+}
+
+func (ep *EventProxy) formatId(eventName string) string {
+ return fmt.Sprintf("Voltha.openolt.%s.%s", eventName, strconv.FormatInt(time.Now().UnixNano(), 10))
+}
+
+func (ep *EventProxy) getEventHeader(eventName string, category adapterif.EventCategory, subCategory adapterif.EventSubCategory, eventType adapterif.EventType, raisedTs int64) *voltha.EventHeader {
+ var header voltha.EventHeader
+ if strings.Contains(eventName, "_") {
+ eventName = strings.Join(strings.Split(eventName, "_")[:len(strings.Split(eventName, "_"))-2], "_")
+ } else {
+ eventName = "UNKNOWN_EVENT"
+ }
+ /* Populating event header */
+ header.Id = ep.formatId(eventName)
+ header.Category = category
+ header.SubCategory = subCategory
+ header.Type = eventType
+ header.TypeVersion = adapterif.EventTypeVersion
+ header.RaisedTs = float32(raisedTs)
+ header.ReportedTs = float32(time.Now().UnixNano())
+ return &header
+}
+
+/* Send out device events*/
+func (ep *EventProxy) SendDeviceEvent(deviceEvent *voltha.DeviceEvent, category adapterif.EventCategory, subCategory adapterif.EventSubCategory, raisedTs int64) error {
+ if deviceEvent == nil {
+ log.Error("Recieved empty device event")
+ return errors.New("Device event nil")
+ }
+ var event voltha.Event
+ var de voltha.Event_DeviceEvent
+ de.DeviceEvent = deviceEvent
+ event.Header = ep.getEventHeader(deviceEvent.DeviceEventName, category, subCategory, voltha.EventType_DEVICE_EVENT, raisedTs)
+ event.EventType = &de
+ if err := ep.sendEvent(&event); err != nil {
+ log.Errorw("Failed to send device event to KAFKA bus", log.Fields{"device-event": deviceEvent})
+ return err
+ }
+ log.Infow("Successfully sent device event KAFKA", log.Fields{"Id": event.Header.Id, "Category": event.Header.Category,
+ "SubCategory": event.Header.SubCategory, "Type": event.Header.Type, "TypeVersion": event.Header.TypeVersion,
+ "ReportedTs": event.Header.ReportedTs, "ResourceId": deviceEvent.ResourceId, "Context": deviceEvent.Context,
+ "DeviceEventName": deviceEvent.DeviceEventName})
+
+ return nil
+
+}
+
+/* TODO: Send out KPI events*/
+
+func (ep *EventProxy) sendEvent(event *voltha.Event) error {
+ if err := ep.kafkaClient.Send(event, &ep.eventTopic); err != nil {
+ return err
+ }
+ log.Debugw("Sent event to kafka", log.Fields{"event": event})
+
+ return nil
+}