pkg/control/subscription.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         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  25         "strconv"
  26         "sync"
  27 )
  28
  29 //-----------------------------------------------------------------------------
  30 //
  31 //-----------------------------------------------------------------------------
  32 type Subscription struct {
  33         mutex    sync.Mutex
  34         registry *Registry
  35         Seq      uint16
  36         Active   bool
  37         //
  38         Meid        *xapp.RMRMeid
  39         RmrEndpoint // xapp endpoint. Now only one xapp can have relation to single subscription. To be changed in merge
  40         Trans       *Transaction
  41 }
  42
  43 func (s *Subscription) String() string {
  44         s.mutex.Lock()
  45         defer s.mutex.Unlock()
  46         return strconv.FormatUint(uint64(s.Seq), 10) + "/" + s.RmrEndpoint.String() + "/" + s.Meid.RanName
  47 }
  48
  49 func (s *Subscription) GetSubId() uint16 {
  50         s.mutex.Lock()
  51         defer s.mutex.Unlock()
  52         return s.Seq
  53 }
  54
  55 func (s *Subscription) GetMeid() *xapp.RMRMeid {
  56         s.mutex.Lock()
  57         defer s.mutex.Unlock()
  58         if s.Meid != nil {
  59                 return s.Meid
  60         }
  61         return nil
  62 }
  63
  64 func (s *Subscription) Confirmed() {
  65         s.mutex.Lock()
  66         defer s.mutex.Unlock()
  67         s.Active = true
  68 }
  69
  70 func (s *Subscription) UnConfirmed() {
  71         s.mutex.Lock()
  72         defer s.mutex.Unlock()
  73         s.Active = false
  74 }
  75
  76 func (s *Subscription) IsConfirmed() bool {
  77         s.mutex.Lock()
  78         defer s.mutex.Unlock()
  79         return s.Active
  80 }
  81
  82 func (s *Subscription) SetTransaction(trans *Transaction) error {
  83         s.mutex.Lock()
  84         defer s.mutex.Unlock()
  85
  86         subString := strconv.FormatUint(uint64(s.Seq), 10) + "/" + s.RmrEndpoint.String() + "/" + s.Meid.RanName
  87
  88         if (s.RmrEndpoint.Addr != trans.RmrEndpoint.Addr) || (s.RmrEndpoint.Port != trans.RmrEndpoint.Port) {
  89                 return fmt.Errorf("Subscription: %s endpoint mismatch with trans: %s", subString, trans)
  90         }
  91         if s.Trans != nil {
  92                 return fmt.Errorf("Subscription: %s trans %s exist, can not register %s", subString, s.Trans, trans)
  93         }
  94         trans.Subs = s
  95         s.Trans = trans
  96         return nil
  97 }
  98
  99 func (s *Subscription) UnSetTransaction(trans *Transaction) bool {
 100         s.mutex.Lock()
 101         defer s.mutex.Unlock()
 102         if trans == nil || trans == s.Trans {
 103                 s.Trans = nil
 104                 return true
 105         }
 106         return false
 107 }
 108
 109 func (s *Subscription) GetTransaction() *Transaction {
 110         s.mutex.Lock()
 111         defer s.mutex.Unlock()
 112         return s.Trans
 113 }
 114
 115 func (s *Subscription) updateRouteImpl(act Action) error {
 116         xapp.Logger.Info("Subscription: Starting routing manager route add. SubId: %d, RmrEndpoint: %s", s.Seq, s.RmrEndpoint)
 117         subRouteAction := SubRouteInfo{act, s.RmrEndpoint.Addr, s.RmrEndpoint.Port, s.Seq}
 118         err := s.registry.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
 119         if err != nil {
 120                 return fmt.Errorf("Subscription: Failed to add route. SubId: %d, RmrEndpoint: %s", s.Seq, s.RmrEndpoint)
 121         }
 122         return nil
 123 }
 124
 125 func (s *Subscription) UpdateRoute(act Action) error {
 126         s.mutex.Lock()
 127         defer s.mutex.Unlock()
 128         return s.updateRouteImpl(act)
 129 }
 130
 131 func (s *Subscription) Release() {
 132         xapp.Logger.Info("Subscription: Releasing %s", s)
 133         s.mutex.Lock()
 134         defer s.mutex.Unlock()
 135         s.registry.DelSubscription(s.Seq)
 136         err := s.updateRouteImpl(DELETE)
 137         if err != nil {
 138                 xapp.Logger.Error("Registry: Failed to del route. SubId: %d, RmrEndpoint: %s", s.Seq, s.RmrEndpoint)
 139         }
 140 }