X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Ftransaction.go;h=a0a260f5f20df41d76d0e830824d75c67a2c5df2;hb=422d018f94aedd9f4c001176b5ff06c786de28eb;hp=cdfdcfb689c0d900aa9934b0f77c8bd07397a4f1;hpb=375c141ce21767f7bda94ec435fdf833b41fbbd2;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/transaction.go b/pkg/control/transaction.go index cdfdcfb..a0a260f 100644 --- a/pkg/control/transaction.go +++ b/pkg/control/transaction.go @@ -20,12 +20,75 @@ package control import ( - "gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap" + "gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/packer" "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp" "strconv" "sync" + "time" ) +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- + +type TransactionBase struct { + mutex sync.Mutex // + Seq uint64 // + tracker *Tracker //tracker instance + Meid *xapp.RMRMeid //meid transaction related + Mtype int //Encoded message type to be send + Payload *packer.PackedData //Encoded message to be send + EventChan chan interface{} +} + +func (t *TransactionBase) SendEvent(event interface{}, waittime time.Duration) (bool, bool) { + if waittime > 0 { + select { + case t.EventChan <- event: + return true, false + case <-time.After(waittime): + return false, true + } + return false, false + } + t.EventChan <- event + return true, false +} + +func (t *TransactionBase) WaitEvent(waittime time.Duration) (interface{}, bool) { + if waittime > 0 { + select { + case event := <-t.EventChan: + return event, false + case <-time.After(waittime): + return nil, true + } + } + event := <-t.EventChan + return event, false +} + +func (t *TransactionBase) GetMtype() int { + t.mutex.Lock() + defer t.mutex.Unlock() + return t.Mtype +} + +func (t *TransactionBase) GetMeid() *xapp.RMRMeid { + t.mutex.Lock() + defer t.mutex.Unlock() + if t.Meid != nil { + return t.Meid + } + return nil +} + +func (t *TransactionBase) GetPayload() *packer.PackedData { + t.mutex.Lock() + defer t.mutex.Unlock() + return t.Payload +} + //----------------------------------------------------------------------------- // //----------------------------------------------------------------------------- @@ -35,96 +98,67 @@ type TransactionXappKey struct { } func (key *TransactionXappKey) String() string { - return key.RmrEndpoint.String() + "/" + key.Xid + return "transkey(" + key.RmrEndpoint.String() + "/" + key.Xid + ")" } //----------------------------------------------------------------------------- // //----------------------------------------------------------------------------- type Transaction struct { - mutex sync.Mutex - tracker *Tracker //tracker instance - Subs *Subscription //related subscription - RmrEndpoint RmrEndpoint //xapp endpoint - Mtype int //type of initiating message - Xid string //xapp xid in req - Meid *xapp.RMRMeid //meid transaction related - SubReqMsg *e2ap.E2APSubscriptionRequest //SubReq TODO: maybe own transactions per type - SubRespMsg *e2ap.E2APSubscriptionResponse //SubResp TODO: maybe own transactions per type - SubFailMsg *e2ap.E2APSubscriptionFailure //SubFail TODO: maybe own transactions per type - SubDelReqMsg *e2ap.E2APSubscriptionDeleteRequest //SubDelReq TODO: maybe own transactions per type - SubDelRespMsg *e2ap.E2APSubscriptionDeleteResponse //SubDelResp TODO: maybe own transactions per type - SubDelFailMsg *e2ap.E2APSubscriptionDeleteFailure //SubDelFail TODO: maybe own transactions per type - Payload []byte //packed message to optimize retransmissions - PayloadLen int //packed message len to optimize retransmissions - RespReceived bool - ForwardRespToXapp bool + TransactionBase // + XappKey *TransactionXappKey // } -func (t *Transaction) String() string { - t.mutex.Lock() - defer t.mutex.Unlock() - var subId string = "?" - if t.Subs != nil { - subId = strconv.FormatUint(uint64(t.Subs.Seq), 10) +func (t *Transaction) StringImpl() string { + var transkey string = "transkey(N/A)" + if t.XappKey != nil { + transkey = t.XappKey.String() } - return subId + "/" + t.RmrEndpoint.String() + "/" + t.Xid -} - -func (t *Transaction) GetXid() string { - t.mutex.Lock() - defer t.mutex.Unlock() - return t.Xid + return "trans(" + strconv.FormatUint(uint64(t.Seq), 10) + "/" + t.Meid.RanName + "/" + transkey + ")" } -func (t *Transaction) GetMtype() int { +func (t *Transaction) String() string { t.mutex.Lock() defer t.mutex.Unlock() - return t.Mtype + return t.StringImpl() } -func (t *Transaction) GetMeid() *xapp.RMRMeid { +func (t *Transaction) GetEndpoint() *RmrEndpoint { t.mutex.Lock() defer t.mutex.Unlock() - if t.Meid != nil { - return t.Meid + if t.XappKey != nil { + return &t.XappKey.RmrEndpoint } return nil } -func (t *Transaction) GetSrc() string { - t.mutex.Lock() - defer t.mutex.Unlock() - return t.RmrEndpoint.String() -} - -func (t *Transaction) CheckResponseReceived() bool { +func (t *Transaction) GetXid() string { t.mutex.Lock() defer t.mutex.Unlock() - if t.RespReceived == false { - t.RespReceived = true - return false + if t.XappKey != nil { + return t.XappKey.Xid } - return true + return "" } -func (t *Transaction) RetryTransaction() { +func (t *Transaction) GetSrc() string { t.mutex.Lock() defer t.mutex.Unlock() - t.RespReceived = false + if t.XappKey != nil { + return t.XappKey.RmrEndpoint.String() + } + return "" } func (t *Transaction) Release() { - xapp.Logger.Info("Transaction: Releasing %s", t) t.mutex.Lock() - defer t.mutex.Unlock() - if t.Subs != nil { - t.Subs.UnSetTransaction(t) - } - if t.tracker != nil { - xappkey := TransactionXappKey{t.RmrEndpoint, t.Xid} - t.tracker.UnTrackTransaction(xappkey) - } - t.Subs = nil + xapp.Logger.Debug("Transaction: Release %s", t.StringImpl()) + tracker := t.tracker + xappkey := t.XappKey t.tracker = nil + t.mutex.Unlock() + + if tracker != nil && xappkey != nil { + tracker.UnTrackTransaction(*xappkey) + } }