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         "net/http"
  25         "os"
  26         "strconv"
  27         "strings"
  28         "time"
  29
  30         "gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap"
  31         rtmgrclient "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/rtmgr_client"
  32         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models"
  33         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  34         httptransport "github.com/go-openapi/runtime/client"
  35         "github.com/go-openapi/strfmt"
  36         "github.com/gorilla/mux"
  37         "github.com/segmentio/ksuid"
  38         "github.com/spf13/viper"
  39 )
  40
  41 //-----------------------------------------------------------------------------
  42 //
  43 //-----------------------------------------------------------------------------
  44
  45 func idstring(err error, entries ...fmt.Stringer) string {
  46         var retval string = ""
  47         var filler string = ""
  48         for _, entry := range entries {
  49                 retval += filler + entry.String()
  50                 filler = " "
  51         }
  52         if err != nil {
  53                 retval += filler + "err(" + err.Error() + ")"
  54                 filler = " "
  55         }
  56         return retval
  57 }
  58
  59 //-----------------------------------------------------------------------------
  60 //
  61 //-----------------------------------------------------------------------------
  62
  63 var e2tSubReqTimeout time.Duration
  64 var e2tSubDelReqTime time.Duration
  65 var e2tRecvMsgTimeout time.Duration
  66 var waitRouteCleanup_ms time.Duration
  67 var e2tMaxSubReqTryCount uint64    // Initial try + retry
  68 var e2tMaxSubDelReqTryCount uint64 // Initial try + retry
  69 var readSubsFromDb string
  70
  71 type Control struct {
  72         *xapp.RMRClient
  73         e2ap          *E2ap
  74         registry      *Registry
  75         tracker       *Tracker
  76         db            Sdlnterface
  77         CntRecvMsg    uint64
  78         ResetTestFlag bool
  79         Counters      map[string]xapp.Counter
  80         LoggerLevel   uint32
  81 }
  82
  83 type RMRMeid struct {
  84         PlmnID  string
  85         EnbID   string
  86         RanName string
  87 }
  88
  89 type SubmgrRestartTestEvent struct{}
  90 type SubmgrRestartUpEvent struct{}
  91
  92 func init() {
  93         xapp.Logger.Info("SUBMGR")
  94         viper.AutomaticEnv()
  95         viper.SetEnvPrefix("submgr")
  96         viper.AllowEmptyEnv(true)
  97 }
  98
  99 func NewControl() *Control {
 100
 101         transport := httptransport.New(viper.GetString("rtmgr.HostAddr")+":"+viper.GetString("rtmgr.port"), viper.GetString("rtmgr.baseUrl"), []string{"http"})
 102         rtmgrClient := RtmgrClient{rtClient: rtmgrclient.New(transport, strfmt.Default)}
 103
 104         registry := new(Registry)
 105         registry.Initialize()
 106         registry.rtmgrClient = &rtmgrClient
 107
 108         tracker := new(Tracker)
 109         tracker.Init()
 110
 111         c := &Control{e2ap: new(E2ap),
 112                 registry:    registry,
 113                 tracker:     tracker,
 114                 db:          CreateSdl(),
 115                 Counters:    xapp.Metric.RegisterCounterGroup(GetMetricsOpts(), "SUBMGR"),
 116                 LoggerLevel: 3,
 117         }
 118         c.ReadConfigParameters("")
 119
 120         // Register REST handler for testing support
 121         xapp.Resource.InjectRoute("/ric/v1/test/{testId}", c.TestRestHandler, "POST")
 122         xapp.Resource.InjectRoute("/ric/v1/symptomdata", c.SymptomDataHandler, "GET")
 123
 124         go xapp.Subscription.Listen(c.SubscriptionHandler, c.QueryHandler, c.SubscriptionDeleteHandlerCB)
 125
 126         if readSubsFromDb == "false" {
 127                 return c
 128         }
 129
 130         // Read subscriptions from db
 131         xapp.Logger.Info("Reading subscriptions from db")
 132         subIds, register, err := c.ReadAllSubscriptionsFromSdl()
 133         if err != nil {
 134                 xapp.Logger.Error("%v", err)
 135         } else {
 136                 c.registry.subIds = subIds
 137                 c.registry.register = register
 138                 c.HandleUncompletedSubscriptions(register)
 139         }
 140         return c
 141 }
 142
 143 func (c *Control) SymptomDataHandler(w http.ResponseWriter, r *http.Request) {
 144         subscriptions, _ := c.registry.QueryHandler()
 145         xapp.Resource.SendSymptomDataJson(w, r, subscriptions, "platform/subscriptions.json")
 146 }
 147
 148 //-------------------------------------------------------------------
 149 //
 150 //-------------------------------------------------------------------
 151 func (c *Control) ReadConfigParameters(f string) {
 152
 153         // viper.GetDuration returns nanoseconds
 154         e2tSubReqTimeout = viper.GetDuration("controls.e2tSubReqTimeout_ms") * 1000000
 155         if e2tSubReqTimeout == 0 {
 156                 e2tSubReqTimeout = 2000 * 1000000
 157         }
 158         xapp.Logger.Info("e2tSubReqTimeout %v", e2tSubReqTimeout)
 159         e2tSubDelReqTime = viper.GetDuration("controls.e2tSubDelReqTime_ms") * 1000000
 160         if e2tSubDelReqTime == 0 {
 161                 e2tSubDelReqTime = 2000 * 1000000
 162         }
 163         xapp.Logger.Info("e2tSubDelReqTime %v", e2tSubDelReqTime)
 164         e2tRecvMsgTimeout = viper.GetDuration("controls.e2tRecvMsgTimeout_ms") * 1000000
 165         if e2tRecvMsgTimeout == 0 {
 166                 e2tRecvMsgTimeout = 2000 * 1000000
 167         }
 168         xapp.Logger.Info("e2tRecvMsgTimeout %v", e2tRecvMsgTimeout)
 169
 170         // Internal cfg parameter, used to define a wait time for RMR route clean-up. None default
 171         // value 100ms used currently only in unittests.
 172         waitRouteCleanup_ms = viper.GetDuration("controls.waitRouteCleanup_ms") * 1000000
 173         if waitRouteCleanup_ms == 0 {
 174                 waitRouteCleanup_ms = 5000 * 1000000
 175         }
 176         xapp.Logger.Info("waitRouteCleanup %v", waitRouteCleanup_ms)
 177
 178         e2tMaxSubReqTryCount = viper.GetUint64("controls.e2tMaxSubReqTryCount")
 179         if e2tMaxSubReqTryCount == 0 {
 180                 e2tMaxSubReqTryCount = 1
 181         }
 182         xapp.Logger.Info("e2tMaxSubReqTryCount %v", e2tMaxSubReqTryCount)
 183
 184         e2tMaxSubDelReqTryCount = viper.GetUint64("controls.e2tMaxSubDelReqTryCount")
 185         if e2tMaxSubDelReqTryCount == 0 {
 186                 e2tMaxSubDelReqTryCount = 1
 187         }
 188         xapp.Logger.Info("e2tMaxSubDelReqTryCount %v", e2tMaxSubDelReqTryCount)
 189
 190         readSubsFromDb = viper.GetString("controls.readSubsFromDb")
 191         if readSubsFromDb == "" {
 192                 readSubsFromDb = "true"
 193         }
 194         xapp.Logger.Info("readSubsFromDb %v", readSubsFromDb)
 195         c.LoggerLevel = viper.GetUint32("logger.level")
 196         if c.LoggerLevel == 0 {
 197                 c.LoggerLevel = 3
 198         }
 199 }
 200
 201 //-------------------------------------------------------------------
 202 //
 203 //-------------------------------------------------------------------
 204 func (c *Control) HandleUncompletedSubscriptions(register map[uint32]*Subscription) {
 205
 206         xapp.Logger.Debug("HandleUncompletedSubscriptions. len(register) = %v", len(register))
 207         for subId, subs := range register {
 208                 if subs.SubRespRcvd == false {
 209                         subs.NoRespToXapp = true
 210                         xapp.Logger.Debug("SendSubscriptionDeleteReq. subId = %v", subId)
 211                         c.SendSubscriptionDeleteReq(subs)
 212                 }
 213         }
 214 }
 215
 216 func (c *Control) ReadyCB(data interface{}) {
 217         if c.RMRClient == nil {
 218                 c.RMRClient = xapp.Rmr
 219         }
 220 }
 221
 222 func (c *Control) Run() {
 223         xapp.SetReadyCB(c.ReadyCB, nil)
 224         xapp.AddConfigChangeListener(c.ReadConfigParameters)
 225         xapp.Run(c)
 226 }
 227
 228 //-------------------------------------------------------------------
 229 //
 230 //-------------------------------------------------------------------
 231 func (c *Control) SubscriptionHandler(params interface{}) (*models.SubscriptionResponse, error) {
 232
 233         c.CntRecvMsg++
 234         c.UpdateCounter(cRestSubReqFromXapp)
 235
 236         subResp := models.SubscriptionResponse{}
 237         p := params.(*models.SubscriptionParams)
 238
 239         if c.LoggerLevel > 2 {
 240                 c.PrintRESTSubscriptionRequest(p)
 241         }
 242
 243         if p.ClientEndpoint == nil {
 244                 xapp.Logger.Error("ClientEndpoint == nil")
 245                 c.UpdateCounter(cRestSubFailToXapp)
 246                 return nil, fmt.Errorf("")
 247         }
 248
 249         _, xAppRmrEndpoint, err := ConstructEndpointAddresses(*p.ClientEndpoint)
 250         if err != nil {
 251                 xapp.Logger.Error("%s", err.Error())
 252                 c.UpdateCounter(cRestSubFailToXapp)
 253                 return nil, err
 254         }
 255         var restSubId string
 256         var restSubscription *RESTSubscription
 257         if p.SubscriptionID == "" {
 258                 restSubId = ksuid.New().String()
 259                 restSubscription, err = c.registry.CreateRESTSubscription(&restSubId, &xAppRmrEndpoint, p.Meid)
 260                 if err != nil {
 261                         xapp.Logger.Error("%s", err.Error())
 262                         c.UpdateCounter(cRestSubFailToXapp)
 263                         return nil, err
 264                 }
 265
 266         } else {
 267                 restSubId = p.SubscriptionID
 268                 restSubscription, err = c.registry.GetRESTSubscription(restSubId, false)
 269                 if err != nil {
 270                         xapp.Logger.Error("%s", err.Error())
 271                         c.UpdateCounter(cRestSubFailToXapp)
 272                         return nil, err
 273                 }
 274         }
 275
 276         subResp.SubscriptionID = &restSubId
 277         subReqList := e2ap.SubscriptionRequestList{}
 278         err = c.e2ap.FillSubscriptionReqMsgs(params, &subReqList, restSubscription)
 279         if err != nil {
 280                 xapp.Logger.Error("%s", err.Error())
 281                 c.registry.DeleteRESTSubscription(&restSubId)
 282                 c.UpdateCounter(cRestSubFailToXapp)
 283                 return nil, err
 284         }
 285
 286         go c.processSubscriptionRequests(restSubscription, &subReqList, p.ClientEndpoint, p.Meid, &restSubId)
 287
 288         c.UpdateCounter(cRestSubRespToXapp)
 289         return &subResp, nil
 290 }
 291
 292 //-------------------------------------------------------------------
 293 //
 294 //-------------------------------------------------------------------
 295
 296 func (c *Control) processSubscriptionRequests(restSubscription *RESTSubscription, subReqList *e2ap.SubscriptionRequestList,
 297         clientEndpoint *models.SubscriptionParamsClientEndpoint, meid *string, restSubId *string) {
 298
 299         xapp.Logger.Info("Subscription Request count=%v ", len(subReqList.E2APSubscriptionRequests))
 300
 301         _, xAppRmrEndpoint, err := ConstructEndpointAddresses(*clientEndpoint)
 302         if err != nil {
 303                 xapp.Logger.Error("%s", err.Error())
 304                 return
 305         }
 306
 307         var xAppEventInstanceID int64
 308         var e2EventInstanceID int64
 309         for index := 0; index < len(subReqList.E2APSubscriptionRequests); index++ {
 310                 subReqMsg := subReqList.E2APSubscriptionRequests[index]
 311
 312                 trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(xAppRmrEndpoint), *restSubId, subReqMsg.RequestId, &xapp.RMRMeid{RanName: *meid})
 313                 if trans == nil {
 314                         c.registry.DeleteRESTSubscription(restSubId)
 315                         xapp.Logger.Error("XAPP-SubReq transaction not created. RESTSubId=%s, EndPoint=%s, Meid=%s", *restSubId, xAppRmrEndpoint, *meid)
 316                         return
 317                 }
 318
 319                 defer trans.Release()
 320                 xAppEventInstanceID = (int64)(subReqMsg.RequestId.Id)
 321                 xapp.Logger.Info("Handle SubscriptionRequest index=%v, %s", index, idstring(nil, trans))
 322                 subRespMsg, err := c.handleSubscriptionRequest(trans, &subReqMsg, meid, restSubId)
 323                 if err != nil {
 324                         // Send notification to xApp that prosessing of a Subscription Request has failed.
 325                         e2EventInstanceID = (int64)(0)
 326                         resp := &models.SubscriptionResponse{
 327                                 SubscriptionID: restSubId,
 328                                 SubscriptionInstances: []*models.SubscriptionInstance{
 329                                         &models.SubscriptionInstance{E2EventInstanceID: &e2EventInstanceID,
 330                                                 ErrorCause:          nil, //TODO: Suitable Error cause.
 331                                                 XappEventInstanceID: &xAppEventInstanceID},
 332                                 },
 333                         }
 334                         // Mark REST subscription request processed.
 335                         restSubscription.SetProcessed()
 336                         xapp.Logger.Info("Sending unsuccessful REST notification to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
 337                                 clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
 338                         xapp.Subscription.Notify(resp, *clientEndpoint)
 339                         c.UpdateCounter(cRestSubFailNotifToXapp)
 340                 } else {
 341                         e2EventInstanceID = (int64)(subRespMsg.RequestId.InstanceId)
 342
 343                         xapp.Logger.Info("SubscriptionRequest index=%v processed successfully. endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
 344                                 index, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
 345
 346                         // Store successfully processed InstanceId for deletion
 347                         restSubscription.AddE2InstanceId(subRespMsg.RequestId.InstanceId)
 348                         restSubscription.AddXappIdToE2Id(xAppEventInstanceID, e2EventInstanceID)
 349
 350                         // Send notification to xApp that a Subscription Request has been processed.
 351                         resp := &models.SubscriptionResponse{
 352                                 SubscriptionID: restSubId,
 353                                 SubscriptionInstances: []*models.SubscriptionInstance{
 354                                         &models.SubscriptionInstance{E2EventInstanceID: &e2EventInstanceID,
 355                                                 ErrorCause:          nil,
 356                                                 XappEventInstanceID: &xAppEventInstanceID},
 357                                 },
 358                         }
 359                         // Mark REST subscription request processesd.
 360                         restSubscription.SetProcessed()
 361                         xapp.Logger.Info("Sending successful REST notification to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s",
 362                                 clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans))
 363                         xapp.Subscription.Notify(resp, *clientEndpoint)
 364                         c.UpdateCounter(cRestSubNotifToXapp)
 365
 366                 }
 367         }
 368 }
 369
 370 //-------------------------------------------------------------------
 371 //
 372 //------------------------------------------------------------------
 373 func (c *Control) handleSubscriptionRequest(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest, meid *string,
 374         restSubId *string) (*e2ap.E2APSubscriptionResponse, error) {
 375
 376         err := c.tracker.Track(trans)
 377         if err != nil {
 378                 err = fmt.Errorf("XAPP-SubReq: %s", idstring(err, trans))
 379                 xapp.Logger.Error("%s", err.Error())
 380                 return nil, err
 381         }
 382
 383         subs, err := c.registry.AssignToSubscription(trans, subReqMsg, c.ResetTestFlag, c)
 384         if err != nil {
 385                 err = fmt.Errorf("XAPP-SubReq: %s", idstring(err, trans))
 386                 xapp.Logger.Error("%s", err.Error())
 387                 return nil, err
 388         }
 389
 390         //
 391         // Wake subs request
 392         //
 393         go c.handleSubscriptionCreate(subs, trans)
 394         event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 395
 396         err = nil
 397         if event != nil {
 398                 switch themsg := event.(type) {
 399                 case *e2ap.E2APSubscriptionResponse:
 400                         trans.Release()
 401                         return themsg, nil
 402                 case *e2ap.E2APSubscriptionFailure:
 403                         err = fmt.Errorf("SubscriptionFailure received")
 404                         return nil, err
 405                 default:
 406                         break
 407                 }
 408         }
 409         err = fmt.Errorf("XAPP-SubReq: failed %s", idstring(err, trans, subs))
 410         xapp.Logger.Error("%s", err.Error())
 411         c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
 412         return nil, err
 413 }
 414
 415 //-------------------------------------------------------------------
 416 //
 417 //-------------------------------------------------------------------
 418 func (c *Control) SubscriptionDeleteHandlerCB(restSubId string) error {
 419
 420         c.CntRecvMsg++
 421         c.UpdateCounter(cRestSubDelReqFromXapp)
 422
 423         xapp.Logger.Info("SubscriptionDeleteRequest from XAPP")
 424
 425         restSubscription, err := c.registry.GetRESTSubscription(restSubId, true)
 426         if err != nil {
 427                 xapp.Logger.Error("%s", err.Error())
 428                 if restSubscription == nil {
 429                         // Subscription was not found
 430                         return nil
 431                 } else {
 432                         if restSubscription.SubReqOngoing == true {
 433                                 err := fmt.Errorf("Handling of the REST Subscription Request still ongoing %s", restSubId)
 434                                 xapp.Logger.Error("%s", err.Error())
 435                                 return err
 436                         } else if restSubscription.SubDelReqOngoing == true {
 437                                 // Previous request for same restSubId still ongoing
 438                                 return nil
 439                         }
 440                 }
 441         }
 442
 443         xAppRmrEndPoint := restSubscription.xAppRmrEndPoint
 444         go func() {
 445                 for _, instanceId := range restSubscription.InstanceIds {
 446                         xAppEventInstanceID, err := c.SubscriptionDeleteHandler(&restSubId, &xAppRmrEndPoint, &restSubscription.Meid, instanceId)
 447
 448                         if err != nil {
 449                                 xapp.Logger.Error("%s", err.Error())
 450                                 //return err
 451                         }
 452                         xapp.Logger.Info("Deleteting instanceId = %v", instanceId)
 453                         restSubscription.DeleteXappIdToE2Id(xAppEventInstanceID)
 454                         restSubscription.DeleteE2InstanceId(instanceId)
 455                 }
 456                 c.registry.DeleteRESTSubscription(&restSubId)
 457         }()
 458
 459         c.UpdateCounter(cRestSubDelRespToXapp)
 460
 461         return nil
 462 }
 463
 464 //-------------------------------------------------------------------
 465 //
 466 //-------------------------------------------------------------------
 467 func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string, meid *string, instanceId uint32) (int64, error) {
 468
 469         var xAppEventInstanceID int64
 470         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{instanceId})
 471         if err != nil {
 472                 xapp.Logger.Info("Subscription Delete Handler subscription for restSubId=%v, E2EventInstanceID=%v not found %s",
 473                         restSubId, instanceId, idstring(err, nil))
 474                 return xAppEventInstanceID, nil
 475         }
 476
 477         xAppEventInstanceID = int64(subs.ReqId.Id)
 478         trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(*endPoint), *restSubId, e2ap.RequestId{subs.ReqId.Id, 0}, &xapp.RMRMeid{RanName: *meid})
 479         if trans == nil {
 480                 err := fmt.Errorf("XAPP-SubDelReq transaction not created. restSubId %s, endPoint %s, meid %s, instanceId %v", *restSubId, *endPoint, *meid, instanceId)
 481                 xapp.Logger.Error("%s", err.Error())
 482         }
 483         defer trans.Release()
 484
 485         err = c.tracker.Track(trans)
 486         if err != nil {
 487                 err := fmt.Errorf("XAPP-SubDelReq %s:", idstring(err, trans))
 488                 xapp.Logger.Error("%s", err.Error())
 489                 return xAppEventInstanceID, &time.ParseError{}
 490         }
 491         //
 492         // Wake subs delete
 493         //
 494         go c.handleSubscriptionDelete(subs, trans)
 495         trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 496
 497         xapp.Logger.Debug("XAPP-SubDelReq: Handling event %s ", idstring(nil, trans, subs))
 498
 499         c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
 500
 501         return xAppEventInstanceID, nil
 502 }
 503
 504 //-------------------------------------------------------------------
 505 //
 506 //-------------------------------------------------------------------
 507 func (c *Control) QueryHandler() (models.SubscriptionList, error) {
 508         xapp.Logger.Info("QueryHandler() called")
 509
 510         c.CntRecvMsg++
 511
 512         return c.registry.QueryHandler()
 513 }
 514
 515 func (c *Control) TestRestHandler(w http.ResponseWriter, r *http.Request) {
 516         xapp.Logger.Info("TestRestHandler() called")
 517
 518         pathParams := mux.Vars(r)
 519         s := pathParams["testId"]
 520
 521         // This can be used to delete single subscription from db
 522         if contains := strings.Contains(s, "deletesubid="); contains == true {
 523                 var splits = strings.Split(s, "=")
 524                 if subId, err := strconv.ParseInt(splits[1], 10, 64); err == nil {
 525                         xapp.Logger.Info("RemoveSubscriptionFromSdl() called. subId = %v", subId)
 526                         c.RemoveSubscriptionFromSdl(uint32(subId))
 527                         return
 528                 }
 529         }
 530
 531         // This can be used to remove all subscriptions db from
 532         if s == "emptydb" {
 533                 xapp.Logger.Info("RemoveAllSubscriptionsFromSdl() called")
 534                 c.RemoveAllSubscriptionsFromSdl()
 535                 return
 536         }
 537
 538         // This is meant to cause submgr's restart in testing
 539         if s == "restart" {
 540                 xapp.Logger.Info("os.Exit(1) called")
 541                 os.Exit(1)
 542         }
 543
 544         xapp.Logger.Info("Unsupported rest command received %s", s)
 545 }
 546
 547 //-------------------------------------------------------------------
 548 //
 549 //-------------------------------------------------------------------
 550
 551 func (c *Control) rmrSendToE2T(desc string, subs *Subscription, trans *TransactionSubs) (err error) {
 552         params := &xapp.RMRParams{}
 553         params.Mtype = trans.GetMtype()
 554         params.SubId = int(subs.GetReqId().InstanceId)
 555         params.Xid = ""
 556         params.Meid = subs.GetMeid()
 557         params.Src = ""
 558         params.PayloadLen = len(trans.Payload.Buf)
 559         params.Payload = trans.Payload.Buf
 560         params.Mbuf = nil
 561         xapp.Logger.Info("MSG to E2T: %s %s %s", desc, trans.String(), params.String())
 562         err = c.SendWithRetry(params, false, 5)
 563         if err != nil {
 564                 xapp.Logger.Error("rmrSendToE2T: Send failed: %+v", err)
 565         }
 566         return err
 567 }
 568
 569 func (c *Control) rmrSendToXapp(desc string, subs *Subscription, trans *TransactionXapp) (err error) {
 570
 571         params := &xapp.RMRParams{}
 572         params.Mtype = trans.GetMtype()
 573         params.SubId = int(subs.GetReqId().InstanceId)
 574         params.Xid = trans.GetXid()
 575         params.Meid = trans.GetMeid()
 576         params.Src = ""
 577         params.PayloadLen = len(trans.Payload.Buf)
 578         params.Payload = trans.Payload.Buf
 579         params.Mbuf = nil
 580         xapp.Logger.Info("MSG to XAPP: %s %s %s", desc, trans.String(), params.String())
 581         err = c.SendWithRetry(params, false, 5)
 582         if err != nil {
 583                 xapp.Logger.Error("rmrSendToXapp: Send failed: %+v", err)
 584         }
 585         return err
 586 }
 587
 588 func (c *Control) Consume(msg *xapp.RMRParams) (err error) {
 589         if c.RMRClient == nil {
 590                 err = fmt.Errorf("Rmr object nil can handle %s", msg.String())
 591                 xapp.Logger.Error("%s", err.Error())
 592                 return
 593         }
 594         c.CntRecvMsg++
 595
 596         defer c.RMRClient.Free(msg.Mbuf)
 597
 598         // xapp-frame might use direct access to c buffer and
 599         // when msg.Mbuf is freed, someone might take it into use
 600         // and payload data might be invalid inside message handle function
 601         //
 602         // subscriptions won't load system a lot so there is no
 603         // real performance hit by cloning buffer into new go byte slice
 604         cPay := append(msg.Payload[:0:0], msg.Payload...)
 605         msg.Payload = cPay
 606         msg.PayloadLen = len(cPay)
 607
 608         switch msg.Mtype {
 609         case xapp.RIC_SUB_REQ:
 610                 go c.handleXAPPSubscriptionRequest(msg)
 611         case xapp.RIC_SUB_RESP:
 612                 go c.handleE2TSubscriptionResponse(msg)
 613         case xapp.RIC_SUB_FAILURE:
 614                 go c.handleE2TSubscriptionFailure(msg)
 615         case xapp.RIC_SUB_DEL_REQ:
 616                 go c.handleXAPPSubscriptionDeleteRequest(msg)
 617         case xapp.RIC_SUB_DEL_RESP:
 618                 go c.handleE2TSubscriptionDeleteResponse(msg)
 619         case xapp.RIC_SUB_DEL_FAILURE:
 620                 go c.handleE2TSubscriptionDeleteFailure(msg)
 621         default:
 622                 xapp.Logger.Info("Unknown Message Type '%d', discarding", msg.Mtype)
 623         }
 624         return
 625 }
 626
 627 //-------------------------------------------------------------------
 628 // handle from XAPP Subscription Request
 629 //------------------------------------------------------------------
 630 func (c *Control) handleXAPPSubscriptionRequest(params *xapp.RMRParams) {
 631         xapp.Logger.Info("MSG from XAPP: %s", params.String())
 632         c.UpdateCounter(cSubReqFromXapp)
 633
 634         subReqMsg, err := c.e2ap.UnpackSubscriptionRequest(params.Payload)
 635         if err != nil {
 636                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, params))
 637                 return
 638         }
 639
 640         trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subReqMsg.RequestId, params.Meid)
 641         if trans == nil {
 642                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 643                 return
 644         }
 645         defer trans.Release()
 646
 647         if err = c.tracker.Track(trans); err != nil {
 648                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 649                 return
 650         }
 651
 652         //TODO handle subscription toward e2term inside AssignToSubscription / hide handleSubscriptionCreate in it?
 653         subs, err := c.registry.AssignToSubscription(trans, subReqMsg, c.ResetTestFlag, c)
 654         if err != nil {
 655                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 656                 return
 657         }
 658
 659         c.wakeSubscriptionRequest(subs, trans)
 660 }
 661
 662 //-------------------------------------------------------------------
 663 // Wake Subscription Request to E2node
 664 //------------------------------------------------------------------
 665 func (c *Control) wakeSubscriptionRequest(subs *Subscription, trans *TransactionXapp) {
 666
 667         go c.handleSubscriptionCreate(subs, trans)
 668         event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 669         var err error
 670         if event != nil {
 671                 switch themsg := event.(type) {
 672                 case *e2ap.E2APSubscriptionResponse:
 673                         themsg.RequestId.Id = trans.RequestId.Id
 674                         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionResponse(themsg)
 675                         if err == nil {
 676                                 trans.Release()
 677                                 c.UpdateCounter(cSubRespToXapp)
 678                                 c.rmrSendToXapp("", subs, trans)
 679                                 return
 680                         }
 681                 case *e2ap.E2APSubscriptionFailure:
 682                         themsg.RequestId.Id = trans.RequestId.Id
 683                         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionFailure(themsg)
 684                         if err == nil {
 685                                 c.UpdateCounter(cSubFailToXapp)
 686                                 c.rmrSendToXapp("", subs, trans)
 687                         }
 688                 default:
 689                         break
 690                 }
 691         }
 692         xapp.Logger.Info("XAPP-SubReq: failed %s", idstring(err, trans, subs))
 693         //c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
 694 }
 695
 696 //-------------------------------------------------------------------
 697 // handle from XAPP Subscription Delete Request
 698 //------------------------------------------------------------------
 699 func (c *Control) handleXAPPSubscriptionDeleteRequest(params *xapp.RMRParams) {
 700         xapp.Logger.Info("MSG from XAPP: %s", params.String())
 701         c.UpdateCounter(cSubDelReqFromXapp)
 702
 703         subDelReqMsg, err := c.e2ap.UnpackSubscriptionDeleteRequest(params.Payload)
 704         if err != nil {
 705                 xapp.Logger.Error("XAPP-SubDelReq %s", idstring(err, params))
 706                 return
 707         }
 708
 709         trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subDelReqMsg.RequestId, params.Meid)
 710         if trans == nil {
 711                 xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 712                 return
 713         }
 714         defer trans.Release()
 715
 716         err = c.tracker.Track(trans)
 717         if err != nil {
 718                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 719                 return
 720         }
 721
 722         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{trans.GetSubId()})
 723         if err != nil {
 724                 xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(err, trans))
 725                 return
 726         }
 727
 728         //
 729         // Wake subs delete
 730         //
 731         go c.handleSubscriptionDelete(subs, trans)
 732         trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 733
 734         xapp.Logger.Debug("XAPP-SubDelReq: Handling event %s ", idstring(nil, trans, subs))
 735
 736         if subs.NoRespToXapp == true {
 737                 // Do no send delete responses to xapps due to submgr restart is deleting uncompleted subscriptions
 738                 return
 739         }
 740
 741         // Whatever is received success, fail or timeout, send successful delete response
 742         subDelRespMsg := &e2ap.E2APSubscriptionDeleteResponse{}
 743         subDelRespMsg.RequestId.Id = trans.RequestId.Id
 744         subDelRespMsg.RequestId.InstanceId = subs.GetReqId().RequestId.InstanceId
 745         subDelRespMsg.FunctionId = subs.SubReqMsg.FunctionId
 746         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteResponse(subDelRespMsg)
 747         if err == nil {
 748                 c.UpdateCounter(cSubDelRespToXapp)
 749                 c.rmrSendToXapp("", subs, trans)
 750         }
 751
 752         //TODO handle subscription toward e2term insiged RemoveFromSubscription / hide handleSubscriptionDelete in it?
 753         //c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
 754 }
 755
 756 //-------------------------------------------------------------------
 757 // SUBS CREATE Handling
 758 //-------------------------------------------------------------------
 759 func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *TransactionXapp) {
 760
 761         var removeSubscriptionFromDb bool = false
 762         trans := c.tracker.NewSubsTransaction(subs)
 763         subs.WaitTransactionTurn(trans)
 764         defer subs.ReleaseTransactionTurn(trans)
 765         defer trans.Release()
 766
 767         xapp.Logger.Debug("SUBS-SubReq: Handling %s ", idstring(nil, trans, subs, parentTrans))
 768
 769         subRfMsg, valid := subs.GetCachedResponse()
 770         if subRfMsg == nil && valid == true {
 771                 event := c.sendE2TSubscriptionRequest(subs, trans, parentTrans)
 772                 switch event.(type) {
 773                 case *e2ap.E2APSubscriptionResponse:
 774                         subRfMsg, valid = subs.SetCachedResponse(event, true)
 775                         subs.SubRespRcvd = true
 776                 case *e2ap.E2APSubscriptionFailure:
 777                         removeSubscriptionFromDb = true
 778                         subRfMsg, valid = subs.SetCachedResponse(event, false)
 779                         xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 780                         c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 781                 case *SubmgrRestartTestEvent:
 782                         // This simulates that no response has been received and after restart subscriptions are restored from db
 783                         xapp.Logger.Debug("Test restart flag is active. Dropping this transaction to test restart case")
 784                         return
 785                 default:
 786                         xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 787                         removeSubscriptionFromDb = true
 788                         subRfMsg, valid = subs.SetCachedResponse(nil, false)
 789                         c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 790                 }
 791                 xapp.Logger.Debug("SUBS-SubReq: Handling (e2t response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
 792         } else {
 793                 xapp.Logger.Debug("SUBS-SubReq: Handling (cached response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
 794         }
 795
 796         //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
 797         if valid == false {
 798                 c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanup_ms, c)
 799         }
 800
 801         c.UpdateSubscriptionInDB(subs, removeSubscriptionFromDb)
 802         parentTrans.SendEvent(subRfMsg, 0)
 803 }
 804
 805 //-------------------------------------------------------------------
 806 // SUBS DELETE Handling
 807 //-------------------------------------------------------------------
 808
 809 func (c *Control) handleSubscriptionDelete(subs *Subscription, parentTrans *TransactionXapp) {
 810
 811         trans := c.tracker.NewSubsTransaction(subs)
 812         subs.WaitTransactionTurn(trans)
 813         defer subs.ReleaseTransactionTurn(trans)
 814         defer trans.Release()
 815
 816         xapp.Logger.Debug("SUBS-SubDelReq: Handling %s", idstring(nil, trans, subs, parentTrans))
 817
 818         subs.mutex.Lock()
 819
 820         if subs.valid && subs.EpList.HasEndpoint(parentTrans.GetEndpoint()) && subs.EpList.Size() == 1 {
 821                 subs.valid = false
 822                 subs.mutex.Unlock()
 823                 c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 824         } else {
 825                 subs.mutex.Unlock()
 826         }
 827         //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
 828         //  If parallel deletes ongoing both might pass earlier sendE2TSubscriptionDeleteRequest(...) if
 829         //  RemoveFromSubscription locates in caller side (now in handleXAPPSubscriptionDeleteRequest(...))
 830         c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanup_ms, c)
 831         c.registry.UpdateSubscriptionToDb(subs, c)
 832         parentTrans.SendEvent(nil, 0)
 833 }
 834
 835 //-------------------------------------------------------------------
 836 // send to E2T Subscription Request
 837 //-------------------------------------------------------------------
 838 func (c *Control) sendE2TSubscriptionRequest(subs *Subscription, trans *TransactionSubs, parentTrans *TransactionXapp) interface{} {
 839         var err error
 840         var event interface{} = nil
 841         var timedOut bool = false
 842         const ricRequestorId = 123
 843
 844         subReqMsg := subs.SubReqMsg
 845         subReqMsg.RequestId = subs.GetReqId().RequestId
 846         subReqMsg.RequestId.Id = ricRequestorId
 847         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionRequest(subReqMsg)
 848         if err != nil {
 849                 xapp.Logger.Error("SUBS-SubReq: %s", idstring(err, trans, subs, parentTrans))
 850                 return event
 851         }
 852
 853         // Write uncompleted subscrition in db. If no response for subscrition it need to be re-processed (deleted) after restart
 854         c.WriteSubscriptionToDb(subs)
 855
 856         for retries := uint64(0); retries < e2tMaxSubReqTryCount; retries++ {
 857                 desc := fmt.Sprintf("(retry %d)", retries)
 858                 if retries == 0 {
 859                         c.UpdateCounter(cSubReqToE2)
 860                 } else {
 861                         c.UpdateCounter(cSubReReqToE2)
 862                 }
 863                 c.rmrSendToE2T(desc, subs, trans)
 864                 if subs.DoNotWaitSubResp == false {
 865                         event, timedOut = trans.WaitEvent(e2tSubReqTimeout)
 866                         if timedOut {
 867                                 c.UpdateCounter(cSubReqTimerExpiry)
 868                                 continue
 869                         }
 870                 } else {
 871                         // Simulating case where subscrition request has been sent but response has not been received before restart
 872                         event = &SubmgrRestartTestEvent{}
 873                 }
 874                 break
 875         }
 876         xapp.Logger.Debug("SUBS-SubReq: Response handling event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 877         return event
 878 }
 879
 880 //-------------------------------------------------------------------
 881 // send to E2T Subscription Delete Request
 882 //-------------------------------------------------------------------
 883
 884 func (c *Control) sendE2TSubscriptionDeleteRequest(subs *Subscription, trans *TransactionSubs, parentTrans *TransactionXapp) interface{} {
 885         var err error
 886         var event interface{}
 887         var timedOut bool
 888         const ricRequestorId = 123
 889
 890         subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{}
 891         subDelReqMsg.RequestId = subs.GetReqId().RequestId
 892         subDelReqMsg.RequestId.Id = ricRequestorId
 893         subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId
 894         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg)
 895         if err != nil {
 896                 xapp.Logger.Error("SUBS-SubDelReq: %s", idstring(err, trans, subs, parentTrans))
 897                 return event
 898         }
 899
 900         for retries := uint64(0); retries < e2tMaxSubDelReqTryCount; retries++ {
 901                 desc := fmt.Sprintf("(retry %d)", retries)
 902                 if retries == 0 {
 903                         c.UpdateCounter(cSubDelReqToE2)
 904                 } else {
 905                         c.UpdateCounter(cSubDelReReqToE2)
 906                 }
 907                 c.rmrSendToE2T(desc, subs, trans)
 908                 event, timedOut = trans.WaitEvent(e2tSubDelReqTime)
 909                 if timedOut {
 910                         c.UpdateCounter(cSubDelReqTimerExpiry)
 911                         continue
 912                 }
 913                 break
 914         }
 915         xapp.Logger.Debug("SUBS-SubDelReq: Response handling event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 916         return event
 917 }
 918
 919 //-------------------------------------------------------------------
 920 // handle from E2T Subscription Response
 921 //-------------------------------------------------------------------
 922 func (c *Control) handleE2TSubscriptionResponse(params *xapp.RMRParams) {
 923         xapp.Logger.Info("MSG from E2T: %s", params.String())
 924         c.UpdateCounter(cSubRespFromE2)
 925
 926         subRespMsg, err := c.e2ap.UnpackSubscriptionResponse(params.Payload)
 927         if err != nil {
 928                 xapp.Logger.Error("MSG-SubResp %s", idstring(err, params))
 929                 return
 930         }
 931         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subRespMsg.RequestId.InstanceId})
 932         if err != nil {
 933                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, params))
 934                 return
 935         }
 936         trans := subs.GetTransaction()
 937         if trans == nil {
 938                 err = fmt.Errorf("Ongoing transaction not found")
 939                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, params, subs))
 940                 return
 941         }
 942         sendOk, timedOut := trans.SendEvent(subRespMsg, e2tRecvMsgTimeout)
 943         if sendOk == false {
 944                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 945                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, trans, subs))
 946         }
 947         return
 948 }
 949
 950 //-------------------------------------------------------------------
 951 // handle from E2T Subscription Failure
 952 //-------------------------------------------------------------------
 953 func (c *Control) handleE2TSubscriptionFailure(params *xapp.RMRParams) {
 954         xapp.Logger.Info("MSG from E2T: %s", params.String())
 955         c.UpdateCounter(cSubFailFromE2)
 956         subFailMsg, err := c.e2ap.UnpackSubscriptionFailure(params.Payload)
 957         if err != nil {
 958                 xapp.Logger.Error("MSG-SubFail %s", idstring(err, params))
 959                 return
 960         }
 961         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subFailMsg.RequestId.InstanceId})
 962         if err != nil {
 963                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, params))
 964                 return
 965         }
 966         trans := subs.GetTransaction()
 967         if trans == nil {
 968                 err = fmt.Errorf("Ongoing transaction not found")
 969                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, params, subs))
 970                 return
 971         }
 972         sendOk, timedOut := trans.SendEvent(subFailMsg, e2tRecvMsgTimeout)
 973         if sendOk == false {
 974                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 975                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, trans, subs))
 976         }
 977         return
 978 }
 979
 980 //-------------------------------------------------------------------
 981 // handle from E2T Subscription Delete Response
 982 //-------------------------------------------------------------------
 983 func (c *Control) handleE2TSubscriptionDeleteResponse(params *xapp.RMRParams) (err error) {
 984         xapp.Logger.Info("MSG from E2T: %s", params.String())
 985         c.UpdateCounter(cSubDelRespFromE2)
 986         subDelRespMsg, err := c.e2ap.UnpackSubscriptionDeleteResponse(params.Payload)
 987         if err != nil {
 988                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params))
 989                 return
 990         }
 991         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelRespMsg.RequestId.InstanceId})
 992         if err != nil {
 993                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params))
 994                 return
 995         }
 996         trans := subs.GetTransaction()
 997         if trans == nil {
 998                 err = fmt.Errorf("Ongoing transaction not found")
 999                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params, subs))
