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