+++ /dev/null
-/*
-==================================================================================
- Copyright (c) 2022 Samsung
-
- 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.
-
- This source code is part of the near-RT RIC (RAN Intelligent Controller)
- platform project (RICP).
-==================================================================================
-*/
-
-package rmr
-
-import (
- "bytes"
- "encoding/json"
- "fmt"
- "io/ioutil"
- "net/http"
- "strconv"
-
- "gerrit.o-ran-sc.org/r/ric-plt/a1/pkg/a1"
- "gerrit.o-ran-sc.org/r/ric-plt/a1/pkg/models"
- "gerrit.o-ran-sc.org/r/ric-plt/a1/pkg/policy"
- "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
-)
-
-const (
- a1SourceName = "service-ricplt-a1mediator-http"
- a1PolicyRequest = 20010
- ecsServiceHost = "http://ecs-service:8083"
- ecsEiTypePath = ecsServiceHost + "/A1-EI/v1/eitypes"
- ecsEiJobPath = ecsServiceHost + "/A1-EI/v1/eijobs/"
- a1EiQueryAllResp = 20014
- a1EiCreateJobResp = 20016
- jobCreationData = `{"ei_job_id": %s.}`
-)
-
-type RmrSender struct {
- rmrclient *xapp.RMRClient
- policyManager *policy.PolicyManager
-}
-
-type IRmrSender interface {
- RmrSendToXapp(httpBodyString string, messagetype int) bool
-}
-
-func NewRMRSender(policyManager *policy.PolicyManager) IRmrSender {
- RMRclient := xapp.NewRMRClientWithParams(&xapp.RMRClientParams{
- StatDesc: "",
- RmrData: xapp.PortData{
- //TODO: Read configuration from config file
- Name: "",
- MaxSize: 65534,
- ThreadType: 0,
- LowLatency: false,
- FastAck: false,
- MaxRetryOnFailure: 1,
- Port: 4561,
- },
- })
-
- rmrsender := &RmrSender{
- rmrclient: RMRclient,
- policyManager: policyManager,
- }
-
- rmrsender.RmrRecieveStart()
- return rmrsender
-}
-
-var RICMessageTypes = map[string]int{
- "A1_POLICY_REQ": 20010,
- "A1_POLICY_RESP": 20011,
- "A1_POLICY_QUERY": 20012,
- "A1_EI_QUERY_ALL": 20013,
- "AI_EI_QUERY_ALL_RESP": 20014,
- "A1_EI_CREATE_JOB": 20015,
- "A1_EI_CREATE_JOB_RESP": 20016,
- "A1_EI_DATA_DELIVERY": 20017,
-}
-
-func (rmr *RmrSender) GetRicMessageName(id int) (s string) {
- for k, v := range RICMessageTypes {
- if id == v {
- return k
- }
- }
- return
-}
-
-func (rmr *RmrSender) RmrSendToXapp(httpBodyString string, messagetype int) bool {
-
- params := &xapp.RMRParams{}
- params.Mtype = messagetype
- params.SubId = -1
- params.Xid = ""
- params.Meid = &xapp.RMRMeid{}
- params.Src = a1SourceName
- params.PayloadLen = len([]byte(httpBodyString))
- params.Payload = []byte(httpBodyString)
- a1.Logger.Debug("MSG to XAPP: %s ", params.String())
- a1.Logger.Debug("len payload %+v", len(params.Payload))
- s := rmr.rmrclient.SendMsg(params)
- a1.Logger.Debug("rmrSendToXapp: sending: %+v", s)
- return s
-}
-
-func (rmr *RmrSender) Consume(msg *xapp.RMRParams) (err error) {
- a1.Logger.Debug("In the Consume function")
- id := rmr.GetRicMessageName(msg.Mtype)
- a1.Logger.Debug("Message received: name=%s meid=%s subId=%d txid=%s len=%d", id, msg.Meid.RanName, msg.SubId, msg.Xid, msg.PayloadLen)
-
- switch id {
-
- case "A1_POLICY_RESP":
- a1.Logger.Debug("Recived policy responose")
- payload := msg.Payload
- a1.Logger.Debug("message recieved : %s", payload)
- var result map[string]interface{}
- err := json.Unmarshal([]byte(payload), &result)
- if err != nil {
- a1.Logger.Error("Unmarshal error : %+v", err)
- return err
- }
- a1.Logger.Debug("message recieved for %d and %d with status : %s", result["policy_type_id"], result["policy_instance_id"], result["status"])
- rmr.policyManager.SetPolicyInstanceStatus(int(result["policy_type_id"].(float64)), int(result["policy_instance_id"].(float64)), result["status"].(string))
- case "A1_POLICY_QUERY":
- a1.Logger.Debug("Recived policy query")
- a1.Logger.Debug("message recieved ", msg.Payload)
- payload := msg.Payload
- var result map[string]interface{}
- json.Unmarshal([]byte(payload), &result)
- a1.Logger.Debug("message recieved : %s for %d and %d", result, result["policy_type_id"], result["policy_instance_id"])
- policytypeid := (result["policy_type_id"].(float64))
- instanceList, err1 := rmr.policyManager.GetAllPolicyInstance(int(policytypeid))
- if err1 != nil {
- a1.Logger.Error("Error : %+v", err1)
- return err1
- }
- a1.Logger.Debug("instanceList ", instanceList)
- a1.Logger.Debug("Received a query for a known policy type: %d", policytypeid)
- for _, policyinstanceid := range instanceList {
- policyinstance, err2 := rmr.policyManager.GetPolicyInstance(models.PolicyTypeID(policytypeid), policyinstanceid)
- if err2 != nil {
- a1.Logger.Error("Error : %+v", err2)
- return err2
- }
- a1.Logger.Debug("policyinstance ", policyinstance.(string))
- message := Message{}
- rmrMessage, err1 := message.PolicyMessage(strconv.FormatInt((int64(policytypeid)), 10), string(policyinstanceid), policyinstance.(string), "CREATE")
- if err1 != nil {
- a1.Logger.Error("error : %v", err1)
- return err1
- }
- a1.Logger.Debug("rmrMessage ", rmrMessage)
- isSent := rmr.RmrSendToXapp(rmrMessage, a1PolicyRequest)
- if isSent {
- a1.Logger.Debug("rmrSendToXapp : message sent")
- } else {
- a1.Logger.Error("rmrSendToXapp : message not sent")
- }
- }
-
- case "A1_EI_QUERY_ALL":
- a1.Logger.Debug("message recieved ", msg.Payload)
- resp, err := http.Get(ecsEiTypePath)
- if err != nil {
- a1.Logger.Error("Received error while fetching health info: %v", err)
- }
- if resp.StatusCode != http.StatusOK {
- a1.Logger.Warning("Received no reponse from A1-EI service1")
- }
- a1.Logger.Debug("response from A1-EI service : ", resp)
-
- defer resp.Body.Close()
- respByte, err := ioutil.ReadAll(resp.Body)
-
- if err != nil {
- a1.Logger.Debug("error in response: %+v", respByte)
- }
-
- a1.Logger.Debug("response : %+v", string(respByte))
-
- isSent := rmr.RmrSendToXapp(string(respByte), a1EiQueryAllResp)
- if isSent {
- a1.Logger.Debug("rmrSendToXapp : message sent")
- } else {
- a1.Logger.Error("rmrSendToXapp : message not sent")
- }
- case "A1_EI_CREATE_JOB":
- payload := msg.Payload
- a1.Logger.Debug("message recieved : %s", payload)
-
- var result map[string]interface{}
-
- err := json.Unmarshal([]byte(payload), &result)
- if err != nil {
- a1.Logger.Error("Unmarshal error : %+v", err)
- return err
- }
- a1.Logger.Debug("Unmarshaled message recieved : %s ", result)
-
- jobIdStr := strconv.FormatInt((int64(result["job-id"].(float64))), 10)
- jsonReq, err := json.Marshal(result)
- if err != nil {
- a1.Logger.Error("marshal error : %v", err)
- return err
- }
-
- a1.Logger.Debug("url to send to :", ecsEiJobPath+jobIdStr)
- req, err := http.NewRequest(http.MethodPut, ecsEiJobPath+jobIdStr, bytes.NewBuffer(jsonReq))
- if err != nil {
- a1.Logger.Error("http error : %v", err)
- return err
- }
-
- req.Header.Set("Content-Type", "application/json; charset=utf-8")
- client := &http.Client{}
- resp, err3 := client.Do(req)
- if err3 != nil {
- a1.Logger.Error("error:", err3)
- return err
- }
-
- defer resp.Body.Close()
-
- a1.Logger.Debug("response status : ", resp.StatusCode)
- if resp.StatusCode == 200 || resp.StatusCode == 201 {
- a1.Logger.Debug("received successful response for ei-job-id : ", jobIdStr)
- rmrData := fmt.Sprintf(jobCreationData, jobIdStr)
- a1.Logger.Debug("rmr_Data to send: ", rmrData)
-
- isSent := rmr.RmrSendToXapp(rmrData, a1EiCreateJobResp)
- if isSent {
- a1.Logger.Debug("rmrSendToXapp : message sent")
- } else {
- a1.Logger.Error("rmrSendToXapp : message not sent")
- }
- } else {
- a1.Logger.Warning("failed to create EIJOB ")
- }
-
- default:
- xapp.Logger.Error("Unknown message type '%d', discarding", msg.Mtype)
- }
-
- defer func() {
- rmr.rmrclient.Free(msg.Mbuf)
- msg.Mbuf = nil
- }()
- return
-}
-
-func (rmr *RmrSender) RmrRecieveStart() {
- a1.Logger.Debug("Inside RmrRecieveStart function ")
- go rmr.rmrclient.Start(rmr)
- a1.Logger.Debug("Reciever started")
-}
Code Review / ric-plt / a1.git / blobdiff