- // Setting new subscription ID in the RMR header
- params.SubId = int(newSubId)
- xapp.Logger.Info("Forwarding SubReq to E2T: Mtype: %v, SubId: %v, Xid %s, Meid %v", params.Mtype, params.SubId, params.Xid, params.Meid)
- err = c.rmrSend(params)
- if err != nil {
- xapp.Logger.Error("SubReq: Failed to send request to E2T %v, SubId: %v, Xid: %s", err, params.SubId, params.Xid)
+ dbRetryForever = viper.GetString("controls.dbRetryForever")
+ if dbRetryForever == "" {
+ dbRetryForever = "true"
+ xapp.Logger.Debug("WARNING: Using hard coded default value for dbRetryForever")
+ }
+ xapp.Logger.Debug("dbRetryForever= %v", dbRetryForever)
+
+ // Internal cfg parameter, used to define a wait time for RMR route clean-up. None default
+ // value 100ms used currently only in unittests.
+ waitRouteCleanup_ms = viper.GetDuration("controls.waitRouteCleanup_ms") * 1000000
+ if waitRouteCleanup_ms == 0 {
+ waitRouteCleanup_ms = 5000 * 1000000
+ xapp.Logger.Debug("WARNING: Using hard coded default value for waitRouteCleanup_ms")
+ }
+ xapp.Logger.Debug("waitRouteCleanup= %v", waitRouteCleanup_ms)
+}
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func (c *Control) HandleUncompletedSubscriptions(register map[uint32]*Subscription) {
+
+ xapp.Logger.Debug("HandleUncompletedSubscriptions. len(register) = %v", len(register))
+ for subId, subs := range register {
+ if subs.SubRespRcvd == false {
+ // If policy subscription has already been made successfully unsuccessful update should not be deleted.
+ if subs.PolicyUpdate == false {
+ subs.NoRespToXapp = true
+ xapp.Logger.Debug("SendSubscriptionDeleteReq. subId = %v", subId)
+ c.SendSubscriptionDeleteReq(subs)
+ }
+ }
+ }
+}
+
+func (c *Control) ReadyCB(data interface{}) {
+ if c.RMRClient == nil {
+ c.RMRClient = xapp.Rmr
+ }
+}
+
+func (c *Control) Run() {
+ xapp.SetReadyCB(c.ReadyCB, nil)
+ xapp.AddConfigChangeListener(c.ReadConfigParameters)
+ xapp.Run(c)
+}
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func (c *Control) GetOrCreateRestSubscription(p *models.SubscriptionParams, md5sum string, xAppRmrEndpoint string, xAppServiceName string) (*RESTSubscription, string, error) {
+
+ var restSubId string
+ var restSubscription *RESTSubscription
+ var err error
+
+ prevRestSubsId, exists := c.restDuplicateCtrl.GetLastKnownRestSubsIdBasedOnMd5sum(md5sum)
+ if p.SubscriptionID == "" {
+ // Subscription does not contain REST subscription Id
+ if exists {
+ restSubscription, err = c.registry.GetRESTSubscription(prevRestSubsId, false)
+ if restSubscription != nil {
+ // Subscription not found
+ restSubId = prevRestSubsId
+ if err == nil {
+ xapp.Logger.Debug("Existing restSubId %s found by MD5sum %s for a request without subscription ID - using previous subscription", prevRestSubsId, md5sum)
+ } else {
+ xapp.Logger.Debug("Existing restSubId %s found by MD5sum %s for a request without subscription ID - Note: %s", prevRestSubsId, md5sum, err.Error())
+ }
+ } else {
+ xapp.Logger.Debug("None existing restSubId %s referred by MD5sum %s for a request without subscription ID - deleting cached entry", prevRestSubsId, md5sum)
+ c.restDuplicateCtrl.DeleteLastKnownRestSubsIdBasedOnMd5sum(md5sum)
+ }
+ }
+
+ if restSubscription == nil {
+ restSubId = ksuid.New().String()
+ restSubscription = c.registry.CreateRESTSubscription(&restSubId, &xAppServiceName, &xAppRmrEndpoint, p.Meid)
+ }