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 import (
  23         "errors"
  24         rtmgrclient "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_client"
  25         rtmgrhandle "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_client/handle"
  26         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  27         httptransport "github.com/go-openapi/runtime/client"
  28         "github.com/go-openapi/strfmt"
  29         "github.com/spf13/viper"
  30         "math/rand"
  31         "sync"
  32         "time"
  33 )
  34
  35 var subReqTime time.Duration = 5 * time.Second
  36 var subDelReqTime time.Duration = 5 * time.Second
  37 var maxSubReqTryCount uint64 = 2    // Initial try + retry
  38 var maxSubDelReqTryCount uint64 = 2 // Initial try + retry
  39
  40 type Control struct {
  41         e2ap         *E2ap
  42         registry     *Registry
  43         rtmgrClient  *RtmgrClient
  44         tracker      *Tracker
  45         timerMap     *TimerMap
  46         rmrSendMutex sync.Mutex
  47         msgCounter   uint64
  48 }
  49
  50 type RMRMeid struct {
  51         PlmnID  string
  52         EnbID   string
  53         RanName string
  54 }
  55
  56 var seedSN uint16
  57
  58 const (
  59         CREATE Action = 0
  60         MERGE  Action = 1
  61         NONE   Action = 2
  62         DELETE Action = 3
  63 )
  64
  65 func init() {
  66         xapp.Logger.Info("SUBMGR")
  67         viper.AutomaticEnv()
  68         viper.SetEnvPrefix("submgr")
  69         viper.AllowEmptyEnv(true)
  70         seedSN = uint16(viper.GetInt("seed_sn"))
  71         if seedSN == 0 {
  72                 rand.Seed(time.Now().UnixNano())
  73                 seedSN = uint16(rand.Intn(65535))
  74         }
  75         if seedSN > 65535 {
  76                 seedSN = 0
  77         }
  78         xapp.Logger.Info("SUBMGR: Initial Sequence Number: %v", seedSN)
  79 }
  80
  81 func NewControl() *Control {
  82         registry := new(Registry)
  83         registry.Initialize(seedSN)
  84
  85         tracker := new(Tracker)
  86         tracker.Init()
  87
  88         timerMap := new(TimerMap)
  89         timerMap.Init()
  90
  91         transport := httptransport.New(viper.GetString("rtmgr.HostAddr")+":"+viper.GetString("rtmgr.port"), viper.GetString("rtmgr.baseUrl"), []string{"http"})
  92         client := rtmgrclient.New(transport, strfmt.Default)
  93         handle := rtmgrhandle.NewProvideXappSubscriptionHandleParamsWithTimeout(10 * time.Second)
  94         deleteHandle := rtmgrhandle.NewDeleteXappSubscriptionHandleParamsWithTimeout(10 * time.Second)
  95         rtmgrClient := RtmgrClient{client, handle, deleteHandle}
  96
  97         return &Control{e2ap: new(E2ap),
  98                 registry:    registry,
  99                 rtmgrClient: &rtmgrClient,
 100                 tracker:     tracker,
 101                 timerMap:    timerMap,
 102                 msgCounter:  0,
 103         }
 104 }
 105
 106 func (c *Control) Run() {
 107         xapp.Run(c)
 108 }
 109
 110 func (c *Control) rmrSend(params *xapp.RMRParams) (err error) {
 111         status := false
 112         i := 1
 113         for ; i <= 10 && status == false; i++ {
 114                 c.rmrSendMutex.Lock()
 115                 status = xapp.Rmr.Send(params, false)
 116                 c.rmrSendMutex.Unlock()
 117                 if status == false {
 118                         xapp.Logger.Info("rmr.Send() failed. Retry count %v, Mtype: %v, SubId: %v, Xid %s", i, params.Mtype, params.SubId, params.Xid)
 119                         time.Sleep(500 * time.Millisecond)
 120                 }
 121         }
 122         if status == false {
 123                 err = errors.New("rmr.Send() failed")
 124                 xapp.Rmr.Free(params.Mbuf)
 125         }
 126         return
 127 }
 128
 129 func (c *Control) rmrReplyToSender(params *xapp.RMRParams) (err error) {
 130         c.rmrSend(params)
 131         return
 132 }
 133
 134 func (c *Control) Consume(msg *xapp.RMRParams) (err error) {
 135         c.msgCounter++
 136         switch msg.Mtype {
 137         case xapp.RICMessageTypes["RIC_SUB_REQ"]:
 138                 go c.handleSubscriptionRequest(msg)
 139         case xapp.RICMessageTypes["RIC_SUB_RESP"]:
 140                 go c.handleSubscriptionResponse(msg)
 141         case xapp.RICMessageTypes["RIC_SUB_FAILURE"]:
 142                 go c.handleSubscriptionFailure(msg)
 143         case xapp.RICMessageTypes["RIC_SUB_DEL_REQ"]:
 144                 go c.handleSubscriptionDeleteRequest(msg)
 145         case xapp.RICMessageTypes["RIC_SUB_DEL_RESP"]:
 146                 go c.handleSubscriptionDeleteResponse(msg)
 147         case xapp.RICMessageTypes["RIC_SUB_DEL_FAILURE"]:
 148                 go c.handleSubscriptionDeleteFailure(msg)
 149         default:
 150                 xapp.Logger.Info("Unknown Message Type '%d', discarding", msg.Mtype)
 151         }
 152         return nil
 153 }
 154
 155 func (c *Control) handleSubscriptionRequest(params *xapp.RMRParams) {
 156         xapp.Logger.Info("SubReq received from Src: %s, Mtype: %v, SubId: %v, Xid: %s, Meid: %v", params.Src, params.Mtype, params.SubId, params.Xid, params.Meid)
 157         xapp.Rmr.Free(params.Mbuf)
 158         params.Mbuf = nil
 159
 160         /* Reserve a sequence number and set it in the payload */
 161         subs := c.registry.ReserveSubscription()
 162         if subs == nil {
 163                 xapp.Logger.Error("SubReq: Failed to reserve sequence number. Dropping this msg. SubId: %v, Xid: %s", params.SubId, params.Xid)
 164                 return
 165         }
 166
 167         params.SubId = int(subs.Seq)
 168         err := c.e2ap.SetSubscriptionRequestSequenceNumber(params.Payload, subs.Seq)
 169         if err != nil {
 170                 xapp.Logger.Error("SubReq: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, SubId: %v, Xid: %s, Payload %X", err, params.SubId, params.Xid, params.Payload)
 171                 c.registry.releaseSequenceNumber(subs.Seq)
 172                 return
 173         }
 174
 175         srcAddr, srcPort, err := c.rtmgrClient.SplitSource(params.Src)
 176         if err != nil {
 177                 xapp.Logger.Error("SubReq: Failed to update routing-manager. Dropping this msg. Err: %s, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 178                 c.registry.releaseSequenceNumber(subs.Seq)
 179                 return
 180         }
 181
 182         // Create transatcion record for every subscription request
 183         var forwardRespToXapp bool = true
 184         var responseReceived bool = false
 185         transaction, err := c.tracker.TrackTransaction(subs.Seq, CREATE, *srcAddr, *srcPort, params, responseReceived, forwardRespToXapp)
 186         if err != nil {
 187                 xapp.Logger.Error("SubReq: Failed to create transaction record. Dropping this msg. Err: %v SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 188                 c.registry.releaseSequenceNumber(subs.Seq)
 189                 return
 190         }
 191
 192         // Update routing manager about the new subscription
 193         subRouteAction := transaction.SubRouteInfo()
 194         xapp.Logger.Info("SubReq: Starting routing manager update. SubId: %v, Xid: %s", params.SubId, params.Xid)
 195
 196         err = c.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
 197         if err != nil {
 198                 xapp.Logger.Error("SubReq: Failed to update routing manager. Dropping this msg. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 199                 c.registry.releaseSequenceNumber(subs.Seq)
 200                 return
 201         }
 202
 203         // Setting new subscription ID in the RMR header
 204         xapp.Logger.Info("SubReq: Forwarding SubReq to E2T: Mtype: %v, SubId: %v, Xid %s, Meid %v", params.Mtype, params.SubId, params.Xid, params.Meid)
 205         err = c.rmrSend(params)
 206         if err != nil {
 207                 xapp.Logger.Error("SubReq: Failed to send request to E2T %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 208         }
 209         c.timerMap.StartTimer("RIC_SUB_REQ", int(subs.Seq), subReqTime, FirstTry, c.handleSubscriptionRequestTimer)
 210         xapp.Logger.Debug("SubReq: Debugging transaction table = %v", c.tracker.transactionTable)
 211         return
 212 }
 213
 214 func (c *Control) handleSubscriptionResponse(params *xapp.RMRParams) {
 215         xapp.Logger.Info("SubResp received from Src: %s, Mtype: %v, SubId: %v, Meid: %v", params.Src, params.Mtype, params.SubId, params.Meid)
 216         xapp.Rmr.Free(params.Mbuf)
 217         params.Mbuf = nil
 218
 219         payloadSeqNum, err := c.e2ap.GetSubscriptionResponseSequenceNumber(params.Payload)
 220         if err != nil {
 221                 xapp.Logger.Error("SubResp: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, SubId: %v, Xid: %s, Payload %X", err, params.SubId, params.Xid, params.Payload)
 222                 return
 223         }
 224         xapp.Logger.Info("SubResp: Received payloadSeqNum: %v", payloadSeqNum)
 225
 226         if !c.registry.IsValidSequenceNumber(payloadSeqNum) {
 227                 xapp.Logger.Error("SubResp: Unknown payloadSeqNum. Dropping this msg. PayloadSeqNum: %v, SubId: %v", payloadSeqNum, params.SubId)
 228                 return
 229         }
 230
 231         c.timerMap.StopTimer("RIC_SUB_REQ", int(payloadSeqNum))
 232
 233         transaction, responseReceived, err := c.tracker.CheckResponseReceived(payloadSeqNum, CREATE)
 234         if err != nil {
 235                 xapp.Logger.Info("SubResp: Dropping this msg. Err: %v SubId: %v", err, payloadSeqNum)
 236                 return
 237         }
 238
 239         if responseReceived == true {
 240                 // Subscription timer already received
 241                 return
 242         }
 243         xapp.Logger.Info("SubResp: SubId: %v, from address: %v:%v.", payloadSeqNum, transaction.Xappkey.Addr, transaction.Xappkey.Port)
 244
 245         c.registry.setSubscriptionToConfirmed(payloadSeqNum)
 246
 247         params.SubId = int(payloadSeqNum)
 248         params.Xid = transaction.OrigParams.Xid
 249
 250         xapp.Logger.Info("SubResp: Forwarding Subscription Response to xApp Mtype: %v, SubId: %v, Meid: %v", params.Mtype, params.SubId, params.Meid)
 251         err = c.rmrReplyToSender(params)
 252         if err != nil {
 253                 xapp.Logger.Error("SubResp: Failed to send response to xApp. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 254         }
 255
 256         xapp.Logger.Info("SubResp: SubId: %v, from address: %v:%v. Deleting transaction record", payloadSeqNum, transaction.Xappkey.Addr, transaction.Xappkey.Port)
 257         _, err = c.tracker.completeTransaction(payloadSeqNum, CREATE)
 258         if err != nil {
 259                 xapp.Logger.Error("SubResp: Failed to delete transaction record. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 260                 return
 261         }
 262         return
 263 }
 264
 265 func (c *Control) handleSubscriptionFailure(params *xapp.RMRParams) {
 266         xapp.Logger.Info("SubFail received from Src: %s, Mtype: %v, SubId: %v, Meid: %v", params.Src, params.Mtype, params.SubId, params.Meid)
 267         xapp.Rmr.Free(params.Mbuf)
 268         params.Mbuf = nil
 269
 270         payloadSeqNum, err := c.e2ap.GetSubscriptionFailureSequenceNumber(params.Payload)
 271         if err != nil {
 272                 xapp.Logger.Error("SubFail: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, SubId: %v, Xid: %s, Payload %X", err, params.SubId, params.Xid, params.Payload)
 273                 return
 274         }
 275         xapp.Logger.Info("SubFail: Received payloadSeqNum: %v", payloadSeqNum)
 276
 277         c.timerMap.StopTimer("RIC_SUB_REQ", int(payloadSeqNum))
 278
 279         transaction, responseReceived, err := c.tracker.CheckResponseReceived(payloadSeqNum, CREATE)
 280         if err != nil {
 281                 xapp.Logger.Info("SubFail: Dropping this msg. Err: %v SubId: %v", err, payloadSeqNum)
 282                 return
 283         }
 284
 285         if responseReceived == true {
 286                 // Subscription timer already received
 287                 return
 288         }
 289         xapp.Logger.Info("SubFail: SubId: %v, from address: %v:%v. Forwarding response to xApp", payloadSeqNum, transaction.Xappkey.Addr, transaction.Xappkey.Port)
 290
 291         time.Sleep(3 * time.Second)
 292
 293         xapp.Logger.Info("SubFail: Starting routing manager update. SubId: %v, Xid: %s", params.SubId, params.Xid)
 294         subRouteAction := transaction.SubRouteInfo()
 295         err = c.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
 296         if err != nil {
 297                 xapp.Logger.Error("SubFail: Failed to update routing manager. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 298         }
 299
 300         xapp.Logger.Info("SubFail: Deleting transaction record. SubId: %v, Xid: %s", params.SubId, params.Xid)
 301         if c.registry.releaseSequenceNumber(payloadSeqNum) {
 302                 _, err = c.tracker.completeTransaction(payloadSeqNum, CREATE)
 303                 if err != nil {
 304                         xapp.Logger.Error("SubFail: Failed to delete transaction record. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 305                         return
 306                 }
 307         } else {
 308                 xapp.Logger.Error("SubFail: Failed to release sequency number. SubId: %v, Xid: %s", params.SubId, params.Xid)
 309                 return
 310         }
 311         return
 312 }
 313
 314 func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCount uint64) {
 315         subId := uint16(nbrId)
 316         xapp.Logger.Info("handleSubTimer: SubReq timer expired. subId: %v,  tryCount: %v", subId, tryCount)
 317
 318         transaction, responseReceived, err := c.tracker.CheckResponseReceived(subId, CREATE)
 319         if err != nil {
 320                 xapp.Logger.Info("handleSubTimer: Dropping this timer action. Err: %v SubId: %v", err, subId)
 321                 return
 322         }
 323
 324         if responseReceived == true {
 325                 // Subscription Response or Failure already received
 326                 return
 327         }
 328
 329         if tryCount < maxSubReqTryCount {
 330                 xapp.Logger.Info("handleSubTimer: Resending SubReq to E2T: Mtype: %v, SubId: %v, Xid %s, Meid %v", transaction.OrigParams.Mtype, transaction.OrigParams.SubId, transaction.OrigParams.Xid, transaction.OrigParams.Meid)
 331                 // Set possible to handle new response for the subId
 332                 err = c.tracker.RetryTransaction(subId, CREATE)
 333                 if err != nil {
 334                         xapp.Logger.Error("handleSubDelTimer: Failed to retry transaction record. Dropping timer action. Err %v, SubId: %v", err, transaction.OrigParams.SubId)
 335                         return
 336                 }
 337
 338                 err = c.rmrSend(transaction.OrigParams)
 339                 if err != nil {
 340                         xapp.Logger.Error("handleSubTimer: Failed to send request to E2T %v, SubId: %v, Xid: %s", err, transaction.OrigParams.SubId, transaction.OrigParams.Xid)
 341                 }
 342
 343                 tryCount++
 344                 c.timerMap.StartTimer("RIC_SUB_REQ", int(subId), subReqTime, tryCount, c.handleSubscriptionRequestTimer)
 345                 return
 346         }
 347
 348         var subDelReqPayload []byte
 349         subDelReqPayload, err = c.e2ap.PackSubscriptionDeleteRequest(transaction.OrigParams.Payload, subId)
 350         if err != nil {
 351                 xapp.Logger.Error("handleSubTimer: Packing SubDelReq failed. Err: %v", err)
 352                 return
 353         }
 354
 355         // Cancel failed subscription
 356         var params xapp.RMRParams
 357         params.Mtype = 12020 // RIC SUBSCRIPTION DELETE
 358         params.SubId = int(subId)
 359         params.Xid = transaction.OrigParams.Xid
 360         params.Meid = transaction.OrigParams.Meid
 361         params.Src = transaction.OrigParams.Src
 362         params.PayloadLen = len(subDelReqPayload)
 363         params.Payload = subDelReqPayload
 364         params.Mbuf = nil
 365
 366         // Delete CREATE transaction
 367         _, err = c.tracker.completeTransaction(subId, CREATE)
 368         if err != nil {
 369                 xapp.Logger.Error("handleSubTimer: Failed to delete create transaction record. Dropping this timer action. Err: %v, SubId: %v, Xid: %s", err, subId, params.Xid)
 370                 return
 371         }
 372
 373         // Create DELETE transaction
 374         var forwardRespToXapp bool = false
 375         _, err = c.trackDeleteTransaction(&params, subId, forwardRespToXapp)
 376         if err != nil {
 377                 xapp.Logger.Error("handleSubTimer: Failed to create delete transaction record. Dropping this timer action. Err: %v, SubId: %v, Xid: %s", err, subId, params.Xid)
 378                 return
 379         }
 380
 381         xapp.Logger.Info("handleSubTimer: Sending SubDelReq to E2T: Mtype: %v, SubId: %v, Meid: %v", params.Mtype, params.SubId, params.Meid)
 382         c.rmrSend(&params)
 383         if err != nil {
 384                 xapp.Logger.Error("handleSubTimer: Failed to send request to E2T %v. SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 385         }
 386         c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(subId), subDelReqTime, FirstTry, c.handleSubscriptionDeleteRequestTimer)
 387         return
 388 }
 389
 390 func (act Action) String() string {
 391         actions := [...]string{
 392                 "CREATE",
 393                 "MERGE",
 394                 "NONE",
 395                 "DELETE",
 396         }
 397
 398         if act < CREATE || act > DELETE {
 399                 return "Unknown"
 400         }
 401         return actions[act]
 402 }
 403
 404 func (act Action) valid() bool {
 405         switch act {
 406         case CREATE, MERGE, DELETE:
 407                 return true
 408         default:
 409                 return false
 410         }
 411 }
 412
 413 func (c *Control) handleSubscriptionDeleteRequest(params *xapp.RMRParams) {
 414         xapp.Logger.Info("SubDelReq received from Src: %s, Mtype: %v, SubId: %v, Xid: %s, Meid: %v", params.Src, params.Mtype, params.SubId, params.Xid, params.Meid)
 415         xapp.Rmr.Free(params.Mbuf)
 416         params.Mbuf = nil
 417
 418         payloadSeqNum, err := c.e2ap.GetSubscriptionDeleteRequestSequenceNumber(params.Payload)
 419         if err != nil {
 420                 xapp.Logger.Error("SubDelReq: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, SubId: %v, Xid: %s, Payload %X", err, params.SubId, params.Xid, params.Payload)
 421                 return
 422         }
 423         xapp.Logger.Info("SubDelReq: Received payloadSeqNum: %v", payloadSeqNum)
 424
 425         if c.registry.IsValidSequenceNumber(payloadSeqNum) {
 426                 var forwardRespToXapp bool = true
 427                 _, err = c.trackDeleteTransaction(params, payloadSeqNum, forwardRespToXapp)
 428                 if err != nil {
 429                         xapp.Logger.Error("SubDelReq: Failed to create transaction record. Dropping this msg. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 430                         return
 431                 }
 432                 c.registry.setSubscriptionToUnConfirmed(payloadSeqNum)
 433         } else {
 434                 xapp.Logger.Error("SubDelReq: Not valid sequence number. Dropping this msg. SubId: %v, Xid: %s", params.SubId, params.Xid)
 435                 return
 436         }
 437
 438         xapp.Logger.Info("SubDelReq: Forwarding Request to E2T. Mtype: %v, SubId: %v, Xid: %s, Meid: %v", params.Mtype, params.SubId, params.Xid, params.Meid)
 439         c.rmrSend(params)
 440         if err != nil {
 441                 xapp.Logger.Error("SubDelReq: Failed to send request to E2T. Err %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 442         }
 443         c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(payloadSeqNum), subDelReqTime, FirstTry, c.handleSubscriptionDeleteRequestTimer)
 444         return
 445 }
 446
 447 func (c *Control) trackDeleteTransaction(params *xapp.RMRParams, payloadSeqNum uint16, forwardRespToXapp bool) (transaction *Transaction, err error) {
 448         srcAddr, srcPort, err := c.rtmgrClient.SplitSource(params.Src)
 449         if err != nil {
 450                 xapp.Logger.Error("Failed to split source address. Err: %s, SubId: %v, Xid: %s", err, payloadSeqNum, params.Xid)
 451         }
 452         var respReceived bool = false
 453         transaction, err = c.tracker.TrackTransaction(payloadSeqNum, DELETE, *srcAddr, *srcPort, params, respReceived, forwardRespToXapp)
 454         return
 455 }
 456
 457 func (c *Control) handleSubscriptionDeleteResponse(params *xapp.RMRParams) (err error) {
 458         xapp.Logger.Info("SubDelResp received from Src: %s, Mtype: %v, SubId: %v, Meid: %v", params.Src, params.Mtype, params.SubId, params.Meid)
 459         xapp.Rmr.Free(params.Mbuf)
 460         params.Mbuf = nil
 461
 462         payloadSeqNum, err := c.e2ap.GetSubscriptionDeleteResponseSequenceNumber(params.Payload)
 463         if err != nil {
 464                 xapp.Logger.Error("SubDelResp: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, SubId: %v, Xid: %s, Payload %X", err, params.SubId, params.Xid, params.Payload)
 465                 return
 466         }
 467         xapp.Logger.Info("SubDelResp: Received payloadSeqNum: %v", payloadSeqNum)
 468
 469         c.timerMap.StopTimer("RIC_SUB_DEL_REQ", int(payloadSeqNum))
 470
 471         transaction, responseReceived, err := c.tracker.CheckResponseReceived(payloadSeqNum, DELETE)
 472         if err != nil {
 473                 xapp.Logger.Info("SubDelResp: Dropping this msg. Err: %v SubId: %v", err, payloadSeqNum)
 474                 return
 475         }
 476
 477         if responseReceived == true {
 478                 // Subscription Delete timer already received
 479                 return
 480         }
 481         xapp.Logger.Info("SubDelResp: SubId: %v, from address: %v:%v. Forwarding response to xApp", payloadSeqNum, transaction.Xappkey.Addr, transaction.Xappkey.Port)
 482
 483         if transaction.ForwardRespToXapp == true {
 484                 params.SubId = int(payloadSeqNum)
 485                 params.Xid = transaction.OrigParams.Xid
 486                 xapp.Logger.Info("Forwarding SubDelResp to xApp: Mtype: %v, SubId: %v, Xid: %v, Meid: %v", params.Mtype, params.SubId, params.Xid, params.Meid)
 487                 err = c.rmrReplyToSender(params)
 488                 if err != nil {
 489                         xapp.Logger.Error("SubDelResp: Failed to send response to xApp. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 490                 }
 491
 492                 time.Sleep(3 * time.Second)
 493         }
 494
 495         xapp.Logger.Info("SubDelResp: Starting routing manager update. SubId: %v, Xid: %s", params.SubId, params.Xid)
 496         subRouteAction := SubRouteInfo{DELETE, transaction.Xappkey.Addr, transaction.Xappkey.Port, payloadSeqNum}
 497         err = c.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
 498         if err != nil {
 499                 xapp.Logger.Error("SubDelResp: Failed to update routing manager. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 500                 return
 501         }
 502
 503         xapp.Logger.Info("SubDelResp: Deleting transaction record. SubId: %v, Xid: %s", params.SubId, params.Xid)
 504         if c.registry.releaseSequenceNumber(payloadSeqNum) {
 505                 _, err = c.tracker.completeTransaction(payloadSeqNum, DELETE)
 506                 if err != nil {
 507                         xapp.Logger.Error("SubDelResp: Failed to delete transaction record. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 508                         return
 509                 }
 510         } else {
 511                 xapp.Logger.Error("SubDelResp: Failed to release sequency number. SubId: %v, Xid: %s", params.SubId, params.Xid)
 512                 return
 513         }
 514         return
 515 }
 516
 517 func (c *Control) handleSubscriptionDeleteFailure(params *xapp.RMRParams) {
 518         xapp.Logger.Info("SubDelFail received from Src: %s, Mtype: %v, SubId: %v, Meid: %v", params.Src, params.Mtype, params.SubId, params.Meid)
 519         xapp.Rmr.Free(params.Mbuf)
 520         params.Mbuf = nil
 521
 522         payloadSeqNum, err := c.e2ap.GetSubscriptionDeleteFailureSequenceNumber(params.Payload)
 523         if err != nil {
 524                 xapp.Logger.Error("SubDelFail: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, SubId: %v, Xid: %s, Payload %X", err, params.SubId, params.Xid, params.Payload)
 525                 return
 526         }
 527         xapp.Logger.Info("SubDelFail: Received payloadSeqNum: %v", payloadSeqNum)
 528
 529         c.timerMap.StopTimer("RIC_SUB_DEL_REQ", int(payloadSeqNum))
 530
 531         transaction, responseReceived, err := c.tracker.CheckResponseReceived(payloadSeqNum, DELETE)
 532         if err != nil {
 533                 xapp.Logger.Info("SubDelFail: Dropping this msg. Err: %v SubId: %v", err, payloadSeqNum)
 534                 return
 535         }
 536
 537         if responseReceived == true {
 538                 // Subscription Delete timer already received
 539                 return
 540         }
 541         xapp.Logger.Info("SubDelFail: SubId: %v, from address: %v:%v. Forwarding response to xApp", payloadSeqNum, transaction.Xappkey.Addr, transaction.Xappkey.Port)
 542
 543         if transaction.ForwardRespToXapp == true {
 544                 var subDelRespPayload []byte
 545                 subDelRespPayload, err = c.e2ap.PackSubscriptionDeleteResponse(transaction.OrigParams.Payload, payloadSeqNum)
 546                 if err != nil {
 547                         xapp.Logger.Error("SubDelFail:Packing SubDelResp failed. Err: %v", err)
 548                         return
 549                 }
 550
 551                 params.Mtype = 12021 // RIC SUBSCRIPTION DELETE RESPONSE
 552                 params.SubId = int(payloadSeqNum)
 553                 params.Xid = transaction.OrigParams.Xid
 554                 params.Meid = transaction.OrigParams.Meid
 555                 params.Src = transaction.OrigParams.Src
 556                 params.PayloadLen = len(subDelRespPayload)
 557                 params.Payload = subDelRespPayload
 558                 params.Mbuf = nil
 559                 xapp.Logger.Info("SubDelFail: Forwarding SubDelFail to xApp: Mtype: %v, SubId: %v, Xid: %v, Meid: %v", params.Mtype, params.SubId, params.Xid, params.Meid)
 560                 err = c.rmrReplyToSender(params)
 561                 if err != nil {
 562                         xapp.Logger.Error("SubDelFail: Failed to send SubDelFail to xApp. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 563                 }
 564
 565                 time.Sleep(3 * time.Second)
 566         }
 567
 568         xapp.Logger.Info("SubDelFail: Starting routing manager update. SubId: %v, Xid: %s", params.SubId, params.Xid)
 569         subRouteAction := SubRouteInfo{DELETE, transaction.Xappkey.Addr, transaction.Xappkey.Port, payloadSeqNum}
 570         err = c.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
 571         if err != nil {
 572                 xapp.Logger.Error("SubDelFail: Failed to update routing manager. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 573                 return
 574         }
 575
 576         xapp.Logger.Info("SubDelFail: Deleting transaction record. SubId: %v, Xid: %s", params.SubId, params.Xid)
 577         if c.registry.releaseSequenceNumber(payloadSeqNum) {
 578                 _, err = c.tracker.completeTransaction(payloadSeqNum, DELETE)
 579                 if err != nil {
 580                         xapp.Logger.Error("SubDelFail: Failed to delete transaction record. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 581                         return
 582                 }
 583         } else {
 584                 xapp.Logger.Error("SubDelFail: Failed to release sequency number. Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 585                 return
 586         }
 587         return
 588 }
 589
 590 func (c *Control) handleSubscriptionDeleteRequestTimer(strId string, nbrId int, tryCount uint64) {
 591         subId := uint16(nbrId)
 592         xapp.Logger.Info("handleSubDelTimer: SubDelReq timer expired. subId: %v, tryCount: %v", subId, tryCount)
 593
 594         transaction, responseReceived, err := c.tracker.CheckResponseReceived(subId, DELETE)
 595         if err != nil {
 596                 xapp.Logger.Info("handleSubTimer: Dropping this timer action. Err: %v SubId: %v", err, subId)
 597                 return
 598         }
 599
 600         if responseReceived == true {
 601                 // Subscription Delete Response or Failure already received
 602                 return
 603         }
 604
 605         if tryCount < maxSubDelReqTryCount {
 606                 xapp.Logger.Info("handleSubDelTimer: Resending SubDelReq to E2T: Mtype: %v, SubId: %v, Xid %s, Meid %v", transaction.OrigParams.Mtype, transaction.OrigParams.SubId, transaction.OrigParams.Xid, transaction.OrigParams.Meid)
 607                 // Set possible to handle new response for the subId
 608                 err = c.tracker.RetryTransaction(subId, DELETE)
 609                 if err != nil {
 610                         xapp.Logger.Error("handleSubDelTimer: Failed to retry transaction record. Dropping  timer action. Err %v, SubId: %v", err, transaction.OrigParams.SubId)
 611                         return
 612                 }
 613
 614                 err = c.rmrSend(transaction.OrigParams)
 615                 if err != nil {
 616                         xapp.Logger.Error("handleSubDelTimer: Failed to send request to E2T %v, SubId: %v, Xid: %s", err, transaction.OrigParams.SubId, transaction.OrigParams.Xid)
 617                 }
 618
 619                 tryCount++
 620                 c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(subId), subReqTime, tryCount, c.handleSubscriptionDeleteRequestTimer)
 621                 return
 622         }
 623
 624         var params xapp.RMRParams
 625         if transaction.ForwardRespToXapp == true {
 626                 var subDelRespPayload []byte
 627                 subDelRespPayload, err = c.e2ap.PackSubscriptionDeleteResponse(transaction.OrigParams.Payload, subId)
 628                 if err != nil {
 629                         xapp.Logger.Error("handleSubDelTimer: Unable to pack payload. Dropping this timer action. Err: %v, SubId: %v, Xid: %s, Payload %x", err, subId, transaction.OrigParams.Xid, transaction.OrigParams.Payload)
 630                         return
 631                 }
 632
 633                 params.Mtype = 12021 // RIC SUBSCRIPTION DELETE RESPONSE
 634                 params.SubId = int(subId)
 635                 params.Meid = transaction.OrigParams.Meid
 636                 params.Xid = transaction.OrigParams.Xid
 637                 params.Src = transaction.OrigParams.Src
 638                 params.PayloadLen = len(subDelRespPayload)
 639                 params.Payload = subDelRespPayload
 640                 params.Mbuf = nil
 641
 642                 xapp.Logger.Info("handleSubDelTimer: Sending SubDelResp to xApp: Mtype: %v, SubId: %v, Xid: %s, Meid: %v", params.Mtype, params.SubId, params.Xid, params.Meid)
 643                 err = c.rmrReplyToSender(&params)
 644                 if err != nil {
 645                         xapp.Logger.Error("handleSubDelTimer: Failed to send response to xApp: Err: %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
 646                 }
 647
 648                 time.Sleep(3 * time.Second)
 649         }
 650
 651         xapp.Logger.Info("handleSubDelTimer: Starting routing manager update. SubId: %v, Xid: %s", subId, params.Xid)
 652         subRouteAction := SubRouteInfo{DELETE, transaction.Xappkey.Addr, transaction.Xappkey.Port, subId}
 653         err = c.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
 654         if err != nil {
 655                 xapp.Logger.Error("handleSubDelTimer: Failed to update routing manager. Err: %v, SubId: %v, Xid: %s", err, subId, params.Xid)
 656                 return
 657         }
 658
 659         xapp.Logger.Info("handleSubDelTimer: Deleting transaction record. SubId: %v, Xid: %s", subId, params.Xid)
 660         if c.registry.releaseSequenceNumber(subId) {
 661                 _, err = c.tracker.completeTransaction(subId, DELETE)
 662                 if err != nil {
 663                         xapp.Logger.Error("handleSubDelTimer: Failed to delete transaction record. Err: %v, SubId: %v, Xid: %s", err, subId, params.Xid)
 664                         return
 665                 }
 666         } else {
 667                 xapp.Logger.Error("handleSubDelTimer: Failed to release sequency number. SubId: %v, Xid: %s", subId, params.Xid)
 668         }
 669         return
 670 }