X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Ftransaction.go;h=df4d7db4147a34c101a05330b2d5a3184bb5dd74;hb=a9bf76cb8dec6e52e7699edf1631c214647f8beb;hp=9adaeca6067090c1cfeb99fef9f11263fa9cb5e1;hpb=31797b49985822f1d402501f16ab2794838bebba;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/transaction.go b/pkg/control/transaction.go index 9adaeca..df4d7db 100644 --- a/pkg/control/transaction.go +++ b/pkg/control/transaction.go @@ -21,64 +21,72 @@ 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 "transkey(" + key.RmrEndpoint.String() + "/" + key.Xid + ")" +type TransactionIf interface { + String() string + Release() + SendEvent(interface{}, time.Duration) (bool, bool) + WaitEvent(time.Duration) (interface{}, bool) } //----------------------------------------------------------------------------- // //----------------------------------------------------------------------------- + type Transaction struct { - mutex sync.Mutex - tracker *Tracker //tracker instance - Subs *Subscription //related subscription - RmrEndpoint RmrEndpoint //xapp endpoint - 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 - Mtype int //Encoded message type to be send - Payload *packer.PackedData //Encoded message to be send - RespReceived bool - ForwardRespToXapp bool -} - -func (t *Transaction) StringImpl() string { - var subId string = "?" - if t.Subs != nil { - subId = strconv.FormatUint(uint64(t.Subs.Seq), 10) - } - return "trans(" + t.RmrEndpoint.String() + "/" + t.Xid + "/" + t.Meid.RanName + "/" + subId + ")" + mutex sync.Mutex // + Seq uint64 //transaction sequence + tracker *Tracker //tracker instance + Meid *xapp.RMRMeid //meid transaction related + ReqId RequestId // + Mtype int //Encoded message type to be send + Payload *e2ap.PackedData //Encoded message to be send + EventChan chan interface{} } func (t *Transaction) String() string { - t.mutex.Lock() - defer t.mutex.Unlock() - return t.StringImpl() + return "trans(" + strconv.FormatUint(uint64(t.Seq), 10) + "/" + t.Meid.RanName + "/" + t.ReqId.String() + ")" } -func (t *Transaction) GetXid() string { +func (t *Transaction) 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 *Transaction) 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) GetReqId() *RequestId { t.mutex.Lock() defer t.mutex.Unlock() - return t.Xid + return &t.ReqId } func (t *Transaction) GetMtype() int { @@ -96,41 +104,94 @@ func (t *Transaction) GetMeid() *xapp.RMRMeid { return nil } -func (t *Transaction) GetSrc() string { +func (t *Transaction) GetPayload() *e2ap.PackedData { t.mutex.Lock() defer t.mutex.Unlock() - return t.RmrEndpoint.String() + return t.Payload +} + +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- +type TransactionSubs struct { + Transaction // } -func (t *Transaction) CheckResponseReceived() bool { +func (t *TransactionSubs) String() string { + return "transsubs(" + t.Transaction.String() + ")" +} + +func (t *TransactionSubs) Release() { + t.mutex.Lock() + xapp.Logger.Debug("RELEASE %s", t.String()) + t.tracker = nil + t.mutex.Unlock() +} + +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- +type TransactionXappKey struct { + RmrEndpoint + Xid string // xapp xid in req +} + +func (key *TransactionXappKey) String() string { + return "transkey(" + key.RmrEndpoint.String() + "/" + key.Xid + ")" +} + +//----------------------------------------------------------------------------- +// +//----------------------------------------------------------------------------- +type TransactionXapp struct { + Transaction // + XappKey *TransactionXappKey // +} + +func (t *TransactionXapp) String() string { + var transkey string = "transkey(N/A)" + if t.XappKey != nil { + transkey = t.XappKey.String() + } + return "transxapp(" + t.Transaction.String() + "/" + transkey + ")" +} + +func (t *TransactionXapp) GetEndpoint() *RmrEndpoint { + t.mutex.Lock() + defer t.mutex.Unlock() + if t.XappKey != nil { + return &t.XappKey.RmrEndpoint + } + return nil +} + +func (t *TransactionXapp) 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 *TransactionXapp) 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() { +func (t *TransactionXapp) Release() { t.mutex.Lock() - subs := t.Subs + xapp.Logger.Debug("RELEASE %s", t.String()) tracker := t.tracker - xappkey := TransactionXappKey{t.RmrEndpoint, t.Xid} - t.Subs = nil + xappkey := t.XappKey t.tracker = nil t.mutex.Unlock() - if subs != nil { - subs.UnSetTransaction(t) - } - if tracker != nil { - tracker.UnTrackTransaction(xappkey) + if tracker != nil && xappkey != nil { + tracker.UnTrackTransaction(*xappkey) } }