registry *Registry
rtmgrClient *RtmgrClient
tracker *Tracker
+ rc_chan chan *xapp.RMRParams
}
type RMRMeid struct {
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)
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}
+ delete_handle := rtmgrhandle.NewDeleteXappSubscriptionHandleParamsWithTimeout(10 * time.Second)
+ rtmgrClient := RtmgrClient{client, handle, delete_handle}
- return Control{new(E2ap), registry, &rtmgrClient, tracker}
+ return Control{new(E2ap), registry, &rtmgrClient, tracker, make(chan *xapp.RMRParams)}
}
func (c *Control) Run() {
+ go c.controlLoop()
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")
- }
+ c.rc_chan <- rp
return
}
return
}
+func (c *Control) rmrReplyToSender(params *xapp.RMRParams) (err error) {
+ if !xapp.Rmr.Send(params, true) {
+ err = errors.New("rmr.Send() failed")
+ }
+ return
+}
+
+func (c *Control) controlLoop() {
+ for {
+ msg := <-c.rc_chan
+ switch msg.Mtype {
+ case C.RIC_SUB_REQ:
+ c.handleSubscriptionRequest(msg)
+ case C.RIC_SUB_RESP:
+ c.handleSubscriptionResponse(msg)
+ case C.RIC_SUB_DEL_REQ:
+ c.handleSubscriptionDeleteRequest(msg)
+ case C.RIC_SUB_DEL_RESP:
+ c.handleSubscriptionDeleteResponse(msg)
+ default:
+ err := errors.New("Message Type " + strconv.Itoa(msg.Mtype) + " is discarded")
+ xapp.Logger.Error("Unknown message type: %v", err)
+ }
+ }
+}
+
func (c *Control) handleSubscriptionRequest(params *xapp.RMRParams) (err error) {
payload_seq_num, err := c.e2ap.GetSubscriptionRequestSequenceNumber(params.Payload)
if err != nil {
/* Create transatcion records for every subscription request */
xact_key := Transaction_key{new_sub_id, CREATE}
- xact_value := Transaction{*src_addr, *src_port, params.Payload}
+ xact_value := Transaction{*src_addr, *src_port, params.Payload, params.Mbuf}
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)
xapp.Logger.Info("Generated ID: %v. Forwarding to E2 Termination...", int(new_sub_id))
c.rmrSend(params)
+ xapp.Logger.Info("--- Debugging transaction table = %v", c.tracker.transaction_table)
return
}
}
c.registry.setSubscriptionToConfirmed(payload_seq_num)
xapp.Logger.Info("Subscription Response Registered. Forwarding to Requestor...")
- c.rmrSend(params)
+ transaction, err := c.tracker.complete_transaction(payload_seq_num, DELETE)
+ if err != nil {
+ xapp.Logger.Error("Failed to create a transaction record due to %v", err)
+ return
+ }
+ xapp.Logger.Info("Subscription ID: %v, from address: %v:%v. Forwarding to E2 Termination...", int(payload_seq_num), transaction.Xapp_instance_address, transaction.Xapp_port)
+ params.Mbuf = transaction.Mbuf
+ c.rmrReplyToSender(params)
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)
+ trackErr := c.trackDeleteTransaction(params, payload_seq_num)
+ if trackErr != nil {
+ xapp.Logger.Error("Failed to create a transaction record due to %v", err)
+ return trackErr
+ }
}
xapp.Logger.Info("Subscription ID: %v. Forwarding to E2 Termination...", int(payload_seq_num))
c.rmrSend(params)
return
}
+
+func (c *Control) trackDeleteTransaction(params *xapp.RMRParams, payload_seq_num uint16) (err error) {
+ src_addr, src_port, err := c.rtmgrClient.SplitSource(params.Src)
+ xact_key := Transaction_key{payload_seq_num, DELETE}
+ xact_value := Transaction{*src_addr, *src_port, params.Payload, params.Mbuf}
+ err = c.tracker.Track_transaction(xact_key, xact_value)
+ return
+}
+
+func (c *Control) handleSubscriptionDeleteResponse(params *xapp.RMRParams) (err error) {
+ payload_seq_num, err := c.e2ap.GetSubscriptionDeleteResponseSequenceNumber(params.Payload)
+ if err != nil {
+ err = errors.New("Unable to get Subscription Sequence Number from Payload due to: " + err.Error())
+ return
+ }
+ var transaction , _= c.tracker.Retrive_transaction(payload_seq_num, DELETE)
+ sub_route_action := subRouteInfo{DELETE, transaction.Xapp_instance_address, transaction.Xapp_port, payload_seq_num }
+ go c.rtmgrClient.SubscriptionRequestUpdate()
+ SubscriptionReqChan <- sub_route_action
+
+ xapp.Logger.Info("Subscription Delete Response Received. RMR SUBSCRIPTION_ID: %v | PAYLOAD SEQUENCE_NUMBER: %v", params.SubId, payload_seq_num)
+ if c.registry.releaseSequenceNumber(payload_seq_num) {
+ transaction, err = c.tracker.complete_transaction(payload_seq_num, DELETE)
+ if err != nil {
+ xapp.Logger.Error("Failed to create a transaction record due to %v", err)
+ return
+ }
+ xapp.Logger.Info("Subscription ID: %v, from address: %v:%v. Forwarding to E2 Termination...", int(payload_seq_num), transaction.Xapp_instance_address, transaction.Xapp_port)
+ //params.Src = xAddress + ":" + strconv.Itoa(int(xPort))
+ params.Mbuf = transaction.Mbuf
+ c.rmrReplyToSender(params)
+ }
+ return
+}
Code Review / ric-plt / submgr.git / blobdiff