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         "fmt"
  24         "gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap"
  25         rtmgrclient "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_client"
  26         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models"
  27         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  28         httptransport "github.com/go-openapi/runtime/client"
  29         "github.com/go-openapi/strfmt"
  30         "github.com/spf13/viper"
  31         "time"
  32 )
  33
  34 //-----------------------------------------------------------------------------
  35 //
  36 //-----------------------------------------------------------------------------
  37
  38 func idstring(err error, entries ...fmt.Stringer) string {
  39         var retval string = ""
  40         var filler string = ""
  41         for _, entry := range entries {
  42                 retval += filler + entry.String()
  43                 filler = " "
  44         }
  45         if err != nil {
  46                 retval += filler + "err(" + err.Error() + ")"
  47                 filler = " "
  48
  49         }
  50         return retval
  51 }
  52
  53 //-----------------------------------------------------------------------------
  54 //
  55 //-----------------------------------------------------------------------------
  56
  57 var e2tSubReqTimeout time.Duration
  58 var e2tSubDelReqTime time.Duration
  59 var e2tRecvMsgTimeout time.Duration
  60 var e2tMaxSubReqTryCount uint64    // Initial try + retry
  61 var e2tMaxSubDelReqTryCount uint64 // Initial try + retry
  62
  63 type Control struct {
  64         *xapp.RMRClient
  65         e2ap     *E2ap
  66         registry *Registry
  67         tracker  *Tracker
  68         //subscriber *xapp.Subscriber
  69         CntRecvMsg uint64
  70 }
  71
  72 type RMRMeid struct {
  73         PlmnID  string
  74         EnbID   string
  75         RanName string
  76 }
  77
  78 func init() {
  79         xapp.Logger.Info("SUBMGR")
  80         viper.AutomaticEnv()
  81         viper.SetEnvPrefix("submgr")
  82         viper.AllowEmptyEnv(true)
  83 }
  84
  85 func NewControl() *Control {
  86
  87         // viper.GetDuration returns nanoseconds
  88         e2tSubReqTimeout = viper.GetDuration("controls.e2tSubReqTimeout_ms") * 1000000
  89         xapp.Logger.Info("e2tSubReqTimeout %v", e2tSubReqTimeout)
  90         e2tSubDelReqTime = viper.GetDuration("controls.e2tSubDelReqTime_ms") * 1000000
  91         xapp.Logger.Info("e2tSubDelReqTime %v", e2tSubDelReqTime)
  92         e2tRecvMsgTimeout = viper.GetDuration("controls.e2tRecvMsgTimeout_ms") * 1000000
  93         xapp.Logger.Info("e2tRecvMsgTimeout %v", e2tRecvMsgTimeout)
  94         e2tMaxSubReqTryCount = viper.GetUint64("controls.e2tMaxSubReqTryCount")
  95         xapp.Logger.Info("e2tMaxSubReqTryCount %v", e2tMaxSubReqTryCount)
  96         e2tMaxSubDelReqTryCount = viper.GetUint64("controls.e2tMaxSubDelReqTryCount")
  97         xapp.Logger.Info("e2tMaxSubDelReqTryCount %v", e2tMaxSubDelReqTryCount)
  98
  99         transport := httptransport.New(viper.GetString("rtmgr.HostAddr")+":"+viper.GetString("rtmgr.port"), viper.GetString("rtmgr.baseUrl"), []string{"http"})
 100         rtmgrClient := RtmgrClient{rtClient: rtmgrclient.New(transport, strfmt.Default)}
 101
 102         registry := new(Registry)
 103         registry.Initialize()
 104         registry.rtmgrClient = &rtmgrClient
 105
 106         tracker := new(Tracker)
 107         tracker.Init()
 108
 109         //subscriber := xapp.NewSubscriber(viper.GetString("subscription.host"), viper.GetInt("subscription.timeout"))
 110
 111         c := &Control{e2ap: new(E2ap),
 112                 registry: registry,
 113                 tracker:  tracker,
 114                 //subscriber: subscriber,
 115         }
 116         go xapp.Subscription.Listen(c.SubscriptionHandler, c.QueryHandler, c.SubscriptionDeleteHandler)
 117         //go c.subscriber.Listen(c.SubscriptionHandler, c.QueryHandler)
 118         return c
 119 }
 120
 121 func (c *Control) ReadyCB(data interface{}) {
 122         if c.RMRClient == nil {
 123                 c.RMRClient = xapp.Rmr
 124         }
 125 }
 126
 127 func (c *Control) Run() {
 128         xapp.SetReadyCB(c.ReadyCB, nil)
 129         xapp.Run(c)
 130 }
 131
 132 //-------------------------------------------------------------------
 133 //
 134 //-------------------------------------------------------------------
 135 func (c *Control) SubscriptionHandler(stype models.SubscriptionType, params interface{}) (*models.SubscriptionResponse, error) {
 136         /*
 137            switch p := params.(type) {
 138            case *models.ReportParams:
 139                trans := c.tracker.NewXappTransaction(NewRmrEndpoint(p.ClientEndpoint),"" , 0, &xapp.RMRMeid{RanName: p.Meid})
 140                if trans == nil {
 141                      xapp.Logger.Error("XAPP-SubReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 142                      return
 143                }
 144                defer trans.Release()
 145            case *models.ControlParams:
 146            case *models.PolicyParams:
 147            }
 148         */
 149         return &models.SubscriptionResponse{}, fmt.Errorf("Subscription rest interface not implemented")
 150 }
 151
 152 func (c *Control) SubscriptionDeleteHandler(string) error {
 153         return fmt.Errorf("Subscription rest interface not implemented")
 154 }
 155
 156 func (c *Control) QueryHandler() (models.SubscriptionList, error) {
 157         return c.registry.QueryHandler()
 158 }
 159
 160 //-------------------------------------------------------------------
 161 //
 162 //-------------------------------------------------------------------
 163
 164 func (c *Control) rmrSendToE2T(desc string, subs *Subscription, trans *TransactionSubs) (err error) {
 165         params := &xapp.RMRParams{}
 166         params.Mtype = trans.GetMtype()
 167         params.SubId = int(subs.GetReqId().InstanceId)
 168         params.Xid = ""
 169         params.Meid = subs.GetMeid()
 170         params.Src = ""
 171         params.PayloadLen = len(trans.Payload.Buf)
 172         params.Payload = trans.Payload.Buf
 173         params.Mbuf = nil
 174         xapp.Logger.Info("MSG to E2T: %s %s %s", desc, trans.String(), params.String())
 175         return c.SendWithRetry(params, false, 5)
 176 }
 177
 178 func (c *Control) rmrSendToXapp(desc string, subs *Subscription, trans *TransactionXapp) (err error) {
 179
 180         params := &xapp.RMRParams{}
 181         params.Mtype = trans.GetMtype()
 182         params.SubId = int(subs.GetReqId().InstanceId)
 183         params.Xid = trans.GetXid()
 184         params.Meid = trans.GetMeid()
 185         params.Src = ""
 186         params.PayloadLen = len(trans.Payload.Buf)
 187         params.Payload = trans.Payload.Buf
 188         params.Mbuf = nil
 189         xapp.Logger.Info("MSG to XAPP: %s %s %s", desc, trans.String(), params.String())
 190         return c.SendWithRetry(params, false, 5)
 191 }
 192
 193 func (c *Control) Consume(msg *xapp.RMRParams) (err error) {
 194         if c.RMRClient == nil {
 195                 err = fmt.Errorf("Rmr object nil can handle %s", msg.String())
 196                 xapp.Logger.Error("%s", err.Error())
 197                 return
 198         }
 199         c.CntRecvMsg++
 200
 201         defer c.RMRClient.Free(msg.Mbuf)
 202
 203         // xapp-frame might use direct access to c buffer and
 204         // when msg.Mbuf is freed, someone might take it into use
 205         // and payload data might be invalid inside message handle function
 206         //
 207         // subscriptions won't load system a lot so there is no
 208         // real performance hit by cloning buffer into new go byte slice
 209         cPay := append(msg.Payload[:0:0], msg.Payload...)
 210         msg.Payload = cPay
 211         msg.PayloadLen = len(cPay)
 212
 213         switch msg.Mtype {
 214         case xapp.RIC_SUB_REQ:
 215                 go c.handleXAPPSubscriptionRequest(msg)
 216         case xapp.RIC_SUB_RESP:
 217                 go c.handleE2TSubscriptionResponse(msg)
 218         case xapp.RIC_SUB_FAILURE:
 219                 go c.handleE2TSubscriptionFailure(msg)
 220         case xapp.RIC_SUB_DEL_REQ:
 221                 go c.handleXAPPSubscriptionDeleteRequest(msg)
 222         case xapp.RIC_SUB_DEL_RESP:
 223                 go c.handleE2TSubscriptionDeleteResponse(msg)
 224         case xapp.RIC_SUB_DEL_FAILURE:
 225                 go c.handleE2TSubscriptionDeleteFailure(msg)
 226         default:
 227                 xapp.Logger.Info("Unknown Message Type '%d', discarding", msg.Mtype)
 228         }
 229         return
 230 }
 231
 232 //-------------------------------------------------------------------
 233 // handle from XAPP Subscription Request
 234 //------------------------------------------------------------------
 235 func (c *Control) handleXAPPSubscriptionRequest(params *xapp.RMRParams) {
 236         xapp.Logger.Info("MSG from XAPP: %s", params.String())
 237
 238         subReqMsg, err := c.e2ap.UnpackSubscriptionRequest(params.Payload)
 239         if err != nil {
 240                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, params))
 241                 return
 242         }
 243
 244         trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subReqMsg.RequestId.InstanceId, params.Meid)
 245         if trans == nil {
 246                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 247                 return
 248         }
 249         defer trans.Release()
 250
 251         err = c.tracker.Track(trans)
 252         if err != nil {
 253                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 254                 return
 255         }
 256
 257         //TODO handle subscription toward e2term inside AssignToSubscription / hide handleSubscriptionCreate in it?
 258         subs, err := c.registry.AssignToSubscription(trans, subReqMsg)
 259         if err != nil {
 260                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 261                 return
 262         }
 263
 264         //
 265         // Wake subs request
 266         //
 267         go c.handleSubscriptionCreate(subs, trans)
 268         event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 269
 270         err = nil
 271         if event != nil {
 272                 switch themsg := event.(type) {
 273                 case *e2ap.E2APSubscriptionResponse:
 274                         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionResponse(themsg)
 275                         if err == nil {
 276                                 trans.Release()
 277                                 c.rmrSendToXapp("", subs, trans)
 278                                 return
 279                         }
 280                 case *e2ap.E2APSubscriptionFailure:
 281                         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionFailure(themsg)
 282                         if err == nil {
 283                                 c.rmrSendToXapp("", subs, trans)
 284                         }
 285                 default:
 286                         break
 287                 }
 288         }
 289         xapp.Logger.Info("XAPP-SubReq: failed %s", idstring(err, trans, subs))
 290         //c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
 291 }
 292
 293 //-------------------------------------------------------------------
 294 // handle from XAPP Subscription Delete Request
 295 //------------------------------------------------------------------
 296 func (c *Control) handleXAPPSubscriptionDeleteRequest(params *xapp.RMRParams) {
 297         xapp.Logger.Info("MSG from XAPP: %s", params.String())
 298
 299         subDelReqMsg, err := c.e2ap.UnpackSubscriptionDeleteRequest(params.Payload)
 300         if err != nil {
 301                 xapp.Logger.Error("XAPP-SubDelReq %s", idstring(err, params))
 302                 return
 303         }
 304
 305         trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subDelReqMsg.RequestId.InstanceId, params.Meid)
 306         if trans == nil {
 307                 xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 308                 return
 309         }
 310         defer trans.Release()
 311
 312         err = c.tracker.Track(trans)
 313         if err != nil {
 314                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 315                 return
 316         }
 317
 318         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{trans.GetSubId()})
 319         if err != nil {
 320                 xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(err, trans))
 321                 return
 322         }
 323
 324         //
 325         // Wake subs delete
 326         //
 327         go c.handleSubscriptionDelete(subs, trans)
 328         trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 329
 330         xapp.Logger.Debug("XAPP-SubDelReq: Handling event %s ", idstring(nil, trans, subs))
 331
 332         // Whatever is received send ok delete response
 333         subDelRespMsg := &e2ap.E2APSubscriptionDeleteResponse{}
 334         subDelRespMsg.RequestId = subs.GetReqId().RequestId
 335         subDelRespMsg.FunctionId = subs.SubReqMsg.FunctionId
 336         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteResponse(subDelRespMsg)
 337         if err == nil {
 338                 c.rmrSendToXapp("", subs, trans)
 339         }
 340
 341         //TODO handle subscription toward e2term insiged RemoveFromSubscription / hide handleSubscriptionDelete in it?
 342         //c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
 343 }
 344
 345 //-------------------------------------------------------------------
 346 // SUBS CREATE Handling
 347 //-------------------------------------------------------------------
 348 func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *TransactionXapp) {
 349
 350         trans := c.tracker.NewSubsTransaction(subs)
 351         subs.WaitTransactionTurn(trans)
 352         defer subs.ReleaseTransactionTurn(trans)
 353         defer trans.Release()
 354
 355         xapp.Logger.Debug("SUBS-SubReq: Handling %s ", idstring(nil, trans, subs, parentTrans))
 356
 357         subRfMsg, valid := subs.GetCachedResponse()
 358         if subRfMsg == nil && valid == true {
 359
 360                 //
 361                 // In case of failure
 362                 // - make internal delete
 363                 // - in case duplicate cause, retry (currently max 1 retry)
 364                 //
 365                 maxRetries := uint64(1)
 366                 doRetry := true
 367                 for retries := uint64(0); retries <= maxRetries && doRetry; retries++ {
 368                         doRetry = false
 369
 370                         event := c.sendE2TSubscriptionRequest(subs, trans, parentTrans)
 371                         switch themsg := event.(type) {
 372                         case *e2ap.E2APSubscriptionResponse:
 373                                 subRfMsg, valid = subs.SetCachedResponse(event, true)
 374                         case *e2ap.E2APSubscriptionFailure:
 375                                 subRfMsg, valid = subs.SetCachedResponse(event, false)
 376                                 doRetry = true
 377                                 for _, item := range themsg.ActionNotAdmittedList.Items {
 378                                         if item.Cause.Content != e2ap.E2AP_CauseContent_Ric || (item.Cause.Value != e2ap.E2AP_CauseValue_Ric_duplicate_action && item.Cause.Value != e2ap.E2AP_CauseValue_Ric_duplicate_event) {
 379                                                 doRetry = false
 380                                                 break
 381                                         }
 382                                 }
 383                                 xapp.Logger.Info("SUBS-SubReq: internal delete and possible retry due event(%s) retry(%t,%d/%d) %s", typeofSubsMessage(event), doRetry, retries, maxRetries, idstring(nil, trans, subs, parentTrans))
 384                                 c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 385                         default:
 386                                 xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 387                                 subRfMsg, valid = subs.SetCachedResponse(nil, false)
 388                                 c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 389                         }
 390                 }
 391
 392                 xapp.Logger.Debug("SUBS-SubReq: Handling (e2t response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
 393         } else {
 394                 xapp.Logger.Debug("SUBS-SubReq: Handling (cached response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
 395         }
 396
 397         //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
 398         if valid == false {
 399                 c.registry.RemoveFromSubscription(subs, parentTrans, 5*time.Second)
 400         }
 401         parentTrans.SendEvent(subRfMsg, 0)
 402 }
 403
 404 //-------------------------------------------------------------------
 405 // SUBS DELETE Handling
 406 //-------------------------------------------------------------------
 407
 408 func (c *Control) handleSubscriptionDelete(subs *Subscription, parentTrans *TransactionXapp) {
 409
 410         trans := c.tracker.NewSubsTransaction(subs)
 411         subs.WaitTransactionTurn(trans)
 412         defer subs.ReleaseTransactionTurn(trans)
 413         defer trans.Release()
 414
 415         xapp.Logger.Debug("SUBS-SubDelReq: Handling %s", idstring(nil, trans, subs, parentTrans))
 416
 417         subs.mutex.Lock()
 418         if subs.valid && subs.EpList.HasEndpoint(parentTrans.GetEndpoint()) && subs.EpList.Size() == 1 {
 419                 subs.valid = false
 420                 subs.mutex.Unlock()
 421                 c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 422         } else {
 423                 subs.mutex.Unlock()
 424         }
 425         //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
 426         //  If parallel deletes ongoing both might pass earlier sendE2TSubscriptionDeleteRequest(...) if
 427         //  RemoveFromSubscription locates in caller side (now in handleXAPPSubscriptionDeleteRequest(...))
 428         c.registry.RemoveFromSubscription(subs, parentTrans, 5*time.Second)
 429         parentTrans.SendEvent(nil, 0)
 430 }
 431
 432 //-------------------------------------------------------------------
 433 // send to E2T Subscription Request
 434 //-------------------------------------------------------------------
 435 func (c *Control) sendE2TSubscriptionRequest(subs *Subscription, trans *TransactionSubs, parentTrans *TransactionXapp) interface{} {
 436         var err error
 437         var event interface{} = nil
 438         var timedOut bool = false
 439
 440         subReqMsg := subs.SubReqMsg
 441         subReqMsg.RequestId = subs.GetReqId().RequestId
 442         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionRequest(subReqMsg)
 443         if err != nil {
 444                 xapp.Logger.Error("SUBS-SubReq: %s", idstring(err, trans, subs, parentTrans))
 445                 return event
 446         }
 447
 448         for retries := uint64(0); retries < e2tMaxSubReqTryCount; retries++ {
 449                 desc := fmt.Sprintf("(retry %d)", retries)
 450                 c.rmrSendToE2T(desc, subs, trans)
 451                 event, timedOut = trans.WaitEvent(e2tSubReqTimeout)
 452                 if timedOut {
 453                         continue
 454                 }
 455                 break
 456         }
 457         xapp.Logger.Debug("SUBS-SubReq: Response handling event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 458         return event
 459 }
 460
 461 //-------------------------------------------------------------------
 462 // send to E2T Subscription Delete Request
 463 //-------------------------------------------------------------------
 464
 465 func (c *Control) sendE2TSubscriptionDeleteRequest(subs *Subscription, trans *TransactionSubs, parentTrans *TransactionXapp) interface{} {
 466         var err error
 467         var event interface{}
 468         var timedOut bool
 469
 470         subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{}
 471         subDelReqMsg.RequestId = subs.GetReqId().RequestId
 472         subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId
 473         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg)
 474         if err != nil {
 475                 xapp.Logger.Error("SUBS-SubDelReq: %s", idstring(err, trans, subs, parentTrans))
 476                 return event
 477         }
 478
 479         for retries := uint64(0); retries < e2tMaxSubDelReqTryCount; retries++ {
 480                 desc := fmt.Sprintf("(retry %d)", retries)
 481                 c.rmrSendToE2T(desc, subs, trans)
 482                 event, timedOut = trans.WaitEvent(e2tSubDelReqTime)
 483                 if timedOut {
 484                         continue
 485                 }
 486                 break
 487         }
 488         xapp.Logger.Debug("SUBS-SubDelReq: Response handling event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 489         return event
 490 }
 491
 492 //-------------------------------------------------------------------
 493 // handle from E2T Subscription Reponse
 494 //-------------------------------------------------------------------
 495 func (c *Control) handleE2TSubscriptionResponse(params *xapp.RMRParams) {
 496         xapp.Logger.Info("MSG from E2T: %s", params.String())
 497         subRespMsg, err := c.e2ap.UnpackSubscriptionResponse(params.Payload)
 498         if err != nil {
 499                 xapp.Logger.Error("MSG-SubResp %s", idstring(err, params))
 500                 return
 501         }
 502         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subRespMsg.RequestId.InstanceId})
 503         if err != nil {
 504                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, params))
 505                 return
 506         }
 507         trans := subs.GetTransaction()
 508         if trans == nil {
 509                 err = fmt.Errorf("Ongoing transaction not found")
 510                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, params, subs))
 511                 return
 512         }
 513         sendOk, timedOut := trans.SendEvent(subRespMsg, e2tRecvMsgTimeout)
 514         if sendOk == false {
 515                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 516                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, trans, subs))
 517         }
 518         return
 519 }
 520
 521 //-------------------------------------------------------------------
 522 // handle from E2T Subscription Failure
 523 //-------------------------------------------------------------------
 524 func (c *Control) handleE2TSubscriptionFailure(params *xapp.RMRParams) {
 525         xapp.Logger.Info("MSG from E2T: %s", params.String())
 526         subFailMsg, err := c.e2ap.UnpackSubscriptionFailure(params.Payload)
 527         if err != nil {
 528                 xapp.Logger.Error("MSG-SubFail %s", idstring(err, params))
 529                 return
 530         }
 531         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subFailMsg.RequestId.InstanceId})
 532         if err != nil {
 533                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, params))
 534                 return
 535         }
 536         trans := subs.GetTransaction()
 537         if trans == nil {
 538                 err = fmt.Errorf("Ongoing transaction not found")
 539                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, params, subs))
 540                 return
 541         }
 542         sendOk, timedOut := trans.SendEvent(subFailMsg, e2tRecvMsgTimeout)
 543         if sendOk == false {
 544                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 545                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, trans, subs))
 546         }
 547         return
 548 }
 549
 550 //-------------------------------------------------------------------
 551 // handle from E2T Subscription Delete Response
 552 //-------------------------------------------------------------------
 553 func (c *Control) handleE2TSubscriptionDeleteResponse(params *xapp.RMRParams) (err error) {
 554         xapp.Logger.Info("MSG from E2T: %s", params.String())
 555         subDelRespMsg, err := c.e2ap.UnpackSubscriptionDeleteResponse(params.Payload)
 556         if err != nil {
 557                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params))
 558                 return
 559         }
 560         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelRespMsg.RequestId.InstanceId})
 561         if err != nil {
 562                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params))
 563                 return
 564         }
 565         trans := subs.GetTransaction()
 566         if trans == nil {
 567                 err = fmt.Errorf("Ongoing transaction not found")
 568                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params, subs))
 569                 return
 570         }
 571         sendOk, timedOut := trans.SendEvent(subDelRespMsg, e2tRecvMsgTimeout)
 572         if sendOk == false {
 573                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 574                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, trans, subs))
 575         }
 576         return
 577 }
 578
 579 //-------------------------------------------------------------------
 580 // handle from E2T Subscription Delete Failure
 581 //-------------------------------------------------------------------
 582 func (c *Control) handleE2TSubscriptionDeleteFailure(params *xapp.RMRParams) {
 583         xapp.Logger.Info("MSG from E2T: %s", params.String())
 584         subDelFailMsg, err := c.e2ap.UnpackSubscriptionDeleteFailure(params.Payload)
 585         if err != nil {
 586                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
 587                 return
 588         }
 589         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelFailMsg.RequestId.InstanceId})
 590         if err != nil {
 591                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
 592                 return
 593         }
 594         trans := subs.GetTransaction()
 595         if trans == nil {
 596                 err = fmt.Errorf("Ongoing transaction not found")
 597                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params, subs))
 598                 return
 599         }
 600         sendOk, timedOut := trans.SendEvent(subDelFailMsg, e2tRecvMsgTimeout)
 601         if sendOk == false {
 602                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 603                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, trans, subs))
 604         }
 605         return
 606 }
 607
 608 //-------------------------------------------------------------------
 609 //
 610 //-------------------------------------------------------------------
 611 func typeofSubsMessage(v interface{}) string {
 612         if v == nil {
 613                 return "NIL"
 614         }
 615         switch v.(type) {
 616         case *e2ap.E2APSubscriptionRequest:
 617                 return "SubReq"
 618         case *e2ap.E2APSubscriptionResponse:
 619                 return "SubResp"
 620         case *e2ap.E2APSubscriptionFailure:
 621                 return "SubFail"
 622         case *e2ap.E2APSubscriptionDeleteRequest:
 623                 return "SubDelReq"
 624         case *e2ap.E2APSubscriptionDeleteResponse:
 625                 return "SubDelResp"
 626         case *e2ap.E2APSubscriptionDeleteFailure:
 627                 return "SubDelFail"
 628         default:
 629                 return "Unknown"
 630         }
 631 }