Added duplicate detection changes
[ric-plt/submgr.git] / pkg / control / registry.go
index 33d480a..2495d3b 100644 (file)
@@ -21,32 +21,129 @@ package control
 
 import (
        "fmt"
+       "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"
-       "sync"
-       "time"
 )
 
 //-----------------------------------------------------------------------------
 //
 //-----------------------------------------------------------------------------
 
+type RESTSubscription struct {
+       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
+       mutex             sync.Mutex
+       register          map[uint32]*Subscription
+       subIds            []uint32
+       rtmgrClient       *RtmgrClient
+       restSubscriptions map[string]*RESTSubscription
 }
 
 func (r *Registry) Initialize() {
        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) CreateRESTSubscription(restSubId *string, xAppRmrEndPoint *string, maid *string) (*RESTSubscription, error) {
+       r.mutex.Lock()
+       defer r.mutex.Unlock()
+       newRestSubscription := RESTSubscription{}
+       newRestSubscription.xAppRmrEndPoint = *xAppRmrEndPoint
+       newRestSubscription.Meid = *maid
+       newRestSubscription.SubReqOngoing = true
+       newRestSubscription.SubDelReqOngoing = false
+       r.restSubscriptions[*restSubId] = &newRestSubscription
+       newRestSubscription.xAppIdToE2Id = make(map[int64]int64)
+       xapp.Logger.Info("Registry: Created REST subscription successfully. restSubId=%v, subscriptionCount=%v, e2apSubscriptionCount=%v", *restSubId, len(r.restSubscriptions), len(r.register))
+       return &newRestSubscription, nil
+}
+
+func (r *Registry) DeleteRESTSubscription(restSubId *string) {
+       r.mutex.Lock()
+       defer r.mutex.Unlock()
+       delete(r.restSubscriptions, *restSubId)
+       xapp.Logger.Info("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 restSubscription, nil
+       }
+       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()
@@ -54,7 +151,7 @@ func (r *Registry) QueryHandler() (models.SubscriptionList, error) {
        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, Endpoint: subs.EpList.StringList()})
+               resp = append(resp, &models.SubscriptionData{SubscriptionID: int64(subs.ReqId.InstanceId), Meid: subs.Meid.RanName, ClientEndpoint: subs.EpList.StringList()})
                subs.mutex.Unlock()
        }
        return resp, nil
@@ -79,7 +176,7 @@ func (r *Registry) allocateSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubs
                        NoRespToXapp:     false,
                        DoNotWaitSubResp: false,
                }
-               subs.ReqId.Id = 123
+               subs.ReqId.Id = subReqMsg.RequestId.Id
                subs.ReqId.InstanceId = subId
                if resetTestFlag == true {
                        subs.DoNotWaitSubResp = true
@@ -139,7 +236,8 @@ func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.
        //
        actionType, err := r.CheckActionTypes(subReqMsg)
        if err != nil {
-               xapp.Logger.Debug("CREATE %s", err)
+               xapp.Logger.Info("CREATE %s", err)
+               err = fmt.Errorf("E2 content validation failed")
                return nil, err
        }
 
@@ -159,13 +257,16 @@ func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.
        subs, endPointFound := r.findExistingSubs(trans, subReqMsg)
        if subs == nil {
                if subs, err = r.allocateSubs(trans, subReqMsg, resetTestFlag); err != nil {
+                       xapp.Logger.Error("%s", err.Error())
+                       err = fmt.Errorf("subscription not allocated")
                        return nil, 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: subscription already exists. %s", subs.String())
+               xapp.Logger.Debug("CREATE subReqMsg.InstanceId=%v. Same subscription %s already exists.", subReqMsg.InstanceId, subs.String())
+               c.UpdateCounter(cDuplicateE2SubReq)
                return subs, nil
        }
 
@@ -176,7 +277,7 @@ func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.
        defer subs.mutex.Unlock()
 
        epamount := subs.EpList.Size()
-       xapp.Logger.Info("AssignToSubscription subs.EpList.Size() = %v", subs.EpList.Size())
+       xapp.Logger.Info("AssignToSubscription subs.EpList.Size()=%v", subs.EpList.Size())
 
        r.mutex.Unlock()
        //
@@ -211,6 +312,8 @@ func (r *Registry) RouteCreate(subs *Subscription, c *Control) error {
        err := r.rtmgrClient.SubscriptionRequestCreate(subRouteAction)
        if err != nil {
                c.UpdateCounter(cRouteCreateFail)
+               xapp.Logger.Error("%s", err.Error())
+               err = fmt.Errorf("RTMGR route create failure")
        }
        return err
 }
@@ -220,6 +323,8 @@ func (r *Registry) RouteCreateUpdate(subs *Subscription, c *Control) error {
        err := r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
        if err != nil {
                c.UpdateCounter(cRouteCreateUpdateFail)
+               xapp.Logger.Error("%s", err.Error())
+               err = fmt.Errorf("RTMGR route update failure")
                return err
        }
        c.UpdateCounter(cMergedSubscriptions)
@@ -275,6 +380,7 @@ func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *Transaction
 
        go func() {
                if waitRouteClean > 0 {
+                       xapp.Logger.Debug("Pending %v in order to wait route cleanup", waitRouteClean)
                        time.Sleep(waitRouteClean)
                }