pkg/control/tracker.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 TransactionXappKey struct {
  33         RmrEndpoint
  34         Xid string // xapp xid in req
  35 }
  36
  37 func (key *TransactionXappKey) String() string {
  38         return key.RmrEndpoint.String() + "/" + key.Xid
  39 }
  40
  41 //-----------------------------------------------------------------------------
  42 //
  43 //-----------------------------------------------------------------------------
  44 type Transaction struct {
  45         tracker           *Tracker // tracker instance
  46         Subs              *Subscription
  47         RmrEndpoint       RmrEndpoint
  48         Xid               string          // xapp xid in req
  49         OrigParams        *xapp.RMRParams // request orginal params
  50         RespReceived      bool
  51         ForwardRespToXapp bool
  52         mutex             sync.Mutex
  53 }
  54
  55 func (t *Transaction) String() string {
  56         t.mutex.Lock()
  57         defer t.mutex.Unlock()
  58         var subId string = "?"
  59         if t.Subs != nil {
  60                 subId = strconv.FormatUint(uint64(t.Subs.Seq), 10)
  61         }
  62         return subId + "/" + t.RmrEndpoint.String() + "/" + t.Xid
  63 }
  64
  65 func (t *Transaction) CheckResponseReceived() bool {
  66         t.mutex.Lock()
  67         defer t.mutex.Unlock()
  68         if t.RespReceived == false {
  69                 t.RespReceived = true
  70                 return false
  71         }
  72         return true
  73 }
  74
  75 func (t *Transaction) RetryTransaction() {
  76         t.mutex.Lock()
  77         defer t.mutex.Unlock()
  78         t.RespReceived = false
  79 }
  80
  81 func (t *Transaction) Release() {
  82         t.mutex.Lock()
  83         defer t.mutex.Unlock()
  84         if t.Subs != nil {
  85                 t.Subs.UnSetTransaction(t)
  86         }
  87         if t.tracker != nil {
  88                 xappkey := TransactionXappKey{t.RmrEndpoint, t.Xid}
  89                 t.tracker.UnTrackTransaction(xappkey)
  90         }
  91         t.Subs = nil
  92         t.tracker = nil
  93 }
  94
  95 //-----------------------------------------------------------------------------
  96 //
  97 //-----------------------------------------------------------------------------
  98 type Tracker struct {
  99         transactionXappTable map[TransactionXappKey]*Transaction
 100         mutex                sync.Mutex
 101 }
 102
 103 func (t *Tracker) Init() {
 104         t.transactionXappTable = make(map[TransactionXappKey]*Transaction)
 105 }
 106
 107 func (t *Tracker) TrackTransaction(subs *Subscription, endpoint RmrEndpoint, params *xapp.RMRParams, respReceived bool, forwardRespToXapp bool) (*Transaction, error) {
 108
 109         trans := &Transaction{
 110                 tracker:           nil,
 111                 Subs:              nil,
 112                 RmrEndpoint:       endpoint,
 113                 Xid:               params.Xid,
 114                 OrigParams:        params,
 115                 RespReceived:      respReceived,
 116                 ForwardRespToXapp: forwardRespToXapp,
 117         }
 118
 119         t.mutex.Lock()
 120         defer t.mutex.Unlock()
 121
 122         xappkey := TransactionXappKey{endpoint, params.Xid}
 123         if _, ok := t.transactionXappTable[xappkey]; ok {
 124                 err := fmt.Errorf("Tracker: Similar transaction with xappkey %s is ongoing, transaction %s not created ", xappkey, trans)
 125                 return nil, err
 126         }
 127
 128         if subs.SetTransaction(trans) == false {
 129                 othTrans := subs.GetTransaction()
 130                 err := fmt.Errorf("Tracker: Subscription %s got already transaction ongoing: %s, transaction %s not created", subs, othTrans, trans)
 131                 return nil, err
 132         }
 133         trans.Subs = subs
 134         if (trans.Subs.RmrEndpoint.Addr != trans.RmrEndpoint.Addr) || (trans.Subs.RmrEndpoint.Port != trans.RmrEndpoint.Port) {
 135                 err := fmt.Errorf("Tracker: Subscription endpoint %s mismatch with trans: %s", subs, trans)
 136                 trans.Subs.UnSetTransaction(nil)
 137                 return nil, err
 138         }
 139
 140         trans.tracker = t
 141         t.transactionXappTable[xappkey] = trans
 142         return trans, nil
 143 }
 144
 145 func (t *Tracker) UnTrackTransaction(xappKey TransactionXappKey) (*Transaction, error) {
 146         t.mutex.Lock()
 147         defer t.mutex.Unlock()
 148         if trans, ok2 := t.transactionXappTable[xappKey]; ok2 {
 149                 delete(t.transactionXappTable, xappKey)
 150                 return trans, nil
 151         }
 152         return nil, fmt.Errorf("Tracker: No record for xappkey %s", xappKey)
 153 }