X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fcontrol.go;h=f6cd771e7e20a5e58af049cd5ad5dbbfec76dc7f;hb=refs%2Fchanges%2F92%2F992%2F2;hp=6c75063c74f7d83eb374c8aacb6c39821ecdfbc8;hpb=4511475f03107eb53c88545649898e3cfbcc2765;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/control.go b/pkg/control/control.go index 6c75063..f6cd771 100644 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -21,67 +21,208 @@ package control /* #include +#include #cgo CFLAGS: -I../ #cgo LDFLAGS: -lrmr_nng -lnng */ import "C" - import ( - "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" - "errors" - "strconv" + "errors" + "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" + "github.com/spf13/viper" + "github.com/go-openapi/strfmt" + httptransport "github.com/go-openapi/runtime/client" + rtmgrclient "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_client" + rtmgrhandle "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_client/handle" + "math/rand" + "strconv" + "time" ) type Control struct { - e2ap *E2ap - registry *Registry + e2ap *E2ap + registry *Registry + rtmgrClient *RtmgrClient + tracker *Tracker +} + +type RMRMeid struct { + PlmnID string + EnbID string +} + +type RMRParams struct { + Mtype int + Payload []byte + PayloadLen int + Meid *RMRMeid + Xid string + SubId int + Src string + Mbuf *C.rmr_mbuf_t +} + +var SEEDSN uint16 +var SubscriptionReqChan = make(chan subRouteInfo, 10) + +const ( + CREATE Action = 0 + MERGE Action = 1 + DELETE Action = 3 +) + +func init() { + viper.AutomaticEnv() + viper.SetEnvPrefix("submgr") + viper.AllowEmptyEnv(true) + SEEDSN = uint16(viper.GetInt("seed_sn")) + if SEEDSN == 0 { + rand.Seed(time.Now().UnixNano()) + SEEDSN = uint16(rand.Intn(65535)) + } + if SEEDSN > 65535 { + SEEDSN = 0 + } + xapp.Logger.Info("SUBMGR: Initial Sequence Number: %v", SEEDSN) } func NewControl() Control { - return Control{new(E2ap),new(Registry)} + registry := new(Registry) + registry.Initialize(SEEDSN) + + tracker := new(Tracker) + tracker.Init() + + transport := httptransport.New(viper.GetString("rtmgr.HostAddr") + ":" + viper.GetString("rtmgr.port"), viper.GetString("rtmgr.baseUrl"), []string{"http"}) + client := rtmgrclient.New(transport, strfmt.Default) + handle := rtmgrhandle.NewProvideXappSubscriptionHandleParamsWithTimeout(10 * time.Second) + rtmgrClient := RtmgrClient{client, handle} + + return Control{new(E2ap), registry, &rtmgrClient, tracker} } func (c *Control) Run() { - xapp.Run(c) + xapp.Run(c) +} + +func (c *Control) Consume(rp *xapp.RMRParams) (err error) { + switch rp.Mtype { + case C.RIC_SUB_REQ: + err = c.handleSubscriptionRequest(rp) + case C.RIC_SUB_RESP: + err = c.handleSubscriptionResponse(rp) + case C.RIC_SUB_DEL_REQ: + err = c.handleSubscriptionDeleteRequest(rp) + default: + err = errors.New("Message Type " + strconv.Itoa(rp.Mtype) + " is discarded") + } + return +} + +func (c *Control) rmrSend(params *xapp.RMRParams) (err error) { + if !xapp.Rmr.Send(params, false) { + err = errors.New("rmr.Send() failed") + } + return +} + +func (c *Control) handleSubscriptionRequest(params *xapp.RMRParams) (err error) { + payload_seq_num, err := c.e2ap.GetSubscriptionRequestSequenceNumber(params.Payload) + if err != nil { + err = errors.New("Unable to get Subscription Sequence Number from Payload due to: " + err.Error()) + return + } + xapp.Logger.Info("Subscription Request Received. RMR SUBSCRIPTION_ID: %v | PAYLOAD SEQUENCE_NUMBER: %v", params.SubId, payload_seq_num) + + /* Reserve a sequence number and set it in the payload */ + new_sub_id := c.registry.ReserveSequenceNumber() + + _, err = c.e2ap.SetSubscriptionRequestSequenceNumber(params.Payload, new_sub_id) + if err != nil { + err = errors.New("Unable to set Subscription Sequence Number in Payload due to: " + err.Error()) + return + } + + src_addr, src_port, err := c.rtmgrClient.SplitSource(params.Src) + if err != nil { + xapp.Logger.Error("Failed to update routing-manager about the subscription request with reason: %s", err) + return + } + + /* Create transatcion records for every subscription request */ + xact_key := Transaction_key{new_sub_id, CREATE} + xact_value := Transaction{*src_addr, *src_port, params.Payload} + err = c.tracker.Track_transaction(xact_key, xact_value) + if err != nil { + xapp.Logger.Error("Failed to create a transaction record due to %v", err) + return + } + + /* Update routing manager about the new subscription*/ + sub_route_action := subRouteInfo{CREATE, *src_addr, *src_port, new_sub_id } + go c.rtmgrClient.SubscriptionRequestUpdate() + SubscriptionReqChan <- sub_route_action + + // Setting new subscription ID in the RMR header + params.SubId = int(new_sub_id) + + xapp.Logger.Info("Generated ID: %v. Forwarding to E2 Termination...", int(new_sub_id)) + c.rmrSend(params) + return } -func (c *Control) Consume(mtype, sub_id int, len int, payload []byte) (err error) { - switch mtype { - case C.RIC_SUB_REQ: - err = c.handleSubscriptionRequest(&RmrDatagram{mtype, sub_id, payload}) - case C.RIC_SUB_RESP: - err = c.handleSubscriptionResponse(&RmrDatagram{mtype, sub_id, payload}) - default: - err = errors.New("Message Type "+strconv.Itoa(mtype)+" discarded") - } - return +func (c *Control) handleSubscriptionResponse(params *xapp.RMRParams) (err error) { + payload_seq_num, err := c.e2ap.GetSubscriptionResponseSequenceNumber(params.Payload) + if err != nil { + err = errors.New("Unable to get Subscription Sequence Number from Payload due to: " + err.Error()) + return + } + xapp.Logger.Info("Subscription Response Received. RMR SUBSCRIPTION_ID: %v | PAYLOAD SEQUENCE_NUMBER: %v", params.SubId, payload_seq_num) + if !c.registry.IsValidSequenceNumber(payload_seq_num) { + err = errors.New("Unknown Subscription ID: " + strconv.Itoa(int(payload_seq_num)) + " in Subscritpion Response. Message discarded.") + return + } + c.registry.setSubscriptionToConfirmed(payload_seq_num) + xapp.Logger.Info("Subscription Response Registered. Forwarding to Requestor...") + c.rmrSend(params) + return } -func (c *Control) rmrSend(datagram *RmrDatagram) (err error) { - if !xapp.Rmr.Send(datagram.MessageType, datagram.SubscriptionId, len(datagram.Payload), datagram.Payload) { - err = errors.New("rmr.Send() failed") - } - return +func (act Action) String() string { + actions := [...]string{ + "CREATE", + "MERGE", + "DELETE", + } + + if act < CREATE || act > DELETE { + return "Unknown" + } + return actions[act] } -func (c *Control) handleSubscriptionRequest(datagram *RmrDatagram) ( err error) { - content, err := c.e2ap.GetPayloadContent(datagram.Payload) - xapp.Logger.Info("Subscription Request received: %v", content) - new_sub_id := c.registry.GetSubscriptionId() - payload, err := c.e2ap.SetSubscriptionSequenceNumber(datagram.Payload, new_sub_id) - if err != nil { - xapp.Logger.Error("Unable to set Subscription Sequence Number in Payload due to: "+ err.Error()) - return - } - xapp.Logger.Info("New Subscription Accepted, Forwarding to E2T") - c.rmrSend(&RmrDatagram{C.RIC_SUB_REQ , new_sub_id, payload}) - return +func (act Action) valid() bool { + switch act { + case CREATE, MERGE, DELETE: + return true + default: + return false + } } -func (c *Control) handleSubscriptionResponse(datagram *RmrDatagram) ( err error) { - content, err := c.e2ap.GetPayloadContent(datagram.Payload) - xapp.Logger.Info("Subscription Response received: %v", content) - return -} \ No newline at end of file +func (c *Control) handleSubscriptionDeleteRequest(params *xapp.RMRParams) (err error) { + payload_seq_num, err := c.e2ap.GetSubscriptionDeleteRequestSequenceNumber(params.Payload) + if err != nil { + err = errors.New("Unable to get Subscription Sequence Number from Payload due to: " + err.Error()) + return + } + xapp.Logger.Info("Subscription Delete Request Received. RMR SUBSCRIPTION_ID: %v | PAYLOAD SEQUENCE_NUMBER: %v", params.SubId, payload_seq_num) + if c.registry.IsValidSequenceNumber(payload_seq_num) { + c.registry.deleteSubscription(payload_seq_num) + } + xapp.Logger.Info("Subscription ID: %v. Forwarding to E2 Termination...", int(payload_seq_num)) + c.rmrSend(params) + return +}