1000                 return
1001         }
1002         sendOk, timedOut := trans.SendEvent(subDelRespMsg, e2tRecvMsgTimeout)
1003         if sendOk == false {
1004                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
1005                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, trans, subs))
1006         }
1007         return
1008 }
1009
1010 //-------------------------------------------------------------------
1011 // handle from E2T Subscription Delete Failure
1012 //-------------------------------------------------------------------
1013 func (c *Control) handleE2TSubscriptionDeleteFailure(params *xapp.RMRParams) {
1014         xapp.Logger.Info("MSG from E2T: %s", params.String())
1015         c.UpdateCounter(cSubDelFailFromE2)
1016         subDelFailMsg, err := c.e2ap.UnpackSubscriptionDeleteFailure(params.Payload)
1017         if err != nil {
1018                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
1019                 return
1020         }
1021         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelFailMsg.RequestId.InstanceId})
1022         if err != nil {
1023                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
1024                 return
1025         }
1026         trans := subs.GetTransaction()
1027         if trans == nil {
1028                 err = fmt.Errorf("Ongoing transaction not found")
1029                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params, subs))
1030                 return
1031         }
1032         sendOk, timedOut := trans.SendEvent(subDelFailMsg, e2tRecvMsgTimeout)
1033         if sendOk == false {
1034                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
1035                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, trans, subs))
1036         }
1037         return
1038 }
1039
1040 //-------------------------------------------------------------------
1041 //
1042 //-------------------------------------------------------------------
1043 func typeofSubsMessage(v interface{}) string {
1044         if v == nil {
1045                 return "NIL"
1046         }
1047         switch v.(type) {
1048         //case *e2ap.E2APSubscriptionRequest:
1049         //      return "SubReq"
1050         case *e2ap.E2APSubscriptionResponse:
1051                 return "SubResp"
1052         case *e2ap.E2APSubscriptionFailure:
1053                 return "SubFail"
1054         //case *e2ap.E2APSubscriptionDeleteRequest:
1055         //      return "SubDelReq"
1056         case *e2ap.E2APSubscriptionDeleteResponse:
1057                 return "SubDelResp"
1058         case *e2ap.E2APSubscriptionDeleteFailure:
1059                 return "SubDelFail"
1060         default:
1061                 return "Unknown"
1062         }
1063 }
1064
1065 //-------------------------------------------------------------------
1066 //
1067 //-------------------------------------------------------------------
1068 func (c *Control) WriteSubscriptionToDb(subs *Subscription) {
1069         xapp.Logger.Debug("WriteSubscriptionToDb() subId = %v", subs.ReqId.InstanceId)
1070         err := c.WriteSubscriptionToSdl(subs.ReqId.InstanceId, subs)
1071         if err != nil {
1072                 xapp.Logger.Error("%v", err)
1073         }
1074 }
1075
1076 //-------------------------------------------------------------------
1077 //
1078 //-------------------------------------------------------------------
1079 func (c *Control) UpdateSubscriptionInDB(subs *Subscription, removeSubscriptionFromDb bool) {
1080
1081         if removeSubscriptionFromDb == true {
1082                 // Subscription was written in db already when subscription request was sent to BTS, except for merged request
1083                 c.RemoveSubscriptionFromDb(subs)
1084         } else {
1085                 // Update is needed for successful response and merge case here
1086                 if subs.RetryFromXapp == false {
1087                         c.WriteSubscriptionToDb(subs)
1088                 }
1089         }
1090         subs.RetryFromXapp = false
1091 }
1092
1093 //-------------------------------------------------------------------
1094 //
1095 //-------------------------------------------------------------------
1096 func (c *Control) RemoveSubscriptionFromDb(subs *Subscription) {
1097         xapp.Logger.Debug("RemoveSubscriptionFromDb() subId = %v", subs.ReqId.InstanceId)
1098         err := c.RemoveSubscriptionFromSdl(subs.ReqId.InstanceId)
1099         if err != nil {
1100                 xapp.Logger.Error("%v", err)
1101         }
1102 }
1103
1104 func (c *Control) SendSubscriptionDeleteReq(subs *Subscription) {
1105
1106         const ricRequestorId = 123
1107         xapp.Logger.Debug("Sending subscription delete due to restart. subId = %v", subs.ReqId.InstanceId)
1108
1109         // Send delete for every endpoint in the subscription
1110         subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{}
1111         subDelReqMsg.RequestId = subs.GetReqId().RequestId
1112         subDelReqMsg.RequestId.Id = ricRequestorId
1113         subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId
1114         mType, payload, err := c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg)
1115         if err != nil {
1116                 xapp.Logger.Error("SendSubscriptionDeleteReq() %s", idstring(err))
1117                 return
1118         }
1119         for _, endPoint := range subs.EpList.Endpoints {
1120                 params := &xapp.RMRParams{}
1121                 params.Mtype = mType
1122                 params.SubId = int(subs.GetReqId().InstanceId)
1123                 params.Xid = ""
1124                 params.Meid = subs.Meid
1125                 params.Src = endPoint.String()
1126                 params.PayloadLen = len(payload.Buf)
1127                 params.Payload = payload.Buf
1128                 params.Mbuf = nil
1129                 subs.DeleteFromDb = true
1130                 c.handleXAPPSubscriptionDeleteRequest(params)
1131         }
1132 }
1133
1134 func (c *Control) PrintRESTSubscriptionRequest(p *models.SubscriptionParams) {
1135
1136         fmt.Println("CRESTSubscriptionRequest")
1137         fmt.Printf("  ClientEndpoint.Host = %s\n", p.ClientEndpoint.Host)
1138
1139         if p.ClientEndpoint.HTTPPort != nil {
1140                 fmt.Printf("  ClientEndpoint.HTTPPort = %v\n", *p.ClientEndpoint.HTTPPort)
1141         } else {
1142                 fmt.Println("  ClientEndpoint.HTTPPort = nil")
1143         }
1144
1145         if p.ClientEndpoint.RMRPort != nil {
1146                 fmt.Printf("  ClientEndpoint.RMRPort = %v\n", *p.ClientEndpoint.RMRPort)
1147         } else {
1148                 fmt.Println("  ClientEndpoint.RMRPort = nil")
1149         }
1150
1151         if p.Meid != nil {
1152                 fmt.Printf("  Meid = %s\n", *p.Meid)
1153         } else {
1154                 fmt.Println("  Meid = nil")
1155         }
1156
1157         for _, subscriptionDetail := range p.SubscriptionDetails {
1158                 if p.RANFunctionID != nil {
1159                         fmt.Printf("  RANFunctionID = %v\n", *p.RANFunctionID)
1160                 } else {
1161                         fmt.Println("  RANFunctionID = nil")
1162                 }
1163                 fmt.Printf("  SubscriptionDetail.XappEventInstanceID = %v\n", *subscriptionDetail.XappEventInstanceID)
1164                 fmt.Printf("  SubscriptionDetail.EventTriggers.OctetString = %X\n", subscriptionDetail.EventTriggers.OctetString)
1165
1166                 for _, actionToBeSetup := range subscriptionDetail.ActionToBeSetupList {
1167                         fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionID = %v\n", *actionToBeSetup.ActionID)
1168                         fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionType = %s\n", *actionToBeSetup.ActionType)
1169                         if actionToBeSetup.ActionDefinition != nil {
1170                                 fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionDefinition.OctetString = %X\n", actionToBeSetup.ActionDefinition.OctetString)
1171                         } else {
1172                                 fmt.Println("  SubscriptionDetail.ActionToBeSetup.ActionDefinition = nil")
1173
1174                         }
1175                         if actionToBeSetup.SubsequentAction != nil {
1176                                 fmt.Printf("  SubscriptionDetail.ActionToBeSetup.SubsequentAction.SubsequentActionType = %s\n", *actionToBeSetup.SubsequentAction.SubsequentActionType)
1177                                 fmt.Printf("  SubscriptionDetail.ActionToBeSetup..SubsequentAction.TimeToWait = %s\n", *actionToBeSetup.SubsequentAction.TimeToWait)
1178                         } else {
1179                                 fmt.Println("  SubscriptionDetail.ActionToBeSetup.SubsequentAction = nil")
1180                         }
1181                 }
1182         }
1183 }