X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=pkg%2Fcontrol%2Fcontrol.go;h=27e3fbe4b31d39c05474ed4612001518c6d6038c;hb=93cc3e245f87798c8753209980817727e0648401;hp=9a5022fc885319623947d6cf8f75bd31676eb51b;hpb=fbc56f981b7bd3387f7326401ceccbf31893a2ed;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/control.go b/pkg/control/control.go index 9a5022f..27e3fbe 100644 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -21,6 +21,7 @@ package control /* #include +#include #cgo CFLAGS: -I../ #cgo LDFLAGS: -lrmr_nng -lnng @@ -31,17 +32,46 @@ import ( "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() @@ -61,63 +91,135 @@ func init() { func NewControl() Control { registry := new(Registry) registry.Initialize(SEEDSN) - return Control{new(E2ap), registry} + + 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, new(Tracker)} } func (c *Control) Run() { xapp.Run(c) } -func (c *Control) Consume(mtype, sub_id int, len int, payload []byte) (err error) { - switch mtype { +func (c *Control) Consume(rp *xapp.RMRParams) (err error) { + switch rp.Mtype { case C.RIC_SUB_REQ: - err = c.handleSubscriptionRequest(&RmrDatagram{mtype, uint16(sub_id), payload}) + err = c.handleSubscriptionRequest(rp) case C.RIC_SUB_RESP: - err = c.handleSubscriptionResponse(&RmrDatagram{mtype, uint16(sub_id), payload}) + err = c.handleSubscriptionResponse(rp) + case C.RIC_SUB_DEL_REQ: + err = c.handleSubscriptionDeleteRequest(rp) default: - err = errors.New("Message Type " + strconv.Itoa(mtype) + " is discarded") + err = errors.New("Message Type " + strconv.Itoa(rp.Mtype) + " is discarded") } return } -func (c *Control) rmrSend(datagram *RmrDatagram) (err error) { - if !xapp.Rmr.Send(datagram.MessageType, int(datagram.SubscriptionId), len(datagram.Payload), datagram.Payload) { +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(datagram *RmrDatagram) (err error) { - payload_seq_num, err := c.e2ap.GetSubscriptionRequestSequenceNumber(datagram.Payload) +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", datagram.SubscriptionId, payload_seq_num) + 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() - payload, err := c.e2ap.SetSubscriptionRequestSequenceNumber(datagram.Payload, new_sub_id) + + _, 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(&RmrDatagram{C.RIC_SUB_REQ, new_sub_id, payload}) + c.rmrSend(params) return } -func (c *Control) handleSubscriptionResponse(datagram *RmrDatagram) (err error) { - payload_seq_num, err := c.e2ap.GetSubscriptionResponseSequenceNumber(datagram.Payload) +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", datagram.SubscriptionId, payload_seq_num) + 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(&RmrDatagram{C.RIC_SUB_RESP, payload_seq_num, datagram.Payload}) + c.rmrSend(params) + return +} + +func (act Action) String() string { + actions := [...]string{ + "CREATE", + "MERGE", + "DELETE", + } + + if act < CREATE || act > DELETE { + return "Unknown" + } + return actions[act] +} + +func (act Action) valid() bool { + switch act { + case CREATE, MERGE, DELETE: + return true + default: + return false + } +} + +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 }