pkg/control/control.go

   1 /*
   2 ==================================================================================
   3   Copyright (c) 2019 AT&T Intellectual Property.
   4   Copyright (c) 2019 Nokia
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10        http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17 ==================================================================================
  18 */
  19
  20 package control
  21
  22 /*
  23 #include <rmr/RIC_message_types.h>
  24
  25 #cgo CFLAGS: -I../
  26 #cgo LDFLAGS: -lrmr_nng -lnng
  27 */
  28 import "C"
  29
  30 import (
  31         "errors"
  32         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  33         "github.com/spf13/viper"
  34         "math/rand"
  35         "strconv"
  36         "time"
  37 )
  38
  39 type Control struct {
  40         e2ap     *E2ap
  41         registry *Registry
  42 }
  43
  44 var SEEDSN uint16
  45
  46 func init() {
  47         viper.AutomaticEnv()
  48         viper.SetEnvPrefix("submgr")
  49         viper.AllowEmptyEnv(true)
  50         SEEDSN = uint16(viper.GetInt("seed_sn"))
  51         if SEEDSN == 0 {
  52                 rand.Seed(time.Now().UnixNano())
  53                 SEEDSN = uint16(rand.Intn(65535))
  54         }
  55         if SEEDSN > 65535 {
  56                 SEEDSN = 0
  57         }
  58         xapp.Logger.Info("SUBMGR: Initial Sequence Number: %v", SEEDSN)
  59 }
  60
  61 func NewControl() Control {
  62         registry := new(Registry)
  63         registry.Initialize(SEEDSN)
  64         return Control{new(E2ap), registry}
  65 }
  66
  67 func (c *Control) Run() {
  68         xapp.Run(c)
  69 }
  70
  71 func (c *Control) Consume(mtype, sub_id int, len int, payload []byte) (err error) {
  72         switch mtype {
  73         case C.RIC_SUB_REQ:
  74                 err = c.handleSubscriptionRequest(&RmrDatagram{mtype, uint16(sub_id), payload})
  75         case C.RIC_SUB_RESP:
  76                 err = c.handleSubscriptionResponse(&RmrDatagram{mtype, uint16(sub_id), payload})
  77         default:
  78                 err = errors.New("Message Type " + strconv.Itoa(mtype) + " is discarded")
  79         }
  80         return
  81 }
  82
  83 func (c *Control) rmrSend(datagram *RmrDatagram) (err error) {
  84         if !xapp.Rmr.Send(datagram.MessageType, int(datagram.SubscriptionId), len(datagram.Payload), datagram.Payload) {
  85                 err = errors.New("rmr.Send() failed")
  86         }
  87         return
  88 }
  89
  90 func (c *Control) handleSubscriptionRequest(datagram *RmrDatagram) (err error) {
  91         payload_seq_num, err := c.e2ap.GetSubscriptionRequestSequenceNumber(datagram.Payload)
  92         if err != nil {
  93                 err = errors.New("Unable to get Subscription Sequence Number from Payload due to: " + err.Error())
  94                 return
  95         }
  96         xapp.Logger.Info("Subscription Request Received. RMR SUBSCRIPTION_ID: %v | PAYLOAD SEQUENCE_NUMBER: %v", datagram.SubscriptionId, payload_seq_num)
  97         new_sub_id := c.registry.ReserveSequenceNumber()
  98         payload, err := c.e2ap.SetSubscriptionRequestSequenceNumber(datagram.Payload, new_sub_id)
  99         if err != nil {
 100                 err = errors.New("Unable to set Subscription Sequence Number in Payload due to: " + err.Error())
 101                 return
 102         }
 103         xapp.Logger.Info("Generated ID: %v. Forwarding to E2 Termination...", int(new_sub_id))
 104         c.rmrSend(&RmrDatagram{C.RIC_SUB_REQ, new_sub_id, payload})
 105         return
 106 }
 107
 108 func (c *Control) handleSubscriptionResponse(datagram *RmrDatagram) (err error) {
 109         payload_seq_num, err := c.e2ap.GetSubscriptionResponseSequenceNumber(datagram.Payload)
 110         if err != nil {
 111                 err = errors.New("Unable to get Subscription Sequence Number from Payload due to: " + err.Error())
 112                 return
 113         }
 114         xapp.Logger.Info("Subscription Response Received. RMR SUBSCRIPTION_ID: %v | PAYLOAD SEQUENCE_NUMBER: %v", datagram.SubscriptionId, payload_seq_num)
 115         if !c.registry.IsValidSequenceNumber(payload_seq_num) {
 116                 err = errors.New("Unknown Subscription ID: " + strconv.Itoa(int(payload_seq_num)) + " in Subscritpion Response. Message discarded.")
 117                 return
 118         }
 119         c.registry.setSubscriptionToConfirmed(payload_seq_num)
 120         xapp.Logger.Info("Subscription Response Registered. Forwarding to Requestor...")
 121         c.rmrSend(&RmrDatagram{C.RIC_SUB_RESP, payload_seq_num, datagram.Payload})
 122         return
 123 }