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         restSubId := ksuid.New().String()
 237         subResp := models.SubscriptionResponse{}
 238         subResp.SubscriptionID = &restSubId
 239         p := params.(*models.SubscriptionParams)
 240
 241         if c.LoggerLevel > 2 {
 242                 c.PrintRESTSubscriptionRequest(p)
 243         }
 244
 245         if p.ClientEndpoint == nil {
 246                 xapp.Logger.Error("ClientEndpoint == nil")
 247                 c.UpdateCounter(cRestSubFailToXapp)
 248                 return nil, fmt.Errorf("")
 249         }
 250
 251         _, xAppRmrEndpoint, err := ConstructEndpointAddresses(*p.ClientEndpoint)
 252         if err != nil {
 253                 xapp.Logger.Error("%s", err.Error())
 254                 c.UpdateCounter(cRestSubFailToXapp)
 255                 return nil, err
 256         }
 257
 258         restSubscription, err := c.registry.CreateRESTSubscription(&restSubId, &xAppRmrEndpoint, p.Meid)
 259         if err != nil {
 260                 xapp.Logger.Error("%s", err.Error())
 261                 c.UpdateCounter(cRestSubFailToXapp)
 262                 return nil, err
 263         }
 264
 265         subReqList := e2ap.SubscriptionRequestList{}
 266         err = c.e2ap.FillSubscriptionReqMsgs(params, &subReqList, restSubscription)
 267         if err != nil {
 268                 xapp.Logger.Error("%s", err.Error())
 269                 c.registry.DeleteRESTSubscription(&restSubId)
 270                 c.UpdateCounter(cRestSubFailToXapp)
 271                 return nil, err
 272         }
 273
 274         go c.processSubscriptionRequests(restSubscription, &subReqList, p.ClientEndpoint, p.Meid, &restSubId)
 275
 276         c.UpdateCounter(cRestSubRespToXapp)
 277         return &subResp, nil
 278 }
 279
 280 //-------------------------------------------------------------------
 281 //
 282 //-------------------------------------------------------------------
 283
 284 func (c *Control) processSubscriptionRequests(restSubscription *RESTSubscription, subReqList *e2ap.SubscriptionRequestList,
 285         clientEndpoint *models.SubscriptionParamsClientEndpoint, meid *string, restSubId *string) {
 286
 287         xapp.Logger.Info("Subscription Request count=%v ", len(subReqList.E2APSubscriptionRequests))
 288
 289         _, xAppRmrEndpoint, err := ConstructEndpointAddresses(*clientEndpoint)
 290         if err != nil {
 291                 xapp.Logger.Error("%s", err.Error())
 292                 return
 293         }
 294
 295         var requestorID int64
 296         var instanceId int64
 297         for index := 0; index < len(subReqList.E2APSubscriptionRequests); index++ {
 298                 subReqMsg := subReqList.E2APSubscriptionRequests[index]
 299
 300                 xid := *restSubId + "_" + strconv.FormatUint(uint64(subReqMsg.RequestId.InstanceId), 10)
 301                 trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(xAppRmrEndpoint), xid, subReqMsg.RequestId, &xapp.RMRMeid{RanName: *meid})
 302                 //trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(xAppRmrEndpoint), *restSubId, subReqMsg.RequestId, &xapp.RMRMeid{RanName: *meid})
 303                 if trans == nil {
 304                         c.registry.DeleteRESTSubscription(restSubId)
 305                         xapp.Logger.Error("XAPP-SubReq transaction not created. RESTSubId=%s, EndPoint=%s, Meid=%s", *restSubId, xAppRmrEndpoint, *meid)
 306                         return
 307                 }
 308
 309                 defer trans.Release()
 310                 xapp.Logger.Info("Handle SubscriptionRequest index=%v, %s", index, idstring(nil, trans))
 311                 subRespMsg, err := c.handleSubscriptionRequest(trans, &subReqMsg, meid, restSubId)
 312                 if err != nil {
 313                         // Send notification to xApp that prosessing of a Subscription Request has failed. Currently it is not possible
 314                         // to indicate error. Such possibility should be added. As a workaround requestorID and instanceId are set to zero value
 315                         requestorID = (int64)(0)
 316                         instanceId = (int64)(0)
 317                         resp := &models.SubscriptionResponse{
 318                                 SubscriptionID: restSubId,
 319                                 SubscriptionInstances: []*models.SubscriptionInstance{
 320                                         &models.SubscriptionInstance{RequestorID: &requestorID, InstanceID: &instanceId},
 321                                 },
 322                         }
 323                         // Mark REST subscription request processed.
 324                         restSubscription.SetProcessed()
 325                         xapp.Logger.Info("Sending unsuccessful REST notification to endpoint=%v:%v, InstanceId=%v, %s", clientEndpoint.Host, clientEndpoint.HTTPPort, instanceId, idstring(nil, trans))
 326                         xapp.Subscription.Notify(resp, *clientEndpoint)
 327                         c.UpdateCounter(cRestSubFailNotifToXapp)
 328                 } else {
 329                         xapp.Logger.Info("SubscriptionRequest index=%v processed successfully. endpoint=%v, InstanceId=%v, %s", index, *clientEndpoint, instanceId, idstring(nil, trans))
 330
 331                         // Store successfully processed InstanceId for deletion
 332                         restSubscription.AddInstanceId(subRespMsg.RequestId.InstanceId)
 333
 334                         // Send notification to xApp that a Subscription Request has been processed.
 335                         requestorID = (int64)(subRespMsg.RequestId.Id)
 336                         instanceId = (int64)(subRespMsg.RequestId.InstanceId)
 337                         resp := &models.SubscriptionResponse{
 338                                 SubscriptionID: restSubId,
 339                                 SubscriptionInstances: []*models.SubscriptionInstance{
 340                                         &models.SubscriptionInstance{RequestorID: &requestorID, InstanceID: &instanceId},
 341                                 },
 342                         }
 343                         // Mark REST subscription request processesd.
 344                         restSubscription.SetProcessed()
 345                         xapp.Logger.Info("Sending successful REST notification to endpoint=%v, InstanceId=%v, %s", *clientEndpoint, instanceId, idstring(nil, trans))
 346                         xapp.Subscription.Notify(resp, *clientEndpoint)
 347                         c.UpdateCounter(cRestSubNotifToXapp)
 348
 349                 }
 350         }
 351 }
 352
 353 //-------------------------------------------------------------------
 354 //
 355 //------------------------------------------------------------------
 356 func (c *Control) handleSubscriptionRequest(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest, meid *string,
 357         restSubId *string) (*e2ap.E2APSubscriptionResponse, error) {
 358
 359         err := c.tracker.Track(trans)
 360         if err != nil {
 361                 err = fmt.Errorf("XAPP-SubReq: %s", idstring(err, trans))
 362                 xapp.Logger.Error("%s", err.Error())
 363                 return nil, err
 364         }
 365
 366         subs, err := c.registry.AssignToSubscription(trans, subReqMsg, c.ResetTestFlag, c)
 367         if err != nil {
 368                 err = fmt.Errorf("XAPP-SubReq: %s", idstring(err, trans))
 369                 xapp.Logger.Error("%s", err.Error())
 370                 return nil, err
 371         }
 372
 373         //
 374         // Wake subs request
 375         //
 376         go c.handleSubscriptionCreate(subs, trans)
 377         event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 378
 379         err = nil
 380         if event != nil {
 381                 switch themsg := event.(type) {
 382                 case *e2ap.E2APSubscriptionResponse:
 383                         trans.Release()
 384                         return themsg, nil
 385                 case *e2ap.E2APSubscriptionFailure:
 386                         err = fmt.Errorf("SubscriptionFailure received")
 387                         return nil, err
 388                 default:
 389                         break
 390                 }
 391         }
 392         err = fmt.Errorf("XAPP-SubReq: failed %s", idstring(err, trans, subs))
 393         xapp.Logger.Error("%s", err.Error())
 394         c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
 395         return nil, err
 396 }
 397
 398 //-------------------------------------------------------------------
 399 //
 400 //-------------------------------------------------------------------
 401 func (c *Control) SubscriptionDeleteHandlerCB(restSubId string) error {
 402
 403         c.CntRecvMsg++
 404         c.UpdateCounter(cRestSubDelReqFromXapp)
 405
 406         xapp.Logger.Info("SubscriptionDeleteRequest from XAPP")
 407
 408         restSubscription, err := c.registry.GetRESTSubscription(restSubId)
 409         if err != nil {
 410                 xapp.Logger.Error("%s", err.Error())
 411                 if restSubscription == nil {
 412                         // Subscription was not found
 413                         return nil
 414                 } else {
 415                         if restSubscription.SubReqOngoing == true {
 416                                 err := fmt.Errorf("Handling of the REST Subscription Request still ongoing %s", restSubId)
 417                                 xapp.Logger.Error("%s", err.Error())
 418                                 return err
 419                         } else if restSubscription.SubDelReqOngoing == true {
 420                                 // Previous request for same restSubId still ongoing
 421                                 return nil
 422                         }
 423                 }
 424         }
 425
 426         xAppRmrEndPoint := restSubscription.xAppRmrEndPoint
 427         go func() {
 428                 for _, instanceId := range restSubscription.InstanceIds {
 429                         err := c.SubscriptionDeleteHandler(&restSubId, &xAppRmrEndPoint, &restSubscription.Meid, instanceId)
 430                         if err != nil {
 431                                 xapp.Logger.Error("%s", err.Error())
 432                                 //return err
 433                         }
 434                         xapp.Logger.Info("Deleteting instanceId = %v", instanceId)
 435                         restSubscription.DeleteInstanceId(instanceId)
 436                 }
 437                 c.registry.DeleteRESTSubscription(&restSubId)
 438         }()
 439
 440         c.UpdateCounter(cRestSubDelRespToXapp)
 441
 442         return nil
 443 }
 444
 445 //-------------------------------------------------------------------
 446 //
 447 //-------------------------------------------------------------------
 448 func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string, meid *string, instanceId uint32) error {
 449
 450         xid := *restSubId + "_" + strconv.FormatUint(uint64(instanceId), 10)
 451         trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(*endPoint), xid, e2ap.RequestId{0, 0}, &xapp.RMRMeid{RanName: *meid})
 452         //trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(*endPoint), *restSubId, e2ap.RequestId{0, 0}, &xapp.RMRMeid{RanName: *meid})
 453         if trans == nil {
 454                 err := fmt.Errorf("XAPP-SubDelReq transaction not created. restSubId %s, endPoint %s, meid %s, instanceId %v", *restSubId, *endPoint, *meid, instanceId)
 455                 xapp.Logger.Error("%s", err.Error())
 456         }
 457         defer trans.Release()
 458
 459         err := c.tracker.Track(trans)
 460         if err != nil {
 461                 err := fmt.Errorf("XAPP-SubDelReq %s:", idstring(err, trans))
 462                 xapp.Logger.Error("%s", err.Error())
 463                 return &time.ParseError{}
 464         }
 465
 466         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{instanceId})
 467         if err != nil {
 468                 err := fmt.Errorf("XAPP-SubDelReq %s:", idstring(err, trans))
 469                 xapp.Logger.Error("%s", err.Error())
 470                 return err
 471         }
 472         //
 473         // Wake subs delete
 474         //
 475         go c.handleSubscriptionDelete(subs, trans)
 476         trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 477
 478         xapp.Logger.Debug("XAPP-SubDelReq: Handling event %s ", idstring(nil, trans, subs))
 479
 480         c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c)
 481
 482         return nil
 483 }
 484
 485 //-------------------------------------------------------------------
 486 //
 487 //-------------------------------------------------------------------
 488 func (c *Control) QueryHandler() (models.SubscriptionList, error) {
 489         xapp.Logger.Info("QueryHandler() called")
 490
 491         c.CntRecvMsg++
 492
 493         return c.registry.QueryHandler()
 494 }
 495
 496 func (c *Control) TestRestHandler(w http.ResponseWriter, r *http.Request) {
 497         xapp.Logger.Info("TestRestHandler() called")
 498
 499         pathParams := mux.Vars(r)
 500         s := pathParams["testId"]
 501
 502         // This can be used to delete single subscription from db
 503         if contains := strings.Contains(s, "deletesubid="); contains == true {
 504                 var splits = strings.Split(s, "=")
 505                 if subId, err := strconv.ParseInt(splits[1], 10, 64); err == nil {
 506                         xapp.Logger.Info("RemoveSubscriptionFromSdl() called. subId = %v", subId)
 507                         c.RemoveSubscriptionFromSdl(uint32(subId))
 508                         return
 509                 }
 510         }
 511
 512         // This can be used to remove all subscriptions db from
 513         if s == "emptydb" {
 514                 xapp.Logger.Info("RemoveAllSubscriptionsFromSdl() called")
 515                 c.RemoveAllSubscriptionsFromSdl()
 516                 return
 517         }
 518
 519         // This is meant to cause submgr's restart in testing
 520         if s == "restart" {
 521                 xapp.Logger.Info("os.Exit(1) called")
 522                 os.Exit(1)
 523         }
 524
 525         xapp.Logger.Info("Unsupported rest command received %s", s)
 526 }
 527
 528 //-------------------------------------------------------------------
 529 //
 530 //-------------------------------------------------------------------
 531
 532 func (c *Control) rmrSendToE2T(desc string, subs *Subscription, trans *TransactionSubs) (err error) {
 533         params := &xapp.RMRParams{}
 534         params.Mtype = trans.GetMtype()
 535         params.SubId = int(subs.GetReqId().InstanceId)
 536         params.Xid = ""
 537         params.Meid = subs.GetMeid()
 538         params.Src = ""
 539         params.PayloadLen = len(trans.Payload.Buf)
 540         params.Payload = trans.Payload.Buf
 541         params.Mbuf = nil
 542         xapp.Logger.Info("MSG to E2T: %s %s %s", desc, trans.String(), params.String())
 543         err = c.SendWithRetry(params, false, 5)
 544         if err != nil {
 545                 xapp.Logger.Error("rmrSendToE2T: Send failed: %+v", err)
 546         }
 547         return err
 548 }
 549
 550 func (c *Control) rmrSendToXapp(desc string, subs *Subscription, trans *TransactionXapp) (err error) {
 551
 552         params := &xapp.RMRParams{}
 553         params.Mtype = trans.GetMtype()
 554         params.SubId = int(subs.GetReqId().InstanceId)
 555         params.Xid = trans.GetXid()
 556         params.Meid = trans.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 XAPP: %s %s %s", desc, trans.String(), params.String())
 562         err = c.SendWithRetry(params, false, 5)
 563         if err != nil {
 564                 xapp.Logger.Error("rmrSendToXapp: Send failed: %+v", err)
 565         }
 566         return err
 567 }
 568
 569 func (c *Control) Consume(msg *xapp.RMRParams) (err error) {
 570         if c.RMRClient == nil {
 571                 err = fmt.Errorf("Rmr object nil can handle %s", msg.String())
 572                 xapp.Logger.Error("%s", err.Error())
 573                 return
 574         }
 575         c.CntRecvMsg++
 576
 577         defer c.RMRClient.Free(msg.Mbuf)
 578
 579         // xapp-frame might use direct access to c buffer and
 580         // when msg.Mbuf is freed, someone might take it into use
 581         // and payload data might be invalid inside message handle function
 582         //
 583         // subscriptions won't load system a lot so there is no
 584         // real performance hit by cloning buffer into new go byte slice
 585         cPay := append(msg.Payload[:0:0], msg.Payload...)
 586         msg.Payload = cPay
 587         msg.PayloadLen = len(cPay)
 588
 589         switch msg.Mtype {
 590         case xapp.RIC_SUB_REQ:
 591                 go c.handleXAPPSubscriptionRequest(msg)
 592         case xapp.RIC_SUB_RESP:
 593                 go c.handleE2TSubscriptionResponse(msg)
 594         case xapp.RIC_SUB_FAILURE:
 595                 go c.handleE2TSubscriptionFailure(msg)
 596         case xapp.RIC_SUB_DEL_REQ:
 597                 go c.handleXAPPSubscriptionDeleteRequest(msg)
 598         case xapp.RIC_SUB_DEL_RESP:
 599                 go c.handleE2TSubscriptionDeleteResponse(msg)
 600         case xapp.RIC_SUB_DEL_FAILURE:
 601                 go c.handleE2TSubscriptionDeleteFailure(msg)
 602         default:
 603                 xapp.Logger.Info("Unknown Message Type '%d', discarding", msg.Mtype)
 604         }
 605         return
 606 }
 607
 608 //-------------------------------------------------------------------
 609 // handle from XAPP Subscription Request
 610 //------------------------------------------------------------------
 611 func (c *Control) handleXAPPSubscriptionRequest(params *xapp.RMRParams) {
 612         xapp.Logger.Info("MSG from XAPP: %s", params.String())
 613         c.UpdateCounter(cSubReqFromXapp)
 614
 615         subReqMsg, err := c.e2ap.UnpackSubscriptionRequest(params.Payload)
 616         if err != nil {
 617                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, params))
 618                 return
 619         }
 620
 621         trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subReqMsg.RequestId, params.Meid)
 622         if trans == nil {
 623                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 624                 return
 625         }
 626         defer trans.Release()
 627
 628         if err = c.tracker.Track(trans); err != nil {
 629                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 630                 return
 631         }
 632
 633         //TODO handle subscription toward e2term inside AssignToSubscription / hide handleSubscriptionCreate in it?
 634         subs, err := c.registry.AssignToSubscription(trans, subReqMsg, c.ResetTestFlag, c)
 635         if err != nil {
 636                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 637                 return
 638         }
 639
 640         c.wakeSubscriptionRequest(subs, trans)
 641 }
 642
 643 //-------------------------------------------------------------------
 644 // Wake Subscription Request to E2node
 645 //------------------------------------------------------------------
 646 func (c *Control) wakeSubscriptionRequest(subs *Subscription, trans *TransactionXapp) {
 647
 648         go c.handleSubscriptionCreate(subs, trans)
 649         event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 650         var err error
 651         if event != nil {
 652                 switch themsg := event.(type) {
 653                 case *e2ap.E2APSubscriptionResponse:
 654                         themsg.RequestId.Id = trans.RequestId.Id
 655                         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionResponse(themsg)
 656                         if err == nil {
 657                                 trans.Release()
 658                                 c.UpdateCounter(cSubRespToXapp)
 659                                 c.rmrSendToXapp("", subs, trans)
 660                                 return
 661                         }
 662                 case *e2ap.E2APSubscriptionFailure:
 663                         themsg.RequestId.Id = trans.RequestId.Id
 664                         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionFailure(themsg)
 665                         if err == nil {
 666                                 c.UpdateCounter(cSubFailToXapp)
 667                                 c.rmrSendToXapp("", subs, trans)
 668                         }
 669                 default:
 670                         break
 671                 }
 672         }
 673         xapp.Logger.Info("XAPP-SubReq: failed %s", idstring(err, trans, subs))
 674         //c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
 675 }
 676
 677 //-------------------------------------------------------------------
 678 // handle from XAPP Subscription Delete Request
 679 //------------------------------------------------------------------
 680 func (c *Control) handleXAPPSubscriptionDeleteRequest(params *xapp.RMRParams) {
 681         xapp.Logger.Info("MSG from XAPP: %s", params.String())
 682         c.UpdateCounter(cSubDelReqFromXapp)
 683
 684         subDelReqMsg, err := c.e2ap.UnpackSubscriptionDeleteRequest(params.Payload)
 685         if err != nil {
 686                 xapp.Logger.Error("XAPP-SubDelReq %s", idstring(err, params))
 687                 return
 688         }
 689
 690         trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subDelReqMsg.RequestId, params.Meid)
 691         if trans == nil {
 692                 xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(fmt.Errorf("transaction not created"), params))
 693                 return
 694         }
 695         defer trans.Release()
 696
 697         err = c.tracker.Track(trans)
 698         if err != nil {
 699                 xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
 700                 return
 701         }
 702
 703         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{trans.GetSubId()})
 704         if err != nil {
 705                 xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(err, trans))
 706                 return
 707         }
 708
 709         //
 710         // Wake subs delete
 711         //
 712         go c.handleSubscriptionDelete(subs, trans)
 713         trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
 714
 715         xapp.Logger.Debug("XAPP-SubDelReq: Handling event %s ", idstring(nil, trans, subs))
 716
 717         if subs.NoRespToXapp == true {
 718                 // Do no send delete responses to xapps due to submgr restart is deleting uncompleted subscriptions
 719                 return
 720         }
 721
 722         // Whatever is received success, fail or timeout, send successful delete response
 723         subDelRespMsg := &e2ap.E2APSubscriptionDeleteResponse{}
 724         subDelRespMsg.RequestId.Id = trans.RequestId.Id
 725         subDelRespMsg.RequestId.InstanceId = subs.GetReqId().RequestId.InstanceId
 726         subDelRespMsg.FunctionId = subs.SubReqMsg.FunctionId
 727         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteResponse(subDelRespMsg)
 728         if err == nil {
 729                 c.UpdateCounter(cSubDelRespToXapp)
 730                 c.rmrSendToXapp("", subs, trans)
 731         }
 732
 733         //TODO handle subscription toward e2term insiged RemoveFromSubscription / hide handleSubscriptionDelete in it?
 734         //c.registry.RemoveFromSubscription(subs, trans, 5*time.Second)
 735 }
 736
 737 //-------------------------------------------------------------------
 738 // SUBS CREATE Handling
 739 //-------------------------------------------------------------------
 740 func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *TransactionXapp) {
 741
 742         var removeSubscriptionFromDb bool = false
 743         trans := c.tracker.NewSubsTransaction(subs)
 744         subs.WaitTransactionTurn(trans)
 745         defer subs.ReleaseTransactionTurn(trans)
 746         defer trans.Release()
 747
 748         xapp.Logger.Debug("SUBS-SubReq: Handling %s ", idstring(nil, trans, subs, parentTrans))
 749
 750         subRfMsg, valid := subs.GetCachedResponse()
 751         if subRfMsg == nil && valid == true {
 752                 event := c.sendE2TSubscriptionRequest(subs, trans, parentTrans)
 753                 switch event.(type) {
 754                 case *e2ap.E2APSubscriptionResponse:
 755                         subRfMsg, valid = subs.SetCachedResponse(event, true)
 756                         subs.SubRespRcvd = true
 757                 case *e2ap.E2APSubscriptionFailure:
 758                         removeSubscriptionFromDb = true
 759                         subRfMsg, valid = subs.SetCachedResponse(event, false)
 760                         xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 761                         c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 762                 case *SubmgrRestartTestEvent:
 763                         // This simulates that no response has been received and after restart subscriptions are restored from db
 764                         xapp.Logger.Debug("Test restart flag is active. Dropping this transaction to test restart case")
 765                         return
 766                 default:
 767                         xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 768                         removeSubscriptionFromDb = true
 769                         subRfMsg, valid = subs.SetCachedResponse(nil, false)
 770                         c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 771                 }
 772                 xapp.Logger.Debug("SUBS-SubReq: Handling (e2t response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
 773         } else {
 774                 xapp.Logger.Debug("SUBS-SubReq: Handling (cached response %s) %s", typeofSubsMessage(subRfMsg), idstring(nil, trans, subs, parentTrans))
 775         }
 776
 777         //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
 778         if valid == false {
 779                 c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanup_ms, c)
 780         }
 781
 782         c.UpdateSubscriptionInDB(subs, removeSubscriptionFromDb)
 783         parentTrans.SendEvent(subRfMsg, 0)
 784 }
 785
 786 //-------------------------------------------------------------------
 787 // SUBS DELETE Handling
 788 //-------------------------------------------------------------------
 789
 790 func (c *Control) handleSubscriptionDelete(subs *Subscription, parentTrans *TransactionXapp) {
 791
 792         trans := c.tracker.NewSubsTransaction(subs)
 793         subs.WaitTransactionTurn(trans)
 794         defer subs.ReleaseTransactionTurn(trans)
 795         defer trans.Release()
 796
 797         xapp.Logger.Debug("SUBS-SubDelReq: Handling %s", idstring(nil, trans, subs, parentTrans))
 798
 799         subs.mutex.Lock()
 800
 801         if subs.valid && subs.EpList.HasEndpoint(parentTrans.GetEndpoint()) && subs.EpList.Size() == 1 {
 802                 subs.valid = false
 803                 subs.mutex.Unlock()
 804                 c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
 805         } else {
 806                 subs.mutex.Unlock()
 807         }
 808         //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete)
 809         //  If parallel deletes ongoing both might pass earlier sendE2TSubscriptionDeleteRequest(...) if
 810         //  RemoveFromSubscription locates in caller side (now in handleXAPPSubscriptionDeleteRequest(...))
 811         c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanup_ms, c)
 812         c.registry.UpdateSubscriptionToDb(subs, c)
 813         parentTrans.SendEvent(nil, 0)
 814 }
 815
 816 //-------------------------------------------------------------------
 817 // send to E2T Subscription Request
 818 //-------------------------------------------------------------------
 819 func (c *Control) sendE2TSubscriptionRequest(subs *Subscription, trans *TransactionSubs, parentTrans *TransactionXapp) interface{} {
 820         var err error
 821         var event interface{} = nil
 822         var timedOut bool = false
 823
 824         subReqMsg := subs.SubReqMsg
 825         subReqMsg.RequestId = subs.GetReqId().RequestId
 826         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionRequest(subReqMsg)
 827         if err != nil {
 828                 xapp.Logger.Error("SUBS-SubReq: %s", idstring(err, trans, subs, parentTrans))
 829                 return event
 830         }
 831
 832         // Write uncompleted subscrition in db. If no response for subscrition it need to be re-processed (deleted) after restart
 833         c.WriteSubscriptionToDb(subs)
 834
 835         for retries := uint64(0); retries < e2tMaxSubReqTryCount; retries++ {
 836                 desc := fmt.Sprintf("(retry %d)", retries)
 837                 if retries == 0 {
 838                         c.UpdateCounter(cSubReqToE2)
 839                 } else {
 840                         c.UpdateCounter(cSubReReqToE2)
 841                 }
 842                 c.rmrSendToE2T(desc, subs, trans)
 843                 if subs.DoNotWaitSubResp == false {
 844                         event, timedOut = trans.WaitEvent(e2tSubReqTimeout)
 845                         if timedOut {
 846                                 c.UpdateCounter(cSubReqTimerExpiry)
 847                                 continue
 848                         }
 849                 } else {
 850                         // Simulating case where subscrition request has been sent but response has not been received before restart
 851                         event = &SubmgrRestartTestEvent{}
 852                 }
 853                 break
 854         }
 855         xapp.Logger.Debug("SUBS-SubReq: Response handling event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 856         return event
 857 }
 858
 859 //-------------------------------------------------------------------
 860 // send to E2T Subscription Delete Request
 861 //-------------------------------------------------------------------
 862
 863 func (c *Control) sendE2TSubscriptionDeleteRequest(subs *Subscription, trans *TransactionSubs, parentTrans *TransactionXapp) interface{} {
 864         var err error
 865         var event interface{}
 866         var timedOut bool
 867
 868         subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{}
 869         subDelReqMsg.RequestId = subs.GetReqId().RequestId
 870         subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId
 871         trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg)
 872         if err != nil {
 873                 xapp.Logger.Error("SUBS-SubDelReq: %s", idstring(err, trans, subs, parentTrans))
 874                 return event
 875         }
 876
 877         for retries := uint64(0); retries < e2tMaxSubDelReqTryCount; retries++ {
 878                 desc := fmt.Sprintf("(retry %d)", retries)
 879                 if retries == 0 {
 880                         c.UpdateCounter(cSubDelReqToE2)
 881                 } else {
 882                         c.UpdateCounter(cSubDelReReqToE2)
 883                 }
 884                 c.rmrSendToE2T(desc, subs, trans)
 885                 event, timedOut = trans.WaitEvent(e2tSubDelReqTime)
 886                 if timedOut {
 887                         c.UpdateCounter(cSubDelReqTimerExpiry)
 888                         continue
 889                 }
 890                 break
 891         }
 892         xapp.Logger.Debug("SUBS-SubDelReq: Response handling event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
 893         return event
 894 }
 895
 896 //-------------------------------------------------------------------
 897 // handle from E2T Subscription Response
 898 //-------------------------------------------------------------------
 899 func (c *Control) handleE2TSubscriptionResponse(params *xapp.RMRParams) {
 900         xapp.Logger.Info("MSG from E2T: %s", params.String())
 901         c.UpdateCounter(cSubRespFromE2)
 902
 903         subRespMsg, err := c.e2ap.UnpackSubscriptionResponse(params.Payload)
 904         if err != nil {
 905                 xapp.Logger.Error("MSG-SubResp %s", idstring(err, params))
 906                 return
 907         }
 908         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subRespMsg.RequestId.InstanceId})
 909         if err != nil {
 910                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, params))
 911                 return
 912         }
 913         trans := subs.GetTransaction()
 914         if trans == nil {
 915                 err = fmt.Errorf("Ongoing transaction not found")
 916                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, params, subs))
 917                 return
 918         }
 919         sendOk, timedOut := trans.SendEvent(subRespMsg, e2tRecvMsgTimeout)
 920         if sendOk == false {
 921                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 922                 xapp.Logger.Error("MSG-SubResp: %s", idstring(err, trans, subs))
 923         }
 924         return
 925 }
 926
 927 //-------------------------------------------------------------------
 928 // handle from E2T Subscription Failure
 929 //-------------------------------------------------------------------
 930 func (c *Control) handleE2TSubscriptionFailure(params *xapp.RMRParams) {
 931         xapp.Logger.Info("MSG from E2T: %s", params.String())
 932         c.UpdateCounter(cSubFailFromE2)
 933         subFailMsg, err := c.e2ap.UnpackSubscriptionFailure(params.Payload)
 934         if err != nil {
 935                 xapp.Logger.Error("MSG-SubFail %s", idstring(err, params))
 936                 return
 937         }
 938         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subFailMsg.RequestId.InstanceId})
 939         if err != nil {
 940                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, params))
 941                 return
 942         }
 943         trans := subs.GetTransaction()
 944         if trans == nil {
 945                 err = fmt.Errorf("Ongoing transaction not found")
 946                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, params, subs))
 947                 return
 948         }
 949         sendOk, timedOut := trans.SendEvent(subFailMsg, e2tRecvMsgTimeout)
 950         if sendOk == false {
 951                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 952                 xapp.Logger.Error("MSG-SubFail: %s", idstring(err, trans, subs))
 953         }
 954         return
 955 }
 956
 957 //-------------------------------------------------------------------
 958 // handle from E2T Subscription Delete Response
 959 //-------------------------------------------------------------------
 960 func (c *Control) handleE2TSubscriptionDeleteResponse(params *xapp.RMRParams) (err error) {
 961         xapp.Logger.Info("MSG from E2T: %s", params.String())
 962         c.UpdateCounter(cSubDelRespFromE2)
 963         subDelRespMsg, err := c.e2ap.UnpackSubscriptionDeleteResponse(params.Payload)
 964         if err != nil {
 965                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params))
 966                 return
 967         }
 968         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelRespMsg.RequestId.InstanceId})
 969         if err != nil {
 970                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params))
 971                 return
 972         }
 973         trans := subs.GetTransaction()
 974         if trans == nil {
 975                 err = fmt.Errorf("Ongoing transaction not found")
 976                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, params, subs))
 977                 return
 978         }
 979         sendOk, timedOut := trans.SendEvent(subDelRespMsg, e2tRecvMsgTimeout)
 980         if sendOk == false {
 981                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
 982                 xapp.Logger.Error("MSG-SubDelResp: %s", idstring(err, trans, subs))
 983         }
 984         return
 985 }
 986
 987 //-------------------------------------------------------------------
 988 // handle from E2T Subscription Delete Failure
 989 //-------------------------------------------------------------------
 990 func (c *Control) handleE2TSubscriptionDeleteFailure(params *xapp.RMRParams) {
 991         xapp.Logger.Info("MSG from E2T: %s", params.String())
 992         c.UpdateCounter(cSubDelFailFromE2)
 993         subDelFailMsg, err := c.e2ap.UnpackSubscriptionDeleteFailure(params.Payload)
 994         if err != nil {
 995                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
 996                 return
 997         }
 998         subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{subDelFailMsg.RequestId.InstanceId})
 999         if err != nil {
1000                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params))
1001                 return
1002         }
1003         trans := subs.GetTransaction()
1004         if trans == nil {
1005                 err = fmt.Errorf("Ongoing transaction not found")
1006                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, params, subs))
1007                 return
1008         }
1009         sendOk, timedOut := trans.SendEvent(subDelFailMsg, e2tRecvMsgTimeout)
1010         if sendOk == false {
1011                 err = fmt.Errorf("Passing event to transaction failed: sendOk(%t) timedOut(%t)", sendOk, timedOut)
1012                 xapp.Logger.Error("MSG-SubDelFail: %s", idstring(err, trans, subs))
1013         }
1014         return
1015 }
1016
1017 //-------------------------------------------------------------------
1018 //
1019 //-------------------------------------------------------------------
1020 func typeofSubsMessage(v interface{}) string {
1021         if v == nil {
1022                 return "NIL"
1023         }
1024         switch v.(type) {
1025         //case *e2ap.E2APSubscriptionRequest:
1026         //      return "SubReq"
1027         case *e2ap.E2APSubscriptionResponse:
1028                 return "SubResp"
1029         case *e2ap.E2APSubscriptionFailure:
1030                 return "SubFail"
1031         //case *e2ap.E2APSubscriptionDeleteRequest:
1032         //      return "SubDelReq"
1033         case *e2ap.E2APSubscriptionDeleteResponse:
1034                 return "SubDelResp"
1035         case *e2ap.E2APSubscriptionDeleteFailure:
1036                 return "SubDelFail"
1037         default:
1038                 return "Unknown"
1039         }
1040 }
1041
1042 //-------------------------------------------------------------------
1043 //
1044 //-------------------------------------------------------------------
1045 func (c *Control) WriteSubscriptionToDb(subs *Subscription) {
1046         xapp.Logger.Debug("WriteSubscriptionToDb() subId = %v", subs.ReqId.InstanceId)
1047         err := c.WriteSubscriptionToSdl(subs.ReqId.InstanceId, subs)
1048         if err != nil {
1049                 xapp.Logger.Error("%v", err)
1050         }
1051 }
1052
1053 //-------------------------------------------------------------------
1054 //
1055 //-------------------------------------------------------------------
1056 func (c *Control) UpdateSubscriptionInDB(subs *Subscription, removeSubscriptionFromDb bool) {
1057
1058         if removeSubscriptionFromDb == true {
1059                 // Subscription was written in db already when subscription request was sent to BTS, except for merged request
1060                 c.RemoveSubscriptionFromDb(subs)
1061         } else {
1062                 // Update is needed for successful response and merge case here
1063                 if subs.RetryFromXapp == false {
1064                         c.WriteSubscriptionToDb(subs)
1065                 }
1066         }
1067         subs.RetryFromXapp = false
1068 }
1069
1070 //-------------------------------------------------------------------
1071 //
1072 //-------------------------------------------------------------------
1073 func (c *Control) RemoveSubscriptionFromDb(subs *Subscription) {
1074         xapp.Logger.Debug("RemoveSubscriptionFromDb() subId = %v", subs.ReqId.InstanceId)
1075         err := c.RemoveSubscriptionFromSdl(subs.ReqId.InstanceId)
1076         if err != nil {
1077                 xapp.Logger.Error("%v", err)
1078         }
1079 }
1080
1081 func (c *Control) SendSubscriptionDeleteReq(subs *Subscription) {
1082
1083         xapp.Logger.Debug("Sending subscription delete due to restart. subId = %v", subs.ReqId.InstanceId)
1084
1085         // Send delete for every endpoint in the subscription
1086         subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{}
1087         subDelReqMsg.RequestId = subs.GetReqId().RequestId
1088         subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId
1089         mType, payload, err := c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg)
1090         if err != nil {
1091                 xapp.Logger.Error("SendSubscriptionDeleteReq() %s", idstring(err))
1092                 return
1093         }
1094         for _, endPoint := range subs.EpList.Endpoints {
1095                 params := &xapp.RMRParams{}
1096                 params.Mtype = mType
1097                 params.SubId = int(subs.GetReqId().InstanceId)
1098                 params.Xid = ""
1099                 params.Meid = subs.Meid
1100                 params.Src = endPoint.String()
1101                 params.PayloadLen = len(payload.Buf)
1102                 params.Payload = payload.Buf
1103                 params.Mbuf = nil
1104                 subs.DeleteFromDb = true
1105                 c.handleXAPPSubscriptionDeleteRequest(params)
1106         }
1107 }
1108
1109 func (c *Control) PrintRESTSubscriptionRequest(p *models.SubscriptionParams) {
1110
1111         fmt.Println("CRESTSubscriptionRequest")
1112         fmt.Printf("  ClientEndpoint.Host = %s\n", p.ClientEndpoint.Host)
1113
1114         if p.ClientEndpoint.HTTPPort != nil {
1115                 fmt.Printf("  ClientEndpoint.HTTPPort = %v\n", *p.ClientEndpoint.HTTPPort)
1116         } else {
1117                 fmt.Println("  ClientEndpoint.HTTPPort = nil")
1118         }
1119
1120         if p.ClientEndpoint.RMRPort != nil {
1121                 fmt.Printf("  ClientEndpoint.RMRPort = %v\n", *p.ClientEndpoint.RMRPort)
1122         } else {
1123                 fmt.Println("  ClientEndpoint.RMRPort = nil")
1124         }
1125
1126         if p.Meid != nil {
1127                 fmt.Printf("  Meid = %s\n", *p.Meid)
1128         } else {
1129                 fmt.Println("  Meid = nil")
1130         }
1131
1132         for _, subscriptionDetail := range p.SubscriptionDetails {
1133                 if p.RANFunctionID != nil {
1134                         fmt.Printf("  RANFunctionID = %v\n", *p.RANFunctionID)
1135                 } else {
1136                         fmt.Println("  RANFunctionID = nil")
1137                 }
1138                 fmt.Printf("  SubscriptionDetail.RequestorID = %v\n", *subscriptionDetail.RequestorID)
1139                 fmt.Printf("  SubscriptionDetail.InstanceID = %v\n", *subscriptionDetail.InstanceID)
1140                 fmt.Printf("  SubscriptionDetail.EventTriggers.OctetString = %X\n", subscriptionDetail.EventTriggers.OctetString)
1141
1142                 for _, actionToBeSetup := range subscriptionDetail.ActionToBeSetupList {
1143                         fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionID = %v\n", *actionToBeSetup.ActionID)
1144                         fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionType = %s\n", *actionToBeSetup.ActionType)
1145                         if actionToBeSetup.ActionDefinition != nil {
1146                                 fmt.Printf("  SubscriptionDetail.ActionToBeSetup.ActionDefinition.OctetString = %X\n", actionToBeSetup.ActionDefinition.OctetString)
1147                         } else {
1148                                 fmt.Println("  SubscriptionDetail.ActionToBeSetup.ActionDefinition = nil")
1149
1150                         }
1151                         if actionToBeSetup.SubsequentAction != nil {
1152                                 fmt.Printf("  SubscriptionDetail.ActionToBeSetup.SubsequentAction.SubsequentActionType = %s\n", *actionToBeSetup.SubsequentAction.SubsequentActionType)
1153                                 fmt.Printf("  SubscriptionDetail.ActionToBeSetup..SubsequentAction.TimeToWait = %s\n", *actionToBeSetup.SubsequentAction.TimeToWait)
1154                         } else {
1155                                 fmt.Println("  SubscriptionDetail.ActionToBeSetup.SubsequentAction = nil")
1156                         }
1157                 }
1158         }
1159 }