X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fcontrol.go;h=1bc2ee86de8588e186c0f19009a3ddc559268de8;hb=31797b49985822f1d402501f16ab2794838bebba;hp=eb2852a861bef50ed4ac8b2c7967c5045e7e3a48;hpb=56e0383cad5307302f547a95755c3bcdd9e3251d;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/control.go b/pkg/control/control.go index eb2852a..1bc2ee8 100755 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -28,7 +28,6 @@ import ( httptransport "github.com/go-openapi/runtime/client" "github.com/go-openapi/strfmt" "github.com/spf13/viper" - "math/rand" "sync" "time" ) @@ -57,8 +56,6 @@ type RMRMeid struct { RanName string } -var seedSN uint16 - const ( CREATE Action = 0 MERGE Action = 1 @@ -71,15 +68,6 @@ func init() { viper.AutomaticEnv() viper.SetEnvPrefix("submgr") viper.AllowEmptyEnv(true) - seedSN = uint16(viper.GetInt("seed_sn")) - if seedSN == 0 { - rand.Seed(time.Now().UnixNano()) - seedSN = uint16(rand.Intn(65535)) - } - if seedSN > 65535 { - seedSN = 0 - } - xapp.Logger.Info("SUBMGR: Initial Sequence Number: %v", seedSN) } func NewControl() *Control { @@ -91,7 +79,7 @@ func NewControl() *Control { rtmgrClient := RtmgrClient{client, handle, deleteHandle} registry := new(Registry) - registry.Initialize(seedSN) + registry.Initialize() registry.rtmgrClient = &rtmgrClient tracker := new(Tracker) @@ -134,29 +122,29 @@ func (c *Control) rmrSendRaw(desc string, params *RMRParams) (err error) { return } -func (c *Control) rmrSend(desc string, subs *Subscription, trans *Transaction, payload []byte, payloadLen int) (err error) { +func (c *Control) rmrSend(desc string, subs *Subscription, trans *Transaction) (err error) { params := &RMRParams{&xapp.RMRParams{}} params.Mtype = trans.GetMtype() params.SubId = int(subs.GetSubId()) params.Xid = "" params.Meid = subs.GetMeid() params.Src = "" - params.PayloadLen = payloadLen - params.Payload = payload + params.PayloadLen = len(trans.Payload.Buf) + params.Payload = trans.Payload.Buf params.Mbuf = nil return c.rmrSendRaw(desc, params) } -func (c *Control) rmrReplyToSender(desc string, subs *Subscription, trans *Transaction, mType int, payload []byte, payloadLen int) (err error) { +func (c *Control) rmrReplyToSender(desc string, subs *Subscription, trans *Transaction) (err error) { params := &RMRParams{&xapp.RMRParams{}} - params.Mtype = mType + params.Mtype = trans.GetMtype() params.SubId = int(subs.GetSubId()) params.Xid = trans.GetXid() params.Meid = trans.GetMeid() params.Src = "" - params.PayloadLen = payloadLen - params.Payload = payload + params.PayloadLen = len(trans.Payload.Buf) + params.Payload = trans.Payload.Buf params.Mbuf = nil return c.rmrSendRaw(desc, params) @@ -186,55 +174,86 @@ func (c *Control) Consume(params *xapp.RMRParams) (err error) { return nil } +func idstring(trans fmt.Stringer, subs fmt.Stringer, err error) string { + var retval string = "" + var filler string = "" + if trans != nil { + retval += filler + trans.String() + filler = " " + } + if subs != nil { + retval += filler + subs.String() + filler = " " + } + if err != nil { + retval += filler + "err(" + err.Error() + ")" + filler = " " + } + return retval +} + +func (c *Control) findSubs(ids []int) (*Subscription, error) { + var subs *Subscription = nil + for _, id := range ids { + if id >= 0 { + subs = c.registry.GetSubscription(uint16(id)) + } + if subs != nil { + break + } + } + if subs == nil { + return nil, fmt.Errorf("No valid subscription found with ids %v", ids) + } + return subs, nil +} + +func (c *Control) findSubsAndTrans(ids []int) (*Subscription, *Transaction, error) { + subs, err := c.findSubs(ids) + if err != nil { + return nil, nil, err + } + trans := subs.GetTransaction() + if trans == nil { + return subs, nil, fmt.Errorf("No ongoing transaction found from %s", idstring(nil, subs, nil)) + } + return subs, trans, nil +} func (c *Control) handleSubscriptionRequest(params *RMRParams) { xapp.Logger.Info("SubReq from xapp: %s", params.String()) - // - // - // - trans, err := c.tracker.TrackTransaction(NewRmrEndpoint(params.Src), - params.Mtype, - params.Xid, - params.Meid, - false, - true) - + SubReqMsg, err := c.e2ap.UnpackSubscriptionRequest(params.Payload) if err != nil { - xapp.Logger.Error("SubReq: %s, Dropping this msg. %s", err.Error(), params.String()) + xapp.Logger.Error("SubReq Drop: %s", idstring(params, nil, err)) return } - // - // - // - trans.SubReqMsg, err = c.e2ap.UnpackSubscriptionRequest(params.Payload) + trans, err := c.tracker.TrackTransaction(NewRmrEndpoint(params.Src), params.Xid, params.Meid, false, true) if err != nil { - xapp.Logger.Error("SubReq: %s Dropping this msg. %s", err.Error(), trans) - trans.Release() + xapp.Logger.Error("SubReq Drop: %s", idstring(params, nil, err)) return } + trans.SubReqMsg = SubReqMsg - // - // - // - subs, err := c.registry.ReserveSubscription(&trans.RmrEndpoint, trans.Meid) + subs, err := c.registry.ReserveSubscription(trans.Meid) if err != nil { - xapp.Logger.Error("SubReq: %s, Dropping this msg. %s", err.Error(), trans) + xapp.Logger.Error("SubReq Drop: %s", idstring(trans, nil, err)) trans.Release() return } err = subs.SetTransaction(trans) if err != nil { - xapp.Logger.Error("SubReq: %s, Dropping this msg. %s", err.Error(), trans) + xapp.Logger.Error("SubReq Drop: %s", idstring(trans, subs, err)) subs.Release() trans.Release() return } - trans.SubReqMsg.RequestId.Seq = uint32(subs.GetSubId()) + xapp.Logger.Debug("SubReq: Handling %s", idstring(trans, subs, nil)) + // // TODO: subscription create is in fact owned by subscription and not transaction. // Transaction is toward xapp while Subscription is toward ran. @@ -243,65 +262,36 @@ func (c *Control) handleSubscriptionRequest(params *RMRParams) { // // This is intermediate solution while improving message handling // - packedData, err := c.e2ap.PackSubscriptionRequest(trans.SubReqMsg) + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionRequest(trans.SubReqMsg) if err != nil { - xapp.Logger.Error("SubReq: %s for trans %s", err.Error(), trans) + xapp.Logger.Error("SubResp Drop: %s", idstring(trans, subs, err)) subs.Release() trans.Release() return } - //Optimize and store packed message to be sent (for retransmission). Again owned by subscription? - trans.Payload = packedData.Buf - trans.PayloadLen = len(packedData.Buf) - - c.rmrSend("SubReq to E2T", subs, trans, packedData.Buf, len(packedData.Buf)) - + c.rmrSend("SubReq: SubReq to E2T", subs, trans) c.timerMap.StartTimer("RIC_SUB_REQ", int(subs.GetSubId()), subReqTime, FirstTry, c.handleSubscriptionRequestTimer) - xapp.Logger.Debug("SubReq: Debugging trans table = %v", c.tracker.transactionXappTable) return } func (c *Control) handleSubscriptionResponse(params *RMRParams) { xapp.Logger.Info("SubResp from E2T: %s", params.String()) - // - // - // SubRespMsg, err := c.e2ap.UnpackSubscriptionResponse(params.Payload) if err != nil { - xapp.Logger.Error("SubDelReq: %s Dropping this msg. %s", err.Error(), params.String()) + xapp.Logger.Error("SubResp Drop %s", idstring(params, nil, err)) return } - // - // - // - subs := c.registry.GetSubscription(uint16(SubRespMsg.RequestId.Seq)) - if subs == nil && params.SubId > 0 { - subs = c.registry.GetSubscription(uint16(params.SubId)) - } - - if subs == nil { - xapp.Logger.Error("SubResp: Not valid subscription found payloadSeqNum: %d, SubId: %d. Dropping this msg. %s", SubRespMsg.RequestId.Seq, params.SubId, params.String()) - return - } - xapp.Logger.Info("SubResp: subscription found payloadSeqNum: %d, SubId: %d", SubRespMsg.RequestId.Seq, subs.GetSubId()) - - // - // - // - trans := subs.GetTransaction() - if trans == nil { - xapp.Logger.Error("SubResp: Unknown trans. Dropping this msg. SubId: %d", subs.GetSubId()) + subs, trans, err := c.findSubsAndTrans([]int{int(SubRespMsg.RequestId.Seq), params.SubId}) + if err != nil { + xapp.Logger.Error("SubResp: %s", idstring(params, nil, err)) return } - trans.SubRespMsg = SubRespMsg + xapp.Logger.Debug("SubResp: Handling %s", idstring(trans, subs, nil)) - // - // - // c.timerMap.StopTimer("RIC_SUB_REQ", int(subs.GetSubId())) responseReceived := trans.CheckResponseReceived() @@ -310,84 +300,50 @@ func (c *Control) handleSubscriptionResponse(params *RMRParams) { return } - packedData, err := c.e2ap.PackSubscriptionResponse(trans.SubRespMsg) + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionResponse(trans.SubRespMsg) if err != nil { - xapp.Logger.Error("SubResp: %s for trans %s", err.Error(), trans) + xapp.Logger.Error("SubResp: %s", idstring(trans, subs, err)) trans.Release() return } - //Optimize and store packed message to be sent. - trans.Payload = packedData.Buf - trans.PayloadLen = len(packedData.Buf) - subs.Confirmed() trans.Release() - c.rmrReplyToSender("SubResp to xapp", subs, trans, 12011, trans.Payload, trans.PayloadLen) + c.rmrReplyToSender("SubResp: SubResp to xapp", subs, trans) return } func (c *Control) handleSubscriptionFailure(params *RMRParams) { xapp.Logger.Info("SubFail from E2T: %s", params.String()) - // - // - // SubFailMsg, err := c.e2ap.UnpackSubscriptionFailure(params.Payload) if err != nil { - xapp.Logger.Error("SubFail: %s Dropping this msg. %s", err.Error(), params.String()) - return - } - - // - // - // - subs := c.registry.GetSubscription(uint16(SubFailMsg.RequestId.Seq)) - if subs == nil && params.SubId > 0 { - subs = c.registry.GetSubscription(uint16(params.SubId)) - } - - if subs == nil { - xapp.Logger.Error("SubFail: Not valid subscription found payloadSeqNum: %d, SubId: %d. Dropping this msg. %s", SubFailMsg.RequestId.Seq, params.SubId, params.String()) + xapp.Logger.Error("SubFail Drop %s", idstring(params, nil, err)) return } - xapp.Logger.Info("SubFail: subscription found payloadSeqNum: %d, SubId: %d", SubFailMsg.RequestId.Seq, subs.GetSubId()) - // - // - // - trans := subs.GetTransaction() - if trans == nil { - xapp.Logger.Error("SubFail: Unknown trans. Dropping this msg. SubId: %d", subs.GetSubId()) + subs, trans, err := c.findSubsAndTrans([]int{int(SubFailMsg.RequestId.Seq), params.SubId}) + if err != nil { + xapp.Logger.Error("SubFail: %s", idstring(params, nil, err)) return } trans.SubFailMsg = SubFailMsg + xapp.Logger.Debug("SubFail: Handling %s", idstring(trans, subs, nil)) - // - // - // c.timerMap.StopTimer("RIC_SUB_REQ", int(subs.GetSubId())) - responseReceived := trans.CheckResponseReceived() - if err != nil { - return - } - if responseReceived == true { // Subscription timer already received return } - packedData, err := c.e2ap.PackSubscriptionFailure(trans.SubFailMsg) - if err != nil { - //TODO error handling improvement - xapp.Logger.Error("SubFail: %s for trans %s (continue still)", err.Error(), trans) - } else { - //Optimize and store packed message to be sent. - trans.Payload = packedData.Buf - trans.PayloadLen = len(packedData.Buf) - c.rmrReplyToSender("SubFail to xapp", subs, trans, 12012, trans.Payload, trans.PayloadLen) + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionFailure(trans.SubFailMsg) + if err == nil { + c.rmrReplyToSender("SubFail: SubFail to xapp", subs, trans) time.Sleep(3 * time.Second) + } else { + //TODO error handling improvement + xapp.Logger.Error("SubFail: (continue cleaning) %s", idstring(trans, subs, err)) } trans.Release() @@ -398,50 +354,38 @@ func (c *Control) handleSubscriptionFailure(params *RMRParams) { func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCount uint64) { xapp.Logger.Info("SubReq timeout: subId: %v, tryCount: %v", nbrId, tryCount) - subs := c.registry.GetSubscription(uint16(nbrId)) - if subs == nil { - xapp.Logger.Error("SubReq timeout: Unknown payloadSeqNum. Dropping this msg. SubId: %v", nbrId) - return - } - - trans := subs.GetTransaction() - if trans == nil { - xapp.Logger.Error("SubReq timeout: Unknown trans. Dropping this msg. SubId: %v", subs.GetSubId()) + subs, trans, err := c.findSubsAndTrans(([]int{nbrId})) + if err != nil { + xapp.Logger.Error("SubReq timeout: %s", idstring(nil, nil, err)) return } + xapp.Logger.Debug("SubReq timeout: Handling %s", idstring(trans, subs, nil)) responseReceived := trans.CheckResponseReceived() - if responseReceived == true { // Subscription Response or Failure already received return } if tryCount < maxSubReqTryCount { - xapp.Logger.Info("SubReq timeout: Resending SubReq to E2T: Mtype: %v, SubId: %v, Xid %s, Meid %v", trans.GetMtype(), subs.GetSubId(), trans.GetXid(), trans.GetMeid()) + xapp.Logger.Info("SubReq timeout: %s", idstring(trans, subs, nil)) trans.RetryTransaction() - c.rmrSend("SubReq(SubReq timer retransmit) to E2T", subs, trans, trans.Payload, trans.PayloadLen) + c.rmrSend("SubReq timeout: SubReq to E2T", subs, trans) tryCount++ c.timerMap.StartTimer("RIC_SUB_REQ", int(subs.GetSubId()), subReqTime, tryCount, c.handleSubscriptionRequestTimer) return } - // Delete CREATE transaction + // Release CREATE transaction trans.Release() // Create DELETE transaction (internal and no messages toward xapp) - deltrans, err := c.tracker.TrackTransaction(&trans.RmrEndpoint, - 12020, // RIC SUBSCRIPTION DELETE - trans.GetXid(), - trans.GetMeid(), - false, - false) - + deltrans, err := c.tracker.TrackTransaction(&trans.RmrEndpoint, trans.GetXid(), trans.GetMeid(), false, false) if err != nil { - xapp.Logger.Error("SubReq timeout: %s, Dropping this msg.", err.Error()) + xapp.Logger.Error("SubReq timeout: %s", idstring(trans, subs, err)) //TODO improve error handling. Important at least in merge subs.Release() return @@ -451,27 +395,24 @@ func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCou deltrans.SubDelReqMsg.RequestId.Id = trans.SubReqMsg.RequestId.Id deltrans.SubDelReqMsg.RequestId.Seq = uint32(subs.GetSubId()) deltrans.SubDelReqMsg.FunctionId = trans.SubReqMsg.FunctionId - packedData, err := c.e2ap.PackSubscriptionDeleteRequest(deltrans.SubDelReqMsg) + deltrans.Mtype, deltrans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(deltrans.SubDelReqMsg) if err != nil { - xapp.Logger.Error("SubReq timeout: Packing SubDelReq failed. Err: %v", err) + xapp.Logger.Error("SubReq timeout: %s", idstring(trans, subs, err)) //TODO improve error handling. Important at least in merge deltrans.Release() subs.Release() return } - deltrans.PayloadLen = len(packedData.Buf) - deltrans.Payload = packedData.Buf err = subs.SetTransaction(deltrans) if err != nil { - xapp.Logger.Error("SubReq timeout: %s, Dropping this msg.", err.Error()) + xapp.Logger.Error("SubReq timeout: %s", idstring(trans, subs, err)) //TODO improve error handling. Important at least in merge deltrans.Release() return } - c.rmrSend("SubDelReq(SubReq timer) to E2T", subs, deltrans, deltrans.Payload, deltrans.PayloadLen) - + c.rmrSend("SubReq timer: SubDelReq to E2T", subs, deltrans) c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId()), subDelReqTime, FirstTry, c.handleSubscriptionDeleteRequestTimer) return } @@ -479,53 +420,35 @@ func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCou func (c *Control) handleSubscriptionDeleteRequest(params *RMRParams) { xapp.Logger.Info("SubDelReq from xapp: %s", params.String()) - // - // - // - trans, err := c.tracker.TrackTransaction(NewRmrEndpoint(params.Src), - params.Mtype, - params.Xid, - params.Meid, - false, - true) - + SubDelReqMsg, err := c.e2ap.UnpackSubscriptionDeleteRequest(params.Payload) if err != nil { - xapp.Logger.Error("SubDelReq: %s, Dropping this msg. %s", err.Error(), params.String()) + xapp.Logger.Error("SubDelReq Drop %s", idstring(params, nil, err)) return } - // - // - // - trans.SubDelReqMsg, err = c.e2ap.UnpackSubscriptionDeleteRequest(params.Payload) + trans, err := c.tracker.TrackTransaction(NewRmrEndpoint(params.Src), params.Xid, params.Meid, false, true) if err != nil { - xapp.Logger.Error("SubDelReq: %s Dropping this msg. %s", err.Error(), trans) - trans.Release() + xapp.Logger.Error("SubDelReq Drop %s", idstring(params, nil, err)) return } + trans.SubDelReqMsg = SubDelReqMsg - // - // - // - subs := c.registry.GetSubscription(uint16(trans.SubDelReqMsg.RequestId.Seq)) - if subs == nil && params.SubId > 0 { - subs = c.registry.GetSubscription(uint16(params.SubId)) - } - - if subs == nil { - xapp.Logger.Error("SubDelReq: Not valid subscription found payloadSeqNum: %d, SubId: %d. Dropping this msg. %s", trans.SubDelReqMsg.RequestId.Seq, params.SubId, trans) + subs, err := c.findSubs([]int{int(trans.SubDelReqMsg.RequestId.Seq), params.SubId}) + if err != nil { + xapp.Logger.Error("SubDelReq: %s", idstring(params, nil, err)) trans.Release() return } - xapp.Logger.Info("SubDelReq: subscription found payloadSeqNum: %d, SubId: %d. %s", trans.SubDelReqMsg.RequestId.Seq, params.SubId, trans) err = subs.SetTransaction(trans) if err != nil { - xapp.Logger.Error("SubDelReq: %s, Dropping this msg. %s", err.Error(), trans) + xapp.Logger.Error("SubDelReq: %s", idstring(trans, subs, err)) trans.Release() return } + xapp.Logger.Debug("SubDelReq: Handling %s", idstring(trans, subs, nil)) + // // TODO: subscription delete is in fact owned by subscription and not transaction. // Transaction is toward xapp while Subscription is toward ran. @@ -534,20 +457,16 @@ func (c *Control) handleSubscriptionDeleteRequest(params *RMRParams) { // // This is intermediate solution while improving message handling // - packedData, err := c.e2ap.PackSubscriptionDeleteRequest(trans.SubDelReqMsg) + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(trans.SubDelReqMsg) if err != nil { - xapp.Logger.Error("SubDelReq: %s for trans %s", err.Error(), trans) + xapp.Logger.Error("SubDelReq: %s", idstring(trans, subs, err)) trans.Release() return } - //Optimize and store packed message to be sent (for retransmission). Again owned by subscription? - trans.Payload = packedData.Buf - trans.PayloadLen = len(packedData.Buf) - subs.UnConfirmed() - c.rmrSend("SubDelReq to E2T", subs, trans, trans.Payload, trans.PayloadLen) + c.rmrSend("SubDelReq: SubDelReq to E2T", subs, trans) c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId()), subDelReqTime, FirstTry, c.handleSubscriptionDeleteRequestTimer) return @@ -556,24 +475,19 @@ func (c *Control) handleSubscriptionDeleteRequest(params *RMRParams) { func (c *Control) handleSubscriptionDeleteResponse(params *RMRParams) (err error) { xapp.Logger.Info("SubDelResp from E2T:%s", params.String()) - payloadSeqNum, err := c.e2ap.GetSubscriptionDeleteResponseSequenceNumber(params.Payload) + SubDelRespMsg, err := c.e2ap.UnpackSubscriptionDeleteResponse(params.Payload) if err != nil { - xapp.Logger.Error("SubDelResp: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, SubId: %v, Xid: %s, Payload %X", err, params.SubId, params.Xid, params.Payload) - return - } - xapp.Logger.Info("SubDelResp: Received payloadSeqNum: %v", payloadSeqNum) - - subs := c.registry.GetSubscription(payloadSeqNum) - if subs == nil { - xapp.Logger.Error("SubDelResp: Unknown payloadSeqNum. Dropping this msg. PayloadSeqNum: %v, SubId: %v", payloadSeqNum, params.SubId) + xapp.Logger.Error("SubDelResp: Dropping this msg. %s", idstring(params, nil, err)) return } - trans := subs.GetTransaction() - if trans == nil { - xapp.Logger.Error("SubDelResp: Unknown trans. Dropping this msg. PayloadSeqNum: %v, SubId: %v", subs.GetSubId(), params.SubId) + subs, trans, err := c.findSubsAndTrans([]int{int(SubDelRespMsg.RequestId.Seq), params.SubId}) + if err != nil { + xapp.Logger.Error("SubDelResp: %s", idstring(params, nil, err)) return } + trans.SubDelRespMsg = SubDelRespMsg + xapp.Logger.Debug("SubDelResp: Handling %s", idstring(trans, subs, nil)) c.timerMap.StopTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId())) @@ -583,37 +497,26 @@ func (c *Control) handleSubscriptionDeleteResponse(params *RMRParams) (err error return } - if trans.ForwardRespToXapp == true { - c.rmrReplyToSender("SubDelResp to xapp", subs, trans, params.Mtype, params.Payload, params.PayloadLen) - time.Sleep(3 * time.Second) - } - - trans.Release() - subs.Release() + c.sendSubscriptionDeleteResponse("SubDelResp", trans, subs) return } func (c *Control) handleSubscriptionDeleteFailure(params *RMRParams) { xapp.Logger.Info("SubDelFail from E2T:%s", params.String()) - payloadSeqNum, err := c.e2ap.GetSubscriptionDeleteFailureSequenceNumber(params.Payload) + SubDelFailMsg, err := c.e2ap.UnpackSubscriptionDeleteFailure(params.Payload) if err != nil { - xapp.Logger.Error("SubDelFail: Unable to get Sequence Number from Payload. Dropping this msg. Err: %v, %s", err, params.String()) - return - } - xapp.Logger.Info("SubDelFail: Received payloadSeqNum: %v", payloadSeqNum) - - subs := c.registry.GetSubscription(payloadSeqNum) - if subs == nil { - xapp.Logger.Error("SubDelFail: Unknown payloadSeqNum. Dropping this msg. PayloadSeqNum: %v, SubId: %v", payloadSeqNum, params.SubId) + xapp.Logger.Error("SubDelFail: Dropping this msg. %s", idstring(params, nil, err)) return } - trans := subs.GetTransaction() - if trans == nil { - xapp.Logger.Error("SubDelFail: Unknown trans. Dropping this msg. PayloadSeqNum: %v, SubId: %v", subs.GetSubId(), params.SubId) + subs, trans, err := c.findSubsAndTrans([]int{int(SubDelFailMsg.RequestId.Seq), params.SubId}) + if err != nil { + xapp.Logger.Error("SubDelFail: %s", idstring(params, nil, err)) return } + trans.SubDelFailMsg = SubDelFailMsg + xapp.Logger.Debug("SubDelFail: Handling %s", idstring(trans, subs, nil)) c.timerMap.StopTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId())) @@ -622,38 +525,20 @@ func (c *Control) handleSubscriptionDeleteFailure(params *RMRParams) { // Subscription Delete timer already received return } - if trans.ForwardRespToXapp == true { - var subDelRespPayload []byte - subDelRespPayload, err = c.e2ap.PackSubscriptionDeleteResponseFromSubDelReq(trans.Payload, subs.GetSubId()) - if err != nil { - xapp.Logger.Error("SubDelFail:Packing SubDelResp failed. Err: %v", err) - return - } - - // RIC SUBSCRIPTION DELETE RESPONSE - c.rmrReplyToSender("SubDelFail to xapp", subs, trans, 12021, subDelRespPayload, len(subDelRespPayload)) - time.Sleep(3 * time.Second) - } - trans.Release() - subs.Release() + c.sendSubscriptionDeleteResponse("SubDelFail", trans, subs) return } func (c *Control) handleSubscriptionDeleteRequestTimer(strId string, nbrId int, tryCount uint64) { xapp.Logger.Info("SubDelReq timeout: subId: %v, tryCount: %v", nbrId, tryCount) - subs := c.registry.GetSubscription(uint16(nbrId)) - if subs == nil { - xapp.Logger.Error("SubDelReq timeout: Unknown payloadSeqNum. Dropping this msg. SubId: %v", nbrId) - return - } - - trans := subs.GetTransaction() - if trans == nil { - xapp.Logger.Error("SubDelReq timeout: Unknown trans. Dropping this msg. SubId: %v", subs.GetSubId()) + subs, trans, err := c.findSubsAndTrans([]int{nbrId}) + if err != nil { + xapp.Logger.Error("SubDelReq timeout: %s", idstring(nil, nil, err)) return } + xapp.Logger.Debug("SubDelReq timeout: Handling %s", idstring(trans, subs, nil)) responseReceived := trans.CheckResponseReceived() if responseReceived == true { @@ -664,27 +549,36 @@ func (c *Control) handleSubscriptionDeleteRequestTimer(strId string, nbrId int, if tryCount < maxSubDelReqTryCount { // Set possible to handle new response for the subId trans.RetryTransaction() - c.rmrSend("SubDelReq(SubDelReq timer retransmit) to E2T", subs, trans, trans.Payload, trans.PayloadLen) + c.rmrSend("SubDelReq timeout: SubDelReq to E2T", subs, trans) tryCount++ c.timerMap.StartTimer("RIC_SUB_DEL_REQ", int(subs.GetSubId()), subReqTime, tryCount, c.handleSubscriptionDeleteRequestTimer) return } - if trans.ForwardRespToXapp == true { - var subDelRespPayload []byte - subDelRespPayload, err := c.e2ap.PackSubscriptionDeleteResponseFromSubDelReq(trans.Payload, subs.GetSubId()) - if err != nil { - xapp.Logger.Error("SubDelReq timeout: Unable to pack payload. Dropping this this msg. Err: %v, SubId: %v, Xid: %s, Payload %x", err, subs.GetSubId(), trans.GetXid(), trans.Payload) - return - } - - // RIC SUBSCRIPTION DELETE RESPONSE - c.rmrReplyToSender("SubDelResp(SubDelReq timer) to xapp", subs, trans, 12021, subDelRespPayload, len(subDelRespPayload)) + c.sendSubscriptionDeleteResponse("SubDelReq(timer)", trans, subs) + return +} - time.Sleep(3 * time.Second) +func (c *Control) sendSubscriptionDeleteResponse(desc string, trans *Transaction, subs *Subscription) { + if trans.ForwardRespToXapp == true { + //Always generate SubDelResp + trans.SubDelRespMsg = &e2ap.E2APSubscriptionDeleteResponse{} + trans.SubDelRespMsg.RequestId.Id = trans.SubDelReqMsg.RequestId.Id + trans.SubDelRespMsg.RequestId.Seq = uint32(subs.GetSubId()) + trans.SubDelRespMsg.FunctionId = trans.SubDelReqMsg.FunctionId + + var err error + trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteResponse(trans.SubDelRespMsg) + if err == nil { + c.rmrReplyToSender(desc+": SubDelResp to xapp", subs, trans) + time.Sleep(3 * time.Second) + } else { + //TODO error handling improvement + xapp.Logger.Error("%s: (continue cleaning) %s", desc, idstring(trans, subs, err)) + } } + trans.Release() subs.Release() - return }