| /* |
| ================================================================================== |
| Copyright (c) 2019 AT&T Intellectual Property. |
| Copyright (c) 2019 Nokia |
| |
| Licensed under the Apache License, Version 2.0 (the "License"); |
| you may not use this file except in compliance with the License. |
| You may obtain a copy of the License at |
| |
| http://www.apache.org/licenses/LICENSE-2.0 |
| |
| Unless required by applicable law or agreed to in writing, software |
| distributed under the License is distributed on an "AS IS" BASIS, |
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| See the License for the specific language governing permissions and |
| limitations under the License. |
| ================================================================================== |
| */ |
| |
| package control |
| |
| import ( |
| "encoding/json" |
| "fmt" |
| "strings" |
| "sync" |
| "time" |
| |
| "gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap" |
| "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models" |
| "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" |
| ) |
| |
| //----------------------------------------------------------------------------- |
| // |
| //----------------------------------------------------------------------------- |
| |
| type RESTSubscription struct { |
| Created string |
| xAppServiceName string |
| xAppRmrEndPoint string |
| Meid string |
| InstanceIds []uint32 |
| xAppIdToE2Id map[int64]int64 |
| SubReqOngoing bool |
| SubDelReqOngoing bool |
| lastReqMd5sum string |
| } |
| |
| func (r *RESTSubscription) AddE2InstanceId(instanceId uint32) { |
| |
| for _, v := range r.InstanceIds { |
| if v == instanceId { |
| return |
| } |
| |
| } |
| |
| r.InstanceIds = append(r.InstanceIds, instanceId) |
| } |
| |
| func (r *RESTSubscription) AddMd5Sum(md5sum string) { |
| if md5sum != "" { |
| r.lastReqMd5sum = md5sum |
| } else { |
| xapp.Logger.Error("EMPTY md5sum attempted to be add to subscrition") |
| } |
| } |
| |
| func (r *RESTSubscription) DeleteE2InstanceId(instanceId uint32) { |
| r.InstanceIds = r.InstanceIds[1:] |
| } |
| |
| func (r *RESTSubscription) AddXappIdToE2Id(xAppEventInstanceID int64, e2EventInstanceID int64) { |
| r.xAppIdToE2Id[xAppEventInstanceID] = e2EventInstanceID |
| } |
| |
| func (r *RESTSubscription) GetE2IdFromXappIdToE2Id(xAppEventInstanceID int64) int64 { |
| return r.xAppIdToE2Id[xAppEventInstanceID] |
| } |
| |
| func (r *RESTSubscription) DeleteXappIdToE2Id(xAppEventInstanceID int64) { |
| delete(r.xAppIdToE2Id, xAppEventInstanceID) |
| } |
| |
| func (r *RESTSubscription) SetProcessed(err error) { |
| r.SubReqOngoing = false |
| if err != nil { |
| r.lastReqMd5sum = "" |
| } |
| } |
| |
| type Registry struct { |
| mutex *sync.Mutex |
| register map[uint32]*Subscription |
| subIds []uint32 |
| rtmgrClient *RtmgrClient |
| restSubscriptions map[string]*RESTSubscription |
| } |
| |
| func (r *Registry) Initialize() { |
| r.mutex = new(sync.Mutex) |
| r.register = make(map[uint32]*Subscription) |
| r.restSubscriptions = make(map[string]*RESTSubscription) |
| |
| var i uint32 |
| for i = 1; i < 65535; i++ { |
| r.subIds = append(r.subIds, i) |
| } |
| } |
| |
| func (r *Registry) GetAllRestSubscriptionsJson() []byte { |
| |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| restSubscriptionsJson, err := json.Marshal(r.restSubscriptions) |
| if err != nil { |
| xapp.Logger.Error("GetAllRestSubscriptions() json.Marshal error: %v", err) |
| } |
| return restSubscriptionsJson |
| } |
| |
| func (r *Registry) GetAllE2NodeRestSubscriptionsJson(ranName string) []byte { |
| |
| restSubscriptions := r.GetAllE2NodeRestSubscriptions(ranName) |
| e2NodeRestSubscriptionsJson, err := json.Marshal(restSubscriptions) |
| if err != nil { |
| xapp.Logger.Error("GetE2NodeRestSubscriptions() json.Marshal error: %v", err) |
| } |
| return e2NodeRestSubscriptionsJson |
| } |
| |
| func (r *Registry) GetAllE2NodeRestSubscriptions(ranName string) map[string]RESTSubscription { |
| |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| var restSubscriptions map[string]RESTSubscription |
| restSubscriptions = make(map[string]RESTSubscription) |
| for restSubsId, restSubscription := range r.restSubscriptions { |
| if restSubscription.Meid == ranName { |
| restSubscriptions[restSubsId] = *restSubscription |
| } |
| } |
| return restSubscriptions |
| } |
| |
| func (r *Registry) GetAllXappsJson() []byte { |
| |
| r.mutex.Lock() |
| var xappList []string |
| var xappsMap map[string]string |
| xappsMap = make(map[string]string) |
| for _, restSubscription := range r.restSubscriptions { |
| _, ok := xappsMap[restSubscription.xAppServiceName] |
| if !ok { |
| xappsMap[restSubscription.xAppServiceName] = restSubscription.xAppServiceName |
| xappList = append(xappList, restSubscription.xAppServiceName) |
| } |
| } |
| r.mutex.Unlock() |
| |
| xappsJson, err := json.Marshal(xappList) |
| if err != nil { |
| xapp.Logger.Error("GetXapps() json.Marshal error: %v", err) |
| } |
| return xappsJson |
| } |
| |
| func (r *Registry) GetAllXapps() map[string]string { |
| |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| var xappsMap map[string]string |
| xappsMap = make(map[string]string) |
| for _, restSubscription := range r.restSubscriptions { |
| _, ok := xappsMap[restSubscription.xAppServiceName] |
| if !ok { |
| xappsMap[restSubscription.xAppServiceName] = restSubscription.xAppServiceName |
| } |
| } |
| return xappsMap |
| } |
| |
| func (r *Registry) GetAllXappRestSubscriptionsJson(xAppServiceName string) []byte { |
| |
| xappRestSubscriptions := r.GetAllXappRestSubscriptions(xAppServiceName) |
| xappRestSubscriptionsJson, err := json.Marshal(xappRestSubscriptions) |
| if err != nil { |
| xapp.Logger.Error("GetXappRestSubscriptions() json.Marshal error: %v", err) |
| } |
| return xappRestSubscriptionsJson |
| } |
| |
| func (r *Registry) GetAllXappRestSubscriptions(xAppServiceName string) map[string]RESTSubscription { |
| |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| var xappRestSubscriptions map[string]RESTSubscription |
| xappRestSubscriptions = make(map[string]RESTSubscription) |
| for restSubsId, xappRestSubscription := range r.restSubscriptions { |
| if xappRestSubscription.xAppServiceName == xAppServiceName { |
| xappRestSubscriptions[restSubsId] = *xappRestSubscription |
| } |
| } |
| return xappRestSubscriptions |
| } |
| |
| func (r *Registry) GetE2SubscriptionsJson(restSubsId string) ([]byte, error) { |
| |
| // Get all E2 subscriptions of a REST subscription |
| restSubs, err := r.GetRESTSubscription(restSubsId, false) |
| if err != nil { |
| return nil, err |
| } |
| |
| r.mutex.Lock() |
| var e2Subscriptions []Subscription |
| for _, e2SubId := range restSubs.InstanceIds { |
| e2Subscription, ok := r.register[e2SubId] |
| if ok { |
| e2Subscriptions = append(e2Subscriptions, *e2Subscription) |
| } |
| } |
| r.mutex.Unlock() |
| e2SubscriptionsJson, err := json.Marshal(e2Subscriptions) |
| if err != nil { |
| xapp.Logger.Error("GetE2Subscriptions() json.Marshal error: %v", err) |
| } |
| return e2SubscriptionsJson, nil |
| } |
| |
| func (r *Registry) CreateRESTSubscription(restSubId *string, xappServiceName *string, xAppRmrEndPoint *string, maid *string) *RESTSubscription { |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| newRestSubscription := RESTSubscription{} |
| newRestSubscription.Created = time.Now().Format("2006-01-02 15:04:05.000") |
| newRestSubscription.xAppServiceName = *xappServiceName |
| newRestSubscription.xAppRmrEndPoint = *xAppRmrEndPoint |
| newRestSubscription.Meid = *maid |
| newRestSubscription.SubReqOngoing = true |
| newRestSubscription.SubDelReqOngoing = false |
| r.restSubscriptions[*restSubId] = &newRestSubscription |
| newRestSubscription.xAppIdToE2Id = make(map[int64]int64) |
| xapp.Logger.Debug("Registry: Created REST subscription successfully. restSubId=%v, subscriptionCount=%v, e2apSubscriptionCount=%v", *restSubId, len(r.restSubscriptions), len(r.register)) |
| return &newRestSubscription |
| } |
| |
| func (r *Registry) DeleteRESTSubscription(restSubId *string) { |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| delete(r.restSubscriptions, *restSubId) |
| xapp.Logger.Debug("Registry: Deleted REST subscription successfully. restSubId=%v, subscriptionCount=%v", *restSubId, len(r.restSubscriptions)) |
| } |
| |
| func (r *Registry) GetRESTSubscription(restSubId string, IsDelReqOngoing bool) (*RESTSubscription, error) { |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| if restSubscription, ok := r.restSubscriptions[restSubId]; ok { |
| // Subscription deletion is not allowed if prosessing subscription request in not ready |
| if restSubscription.SubDelReqOngoing == false && restSubscription.SubReqOngoing == false { |
| if IsDelReqOngoing == true { |
| restSubscription.SubDelReqOngoing = true |
| } |
| r.restSubscriptions[restSubId] = restSubscription |
| return restSubscription, nil |
| } else { |
| return restSubscription, fmt.Errorf("Registry: REST request is still ongoing for the endpoint=%v, restSubId=%v, SubDelReqOngoing=%v, SubReqOngoing=%v", restSubscription, restSubId, restSubscription.SubDelReqOngoing, restSubscription.SubReqOngoing) |
| } |
| } |
| return nil, fmt.Errorf("Registry: No valid subscription found with restSubId=%v", restSubId) |
| } |
| |
| func (r *Registry) QueryHandler() (models.SubscriptionList, error) { |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| |
| resp := models.SubscriptionList{} |
| for _, subs := range r.register { |
| subs.mutex.Lock() |
| resp = append(resp, &models.SubscriptionData{SubscriptionID: int64(subs.ReqId.InstanceId), Meid: subs.Meid.RanName, ClientEndpoint: subs.EpList.StringList()}) |
| subs.mutex.Unlock() |
| } |
| return resp, nil |
| } |
| |
| func (r *Registry) allocateSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest, resetTestFlag bool, rmrRoutecreated bool) (*Subscription, error) { |
| if len(r.subIds) > 0 { |
| subId := r.subIds[0] |
| r.subIds = r.subIds[1:] |
| if _, ok := r.register[subId]; ok == true { |
| r.subIds = append(r.subIds, subId) |
| return nil, fmt.Errorf("Registry: Failed to reserve subscription exists") |
| } |
| subs := &Subscription{ |
| registry: r, |
| Meid: trans.Meid, |
| RMRRouteCreated: rmrRoutecreated, |
| SubReqMsg: subReqMsg, |
| OngoingReqCount: 0, |
| OngoingDelCount: 0, |
| valid: true, |
| PolicyUpdate: false, |
| RetryFromXapp: false, |
| SubRespRcvd: false, |
| DeleteFromDb: false, |
| NoRespToXapp: false, |
| DoNotWaitSubResp: false, |
| } |
| subs.ReqId.Id = subReqMsg.RequestId.Id |
| subs.ReqId.InstanceId = subId |
| r.SetResetTestFlag(resetTestFlag, subs) |
| |
| if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false { |
| r.subIds = append(r.subIds, subs.ReqId.InstanceId) |
| return nil, fmt.Errorf("Registry: Endpoint existing already in subscription") |
| } |
| return subs, nil |
| } |
| return nil, fmt.Errorf("Registry: Failed to reserve subscription no free ids") |
| } |
| |
| func (r *Registry) findExistingSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, bool) { |
| |
| for _, subs := range r.register { |
| if subs.IsMergeable(trans, subReqMsg) { |
| |
| // |
| // check if there has been race conditions |
| // |
| subs.mutex.Lock() |
| //subs has been set to invalid |
| if subs.valid == false { |
| subs.mutex.Unlock() |
| continue |
| } |
| // If size is zero, entry is to be deleted |
| if subs.EpList.Size() == 0 { |
| subs.mutex.Unlock() |
| continue |
| } |
| // Try to add to endpointlist. Adding fails if endpoint is already in the list |
| if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false { |
| subs.mutex.Unlock() |
| xapp.Logger.Debug("Registry: Subs with requesting endpoint found. %s for %s", subs.String(), trans.String()) |
| return subs, true |
| } |
| subs.mutex.Unlock() |
| |
| xapp.Logger.Debug("Registry: Mergeable subs found. %s for %s", subs.String(), trans.String()) |
| return subs, false |
| } |
| } |
| return nil, false |
| } |
| |
| func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest, resetTestFlag bool, c *Control, createRMRRoute bool) (*Subscription, ErrorInfo, error) { |
| var err error |
| var newAlloc bool |
| errorInfo := ErrorInfo{} |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| |
| // |
| // Check validity of subscription action types |
| // |
| actionType, err := r.CheckActionTypes(subReqMsg) |
| if err != nil { |
| xapp.Logger.Debug("CREATE %s", err) |
| err = fmt.Errorf("E2 content validation failed") |
| return nil, errorInfo, err |
| } |
| |
| // |
| // Find possible existing Policy subscription |
| // |
| if actionType == e2ap.E2AP_ActionTypePolicy { |
| if subs, ok := r.register[trans.GetSubId()]; ok { |
| xapp.Logger.Debug("CREATE %s. Existing subscription for Policy found.", subs.String()) |
| // Update message data to subscription |
| subs.SubReqMsg = subReqMsg |
| subs.PolicyUpdate = true |
| subs.SetCachedResponse(nil, true) |
| r.SetResetTestFlag(resetTestFlag, subs) |
| return subs, errorInfo, nil |
| } |
| } |
| |
| subs, endPointFound := r.findExistingSubs(trans, subReqMsg) |
| if subs == nil { |
| if subs, err = r.allocateSubs(trans, subReqMsg, resetTestFlag, createRMRRoute); err != nil { |
| xapp.Logger.Error("%s", err.Error()) |
| err = fmt.Errorf("subscription not allocated") |
| return nil, errorInfo, err |
| } |
| newAlloc = true |
| } else if endPointFound == true { |
| // Requesting endpoint is already present in existing subscription. This can happen if xApp is restarted. |
| subs.RetryFromXapp = true |
| xapp.Logger.Debug("CREATE subReqMsg.InstanceId=%v. Same subscription %s already exists.", subReqMsg.InstanceId, subs.String()) |
| c.UpdateCounter(cDuplicateE2SubReq) |
| return subs, errorInfo, nil |
| } |
| |
| // |
| // Add to subscription |
| // |
| subs.mutex.Lock() |
| defer subs.mutex.Unlock() |
| |
| epamount := subs.EpList.Size() |
| xapp.Logger.Debug("AssignToSubscription subs.EpList.Size()=%v", subs.EpList.Size()) |
| |
| r.mutex.Unlock() |
| // |
| // Subscription route updates |
| // |
| if createRMRRoute == true { |
| if epamount == 1 { |
| errorInfo, err = r.RouteCreate(subs, c) |
| } else { |
| errorInfo, err = r.RouteCreateUpdate(subs, c) |
| } |
| } else { |
| xapp.Logger.Debug("RMR route not created: createRMRRoute=%v", createRMRRoute) |
| } |
| r.mutex.Lock() |
| |
| if err != nil { |
| if newAlloc { |
| r.subIds = append(r.subIds, subs.ReqId.InstanceId) |
| } |
| // Delete already added endpoint for the request |
| subs.EpList.DelEndpoint(trans.GetEndpoint()) |
| return nil, errorInfo, err |
| } |
| |
| if newAlloc { |
| r.register[subs.ReqId.InstanceId] = subs |
| } |
| xapp.Logger.Debug("CREATE %s", subs.String()) |
| xapp.Logger.Debug("Registry: substable=%v", r.register) |
| return subs, errorInfo, nil |
| } |
| |
| func (r *Registry) RouteCreate(subs *Subscription, c *Control) (ErrorInfo, error) { |
| errorInfo := ErrorInfo{} |
| subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)} |
| err := r.rtmgrClient.SubscriptionRequestCreate(subRouteAction) |
| if err != nil { |
| if strings.Contains(err.Error(), "status 400") { |
| errorInfo.TimeoutType = models.SubscriptionInstanceTimeoutTypeRTMGRTimeout |
| } else { |
| errorInfo.ErrorSource = models.SubscriptionInstanceErrorSourceRTMGR |
| } |
| errorInfo.ErrorCause = err.Error() |
| c.UpdateCounter(cRouteCreateFail) |
| xapp.Logger.Error("%s", err.Error()) |
| err = fmt.Errorf("RTMGR route create failure") |
| } |
| return errorInfo, err |
| } |
| |
| func (r *Registry) RouteCreateUpdate(subs *Subscription, c *Control) (ErrorInfo, error) { |
| errorInfo := ErrorInfo{} |
| subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)} |
| err := r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction) |
| if err != nil { |
| if strings.Contains(err.Error(), "status 400") { |
| errorInfo.TimeoutType = models.SubscriptionInstanceTimeoutTypeRTMGRTimeout |
| } else { |
| errorInfo.ErrorSource = models.SubscriptionInstanceErrorSourceRTMGR |
| } |
| errorInfo.ErrorCause = err.Error() |
| c.UpdateCounter(cRouteCreateUpdateFail) |
| xapp.Logger.Error("%s", err.Error()) |
| err = fmt.Errorf("RTMGR route update failure") |
| return errorInfo, err |
| } |
| c.UpdateCounter(cMergedSubscriptions) |
| return errorInfo, err |
| } |
| |
| func (r *Registry) CheckActionTypes(subReqMsg *e2ap.E2APSubscriptionRequest) (uint64, error) { |
| var reportFound bool = false |
| var policyFound bool = false |
| var insertFound bool = false |
| |
| for _, acts := range subReqMsg.ActionSetups { |
| if acts.ActionType == e2ap.E2AP_ActionTypeReport { |
| reportFound = true |
| } |
| if acts.ActionType == e2ap.E2AP_ActionTypePolicy { |
| policyFound = true |
| } |
| if acts.ActionType == e2ap.E2AP_ActionTypeInsert { |
| insertFound = true |
| } |
| } |
| if reportFound == true && policyFound == true || reportFound == true && insertFound == true || policyFound == true && insertFound == true { |
| return e2ap.E2AP_ActionTypeInvalid, fmt.Errorf("Different action types (Report, Policy or Insert) in same RICactions-ToBeSetup-List") |
| } |
| if reportFound == true { |
| return e2ap.E2AP_ActionTypeReport, nil |
| } |
| if policyFound == true { |
| return e2ap.E2AP_ActionTypePolicy, nil |
| } |
| if insertFound == true { |
| return e2ap.E2AP_ActionTypeInsert, nil |
| } |
| return e2ap.E2AP_ActionTypeInvalid, fmt.Errorf("Invalid action type in RICactions-ToBeSetup-List") |
| } |
| |
| func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *TransactionXapp, waitRouteClean time.Duration, c *Control) { |
| |
| xapp.Logger.Debug("RemoveFromSubscription %s", idstring(nil, trans, subs, trans)) |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| subs.mutex.Lock() |
| defer subs.mutex.Unlock() |
| |
| delStatus := subs.EpList.DelEndpoint(trans.GetEndpoint()) |
| epamount := subs.EpList.Size() |
| |
| subId := subs.ReqId.InstanceId |
| if delStatus == false { |
| return |
| } |
| |
| if waitRouteClean > 0 { |
| // Wait here that response is delivered to xApp via RMR before route is cleaned |
| xapp.Logger.Debug("Pending %v in order to wait route cleanup", waitRouteClean) |
| r.mutex.Unlock() |
| time.Sleep(waitRouteClean) |
| r.mutex.Lock() |
| } |
| |
| xapp.Logger.Debug("CLEAN %s", subs.String()) |
| |
| if epamount == 0 { |
| // |
| // Subscription route delete |
| // |
| if subs.RMRRouteCreated == true { |
| r.RouteDelete(subs, trans, c) |
| } |
| |
| // Not merged subscription is being deleted |
| xapp.Logger.Debug("Subscription route delete RemoveSubscriptionFromDb") |
| c.RemoveSubscriptionFromDb(subs) |
| |
| // |
| // Subscription release |
| // |
| |
| if _, ok := r.register[subId]; ok { |
| xapp.Logger.Debug("RELEASE %s", subs.String()) |
| delete(r.register, subId) |
| xapp.Logger.Debug("Registry: substable=%v", r.register) |
| } |
| r.subIds = append(r.subIds, subId) |
| } else if subs.EpList.Size() > 0 { |
| // |
| // Subscription route update |
| // |
| if subs.RMRRouteCreated == true { |
| r.RouteDeleteUpdate(subs, c) |
| } |
| |
| // Endpoint of merged subscription is being deleted |
| xapp.Logger.Debug("Subscription route update WriteSubscriptionToDb") |
| err := c.WriteSubscriptionToDb(subs) |
| if err != nil { |
| xapp.Logger.Error("tracker.UnTrackTransaction() failed:%s", err.Error()) |
| } |
| c.UpdateCounter(cUnmergedSubscriptions) |
| } |
| return |
| } |
| |
| func (r *Registry) RouteDelete(subs *Subscription, trans *TransactionXapp, c *Control) { |
| tmpList := xapp.RmrEndpointList{} |
| tmpList.AddEndpoint(trans.GetEndpoint()) |
| subRouteAction := SubRouteInfo{tmpList, uint16(subs.ReqId.InstanceId)} |
| if err := r.rtmgrClient.SubscriptionRequestDelete(subRouteAction); err != nil { |
| c.UpdateCounter(cRouteDeleteFail) |
| } |
| } |
| |
| func (r *Registry) RouteDeleteUpdate(subs *Subscription, c *Control) { |
| subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.InstanceId)} |
| if err := r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction); err != nil { |
| c.UpdateCounter(cRouteDeleteUpdateFail) |
| } |
| } |
| |
| func (r *Registry) GetSubscription(subId uint32) *Subscription { |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| if _, ok := r.register[subId]; ok { |
| return r.register[subId] |
| } |
| return nil |
| } |
| |
| func (r *Registry) GetSubscriptionFirstMatch(subIds []uint32) (*Subscription, error) { |
| r.mutex.Lock() |
| defer r.mutex.Unlock() |
| for _, subId := range subIds { |
| if _, ok := r.register[subId]; ok { |
| return r.register[subId], nil |
| } |
| } |
| return nil, fmt.Errorf("No valid subscription found with subIds %v", subIds) |
| } |
| |
| func (r *Registry) SetResetTestFlag(resetTestFlag bool, subs *Subscription) { |
| if resetTestFlag == true { |
| // This is used in submgr restart unit tests |
| xapp.Logger.Debug("resetTestFlag == true") |
| subs.DoNotWaitSubResp = true |
| } else { |
| xapp.Logger.Debug("resetTestFlag == false") |
| } |
| } |
| |
| func (r *Registry) DeleteAllE2Subscriptions(ranName string, c *Control) { |
| |
| xapp.Logger.Debug("Registry: DeleteAllE2Subscriptions()") |
| for subId, subs := range r.register { |
| if subs.Meid.RanName == ranName { |
| if subs.OngoingReqCount != 0 || subs.OngoingDelCount != 0 { |
| // Subscription creation or deletion processes need to be processed gracefully till the end. |
| // Subscription is deleted at end of the process in both cases. |
| xapp.Logger.Debug("Registry: E2 subscription under prosessing ongoing cannot delete it yet. subId=%v, OngoingReqCount=%v, OngoingDelCount=%v", subId, subs.OngoingReqCount, subs.OngoingDelCount) |
| continue |
| } else { |
| // Delete route |
| if subs.RMRRouteCreated == true { |
| for _, ep := range subs.EpList.Endpoints { |
| tmpList := xapp.RmrEndpointList{} |
| tmpList.AddEndpoint(&ep) |
| subRouteAction := SubRouteInfo{tmpList, uint16(subs.ReqId.InstanceId)} |
| if err := r.rtmgrClient.SubscriptionRequestDelete(subRouteAction); err != nil { |
| c.UpdateCounter(cRouteDeleteFail) |
| } |
| } |
| } |
| // Delete E2 subscription from registry and db |
| xapp.Logger.Debug("Registry: Subscription delete. subId=%v", subId) |
| delete(r.register, subId) |
| r.subIds = append(r.subIds, subId) |
| c.RemoveSubscriptionFromDb(subs) |
| } |
| } |
| } |
| |
| // Delete REST subscription from registry and db |
| for restSubId, restSubs := range r.restSubscriptions { |
| if restSubs.Meid == ranName { |
| if restSubs.SubReqOngoing == true || restSubs.SubDelReqOngoing == true { |
| // Subscription creation or deletion processes need to be processed gracefully till the end. |
| // Subscription is deleted at end of the process in both cases. |
| xapp.Logger.Debug("Registry: REST subscription under prosessing ongoing cannot delete it yet. RestSubId=%v, SubReqOngoing=%v, SubDelReqOngoing=%v", restSubId, restSubs.SubReqOngoing, restSubs.SubDelReqOngoing) |
| continue |
| } else { |
| xapp.Logger.Debug("Registry: REST subscription delete. subId=%v", restSubId) |
| delete(r.restSubscriptions, restSubId) |
| c.RemoveRESTSubscriptionFromDb(restSubId) |
| } |
| } |
| } |
| } |