X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fcontrol.go;h=1bd27ebef23a00cde26a312bd8143e0c0c793105;hb=refs%2Fchanges%2F20%2F2220%2F1;hp=6fc7d33382a7d96b21a9dc6c9f1d834a69cd3c32;hpb=b3dc15673967c1c94a52d82ea8b169fefd6f9154;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/control.go b/pkg/control/control.go index 6fc7d33..1bd27eb 100755 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -263,7 +263,7 @@ func (c *Control) handleSubscriptionRequest(params *RMRParams) { c.rmrSend("SubReq to E2T", subs, trans, packedData.Buf, len(packedData.Buf)) - c.timerMap.StartTimer("RIC_SUB_REQ", int(subs.Seq), subReqTime, FirstTry, c.handleSubscriptionRequestTimer) + 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 } @@ -271,22 +271,44 @@ func (c *Control) handleSubscriptionRequest(params *RMRParams) { func (c *Control) handleSubscriptionResponse(params *RMRParams) { xapp.Logger.Info("SubResp from E2T: %s", params.String()) - payloadSeqNum, err := c.e2ap.GetSubscriptionResponseSequenceNumber(params.Payload) + // + // + // + SubRespMsg, err := c.e2ap.UnpackSubscriptionResponse(params.Payload) if err != nil { - xapp.Logger.Error("SubResp: 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) + xapp.Logger.Error("SubDelReq: %s Dropping this msg. %s", err.Error(), params.String()) return } - xapp.Logger.Info("SubResp: Received payloadSeqNum: %v", payloadSeqNum) - subs := c.registry.GetSubscription(payloadSeqNum) + // + // + // + 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: Unknown payloadSeqNum. Dropping this msg. PayloadSeqNum: %v, SubId: %v", payloadSeqNum, params.SubId) + 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()) + return + } - c.timerMap.StopTimer("RIC_SUB_REQ", int(payloadSeqNum)) + trans.SubRespMsg = SubRespMsg + + // + // + // + c.timerMap.StopTimer("RIC_SUB_REQ", int(subs.GetSubId())) responseReceived := trans.CheckResponseReceived() if responseReceived == true { @@ -294,40 +316,66 @@ func (c *Control) handleSubscriptionResponse(params *RMRParams) { return } + packedData, err := c.e2ap.PackSubscriptionResponse(trans.SubRespMsg) + if err != nil { + xapp.Logger.Error("SubResp: %s for trans %s", err.Error(), trans) + 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, params.Mtype, params.Payload, params.PayloadLen) - xapp.Logger.Info("SubResp: SubId: %v, from address: %s. Deleting trans record", payloadSeqNum, trans.RmrEndpoint) + c.rmrReplyToSender("SubResp to xapp", subs, trans, 12011, trans.Payload, trans.PayloadLen) return } func (c *Control) handleSubscriptionFailure(params *RMRParams) { xapp.Logger.Info("SubFail from E2T: %s", params.String()) - payloadSeqNum, err := c.e2ap.GetSubscriptionFailureSequenceNumber(params.Payload) + // + // + // + SubFailMsg, err := c.e2ap.UnpackSubscriptionFailure(params.Payload) if err != nil { - xapp.Logger.Error("SubFail: 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) + xapp.Logger.Error("SubFail: %s Dropping this msg. %s", err.Error(), params.String()) return } - xapp.Logger.Info("SubFail: Received payloadSeqNum: %v", payloadSeqNum) - subs := c.registry.GetSubscription(payloadSeqNum) + // + // + // + 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: Unknown payloadSeqNum. Dropping this msg. PayloadSeqNum: %v, SubId: %v", payloadSeqNum, params.SubId) + xapp.Logger.Error("SubFail: Not valid subscription found payloadSeqNum: %d, SubId: %d. Dropping this msg. %s", SubFailMsg.RequestId.Seq, params.SubId, params.String()) 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. PayloadSeqNum: %v, SubId: %v", payloadSeqNum, params.SubId) + xapp.Logger.Error("SubFail: Unknown trans. Dropping this msg. SubId: %d", subs.GetSubId()) return } + trans.SubFailMsg = SubFailMsg - c.timerMap.StopTimer("RIC_SUB_REQ", int(payloadSeqNum)) + // + // + // + c.timerMap.StopTimer("RIC_SUB_REQ", int(subs.GetSubId())) responseReceived := trans.CheckResponseReceived() if err != nil { - xapp.Logger.Info("SubFail: Dropping this msg. Err: %v SubId: %v", err, payloadSeqNum) return } @@ -335,16 +383,22 @@ func (c *Control) handleSubscriptionFailure(params *RMRParams) { // Subscription timer already received return } - xapp.Logger.Info("SubFail: SubId: %v, from address: %s. Forwarding response to xApp", payloadSeqNum, trans.RmrEndpoint) - c.rmrReplyToSender("SubFail to xapp", subs, trans, params.Mtype, params.Payload, params.PayloadLen) - - time.Sleep(3 * time.Second) + 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) + time.Sleep(3 * time.Second) + } - xapp.Logger.Info("SubFail: Deleting trans record. SubId: %v, Xid: %s", params.SubId, params.Xid) trans.Release() - if !c.registry.DelSubscription(payloadSeqNum) { - xapp.Logger.Error("SubFail: Failed to release sequency number. SubId: %v, Xid: %s", params.SubId, params.Xid) + if !c.registry.DelSubscription(subs.GetSubId()) { + xapp.Logger.Error("SubFail: Failed to release sequency number. %s", subs) } return } @@ -431,10 +485,11 @@ func (c *Control) handleSubscriptionRequestTimer(strId string, nbrId int, tryCou } func (c *Control) handleSubscriptionDeleteRequest(params *RMRParams) { - var subs *Subscription - xapp.Logger.Info("SubDelReq from xapp: %s", params.String()) + // + // + // trans, err := c.tracker.TrackTransaction(NewRmrEndpoint(params.Src), params.Mtype, params.Xid, @@ -457,7 +512,10 @@ func (c *Control) handleSubscriptionDeleteRequest(params *RMRParams) { return } - subs = c.registry.GetSubscription(uint16(trans.SubDelReqMsg.RequestId.Seq)) + // + // + // + subs := c.registry.GetSubscription(uint16(trans.SubDelReqMsg.RequestId.Seq)) if subs == nil && params.SubId > 0 { subs = c.registry.GetSubscription(uint16(params.SubId)) }