icsversion/internal/odusliceassurance/messageHandler.go

   1 // -
   2 //   ========================LICENSE_START=================================
   3 //   O-RAN-SC
   4 //   %%
   5 //   Copyright (C) 2022: Nordix Foundation
   6 //   %%
   7 //   Licensed under the Apache License, Version 2.0 (the "License");
   8 //   you may not use this file except in compliance with the License.
   9 //   You may obtain a copy of the License at
  10 //
  11 //        http://www.apache.org/licenses/LICENSE-2.0
  12 //
  13 //   Unless required by applicable law or agreed to in writing, software
  14 //   distributed under the License is distributed on an "AS IS" BASIS,
  15 //   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 //   See the License for the specific language governing permissions and
  17 //   limitations under the License.
  18 //   ========================LICENSE_END===================================
  19 //
  20
  21 package sliceassurance
  22
  23 import (
  24         "encoding/json"
  25         "io"
  26         "io/ioutil"
  27
  28         log "github.com/sirupsen/logrus"
  29         "oransc.org/usecase/oduclosedloop/icsversion/internal/structures"
  30         "oransc.org/usecase/oduclosedloop/icsversion/messages"
  31 )
  32
  33 type MessageHandler struct {
  34         data *structures.SliceAssuranceMeas
  35 }
  36
  37 func NewMessageHandler(data *structures.SliceAssuranceMeas) *MessageHandler {
  38         return &MessageHandler{
  39                 data: data,
  40         }
  41 }
  42
  43 func (handler MessageHandler) ProcessMessage(body io.ReadCloser) {
  44         log.Debug("Process messages from Dmaap mediator")
  45
  46         if messages := getVesMessages(body); messages != nil {
  47                 stdMessages := getStdMessages(messages)
  48
  49                 for _, message := range stdMessages {
  50                         for _, meas := range message.GetMeasurements() {
  51                                 log.Infof("Create sliceMetric and check if metric exist and update existing one or create new one measurement:  %+v\n", meas)
  52                                 //Create sliceMetric and check if metric exist and update existing one or create new one
  53                                 if _, err := handler.data.AddOrUpdateMetric(meas); err != nil {
  54                                         log.Error("Metric could not be added ", err)
  55                                 }
  56                         }
  57                 }
  58         }
  59
  60 }
  61
  62 func getVesMessages(r io.ReadCloser) *[]string {
  63         var messages []string
  64         body, err := ioutil.ReadAll(r)
  65         if err != nil {
  66                 log.Warn(err)
  67                 return nil
  68         }
  69         err = json.Unmarshal(body, &messages)
  70         if err != nil {
  71                 log.Warn(err)
  72                 return nil
  73         }
  74         return &messages
  75 }
  76
  77 func getStdMessages(messageStrings *[]string) []messages.StdDefinedMessage {
  78         stdMessages := make([]messages.StdDefinedMessage, 0, len(*messageStrings))
  79         for _, msgString := range *messageStrings {
  80                 var message messages.StdDefinedMessage
  81                 if err := json.Unmarshal([]byte(msgString), &message); err == nil {
  82                         stdMessages = append(stdMessages, message)
  83                 } else {
  84                         log.Warn(err)
  85                 }
  86         }
  87         return stdMessages
  88 }