X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Ftransaction.go;h=735954ef713558923446a8b1d049ba9cdec73484;hb=fa0156680b0bd8f7300f49c65d2ee7bedaaa0e44;hp=cdfdcfb689c0d900aa9934b0f77c8bd07397a4f1;hpb=375c141ce21767f7bda94ec435fdf833b41fbbd2;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/transaction.go b/pkg/control/transaction.go index cdfdcfb..735954e 100644 --- a/pkg/control/transaction.go +++ b/pkg/control/transaction.go @@ -20,70 +20,61 @@ 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 TransactionXappKey struct { - RmrEndpoint - Xid string // xapp xid in req -} -func (key *TransactionXappKey) String() string { - return key.RmrEndpoint.String() + "/" + key.Xid +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{} } -//----------------------------------------------------------------------------- -// -//----------------------------------------------------------------------------- -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 -} - -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 *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 } - return subId + "/" + t.RmrEndpoint.String() + "/" + t.Xid + t.EventChan <- event + return true, false } -func (t *Transaction) GetXid() string { - t.mutex.Lock() - defer t.mutex.Unlock() - return t.Xid +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 *Transaction) GetMtype() int { +func (t *TransactionBase) GetMtype() int { t.mutex.Lock() defer t.mutex.Unlock() return t.Mtype } -func (t *Transaction) GetMeid() *xapp.RMRMeid { +func (t *TransactionBase) GetMeid() *xapp.RMRMeid { t.mutex.Lock() defer t.mutex.Unlock() if t.Meid != nil { @@ -92,39 +83,76 @@ func (t *Transaction) GetMeid() *xapp.RMRMeid { return nil } -func (t *Transaction) GetSrc() string { +func (t *TransactionBase) GetPayload() *packer.PackedData { t.mutex.Lock() defer t.mutex.Unlock() - return t.RmrEndpoint.String() + return t.Payload +} + +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- +type TransactionXappKey struct { + RmrEndpoint + Xid string // xapp xid in req +} + +func (key *TransactionXappKey) String() string { + return "transkey(" + key.RmrEndpoint.String() + "/" + key.Xid + ")" +} + +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- +type Transaction struct { + TransactionBase // + XappKey *TransactionXappKey // +} + +func (t *Transaction) String() string { + var transkey string = "transkey(N/A)" + if t.XappKey != nil { + transkey = t.XappKey.String() + } + return "trans(" + strconv.FormatUint(uint64(t.Seq), 10) + "/" + t.Meid.RanName + "/" + transkey + ")" } -func (t *Transaction) CheckResponseReceived() bool { +func (t *Transaction) GetEndpoint() *RmrEndpoint { t.mutex.Lock() defer t.mutex.Unlock() - if t.RespReceived == false { - t.RespReceived = true - return false + if t.XappKey != nil { + return &t.XappKey.RmrEndpoint } - return true + return nil } -func (t *Transaction) RetryTransaction() { +func (t *Transaction) GetXid() string { t.mutex.Lock() defer t.mutex.Unlock() - t.RespReceived = false + if t.XappKey != nil { + return t.XappKey.Xid + } + return "" } -func (t *Transaction) Release() { - xapp.Logger.Info("Transaction: Releasing %s", t) +func (t *Transaction) GetSrc() string { 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) + if t.XappKey != nil { + return t.XappKey.RmrEndpoint.String() } - t.Subs = nil + return "" +} + +func (t *Transaction) Release() { + t.mutex.Lock() + xapp.Logger.Debug("Transaction: Release %s", t.String()) + tracker := t.tracker + xappkey := t.XappKey t.tracker = nil + t.mutex.Unlock() + + if tracker != nil && xappkey != nil { + tracker.UnTrackTransaction(*xappkey) + } }