X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fsubscription.go;h=45c13ec107f88fc3c67e86049e709d20efd3d2e7;hb=af91f971681842c22dc0fad165c0cbf7ebaa0e4d;hp=6a72f7423f64e3bda3ed34132dd72ccfb0770e5e;hpb=56e0383cad5307302f547a95755c3bcdd9e3251d;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/subscription.go b/pkg/control/subscription.go index 6a72f74..45c13ec 100644 --- a/pkg/control/subscription.go +++ b/pkg/control/subscription.go @@ -20,7 +20,7 @@ package control import ( - "fmt" + "gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap" "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" "strconv" "sync" @@ -30,20 +30,19 @@ import ( // //----------------------------------------------------------------------------- type Subscription struct { - mutex sync.Mutex - registry *Registry - Seq uint16 - Active bool - // - Meid *xapp.RMRMeid - RmrEndpoint // xapp endpoint. Now only one xapp can have relation to single subscription. To be changed in merge - Trans *Transaction + mutex sync.Mutex // Lock + registry *Registry // Registry + Seq uint16 // SubsId + Meid *xapp.RMRMeid // Meid/ RanName + EpList RmrEndpointList // Endpoints + TransLock sync.Mutex // Lock transactions, only one executed per time for subs + TheTrans *Transaction // Ongoing transaction from xapp + SubReqMsg *e2ap.E2APSubscriptionRequest // Subscription information + SubRespMsg *e2ap.E2APSubscriptionResponse // Subscription information } func (s *Subscription) String() string { - s.mutex.Lock() - defer s.mutex.Unlock() - return strconv.FormatUint(uint64(s.Seq), 10) + "/" + s.RmrEndpoint.String() + "/" + s.Meid.RanName + return "subs(" + strconv.FormatUint(uint64(s.Seq), 10) + "/" + s.Meid.RanName + "/" + s.EpList.String() + ")" } func (s *Subscription) GetSubId() uint16 { @@ -61,80 +60,34 @@ func (s *Subscription) GetMeid() *xapp.RMRMeid { return nil } -func (s *Subscription) Confirmed() { - s.mutex.Lock() - defer s.mutex.Unlock() - s.Active = true -} - -func (s *Subscription) UnConfirmed() { - s.mutex.Lock() - defer s.mutex.Unlock() - s.Active = false -} - -func (s *Subscription) IsConfirmed() bool { - s.mutex.Lock() - defer s.mutex.Unlock() - return s.Active -} - -func (s *Subscription) SetTransaction(trans *Transaction) error { - s.mutex.Lock() - defer s.mutex.Unlock() - - subString := strconv.FormatUint(uint64(s.Seq), 10) + "/" + s.RmrEndpoint.String() + "/" + s.Meid.RanName - - if (s.RmrEndpoint.Addr != trans.RmrEndpoint.Addr) || (s.RmrEndpoint.Port != trans.RmrEndpoint.Port) { - return fmt.Errorf("Subscription: %s endpoint mismatch with trans: %s", subString, trans) - } - if s.Trans != nil { - return fmt.Errorf("Subscription: %s trans %s exist, can not register %s", subString, s.Trans, trans) - } - trans.Subs = s - s.Trans = trans - return nil -} - -func (s *Subscription) UnSetTransaction(trans *Transaction) bool { +func (s *Subscription) IsTransactionReserved() bool { s.mutex.Lock() defer s.mutex.Unlock() - if trans == nil || trans == s.Trans { - s.Trans = nil + if s.TheTrans != nil { return true } return false + } func (s *Subscription) GetTransaction() *Transaction { s.mutex.Lock() defer s.mutex.Unlock() - return s.Trans + return s.TheTrans } -func (s *Subscription) updateRouteImpl(act Action) error { - xapp.Logger.Info("Subscription: Starting routing manager route add. SubId: %d, RmrEndpoint: %s", s.Seq, s.RmrEndpoint) - subRouteAction := SubRouteInfo{act, s.RmrEndpoint.Addr, s.RmrEndpoint.Port, s.Seq} - err := s.registry.rtmgrClient.SubscriptionRequestUpdate(subRouteAction) - if err != nil { - return fmt.Errorf("Subscription: Failed to add route. SubId: %d, RmrEndpoint: %s", s.Seq, s.RmrEndpoint) - } - return nil -} - -func (s *Subscription) UpdateRoute(act Action) error { +func (s *Subscription) WaitTransactionTurn(trans *Transaction) { + s.TransLock.Lock() s.mutex.Lock() - defer s.mutex.Unlock() - return s.updateRouteImpl(act) + s.TheTrans = trans + s.mutex.Unlock() } -func (s *Subscription) Release() { - xapp.Logger.Info("Subscription: Releasing %s", s) +func (s *Subscription) ReleaseTransactionTurn(trans *Transaction) { s.mutex.Lock() - defer s.mutex.Unlock() - s.registry.DelSubscription(s.Seq) - err := s.updateRouteImpl(DELETE) - if err != nil { - xapp.Logger.Error("Registry: Failed to del route. SubId: %d, RmrEndpoint: %s", s.Seq, s.RmrEndpoint) + if trans != nil && trans == s.TheTrans { + s.TheTrans = nil } + s.mutex.Unlock() + s.TransLock.Unlock() }