X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fcontrol%2Fcontrol.go;h=977949880448dbb7213ae8d7e86a3b226d64ea20;hb=987f8b1fdb2dd8413d6c8faef1eb8618226e0421;hp=182e06f1d1349fe54fe6c7595c30b224b6aa4e88;hpb=6b3796f1b3cbdc0ec389d50dd98fbf6ef2aae720;p=ric-plt%2Fsubmgr.git diff --git a/pkg/control/control.go b/pkg/control/control.go index 182e06f..9779498 100755 --- a/pkg/control/control.go +++ b/pkg/control/control.go @@ -34,6 +34,7 @@ import ( httptransport "github.com/go-openapi/runtime/client" "github.com/go-openapi/strfmt" "github.com/gorilla/mux" + "github.com/segmentio/ksuid" "github.com/spf13/viper" ) @@ -51,7 +52,6 @@ func idstring(err error, entries ...fmt.Stringer) string { if err != nil { retval += filler + "err(" + err.Error() + ")" filler = " " - } return retval } @@ -63,20 +63,22 @@ func idstring(err error, entries ...fmt.Stringer) string { var e2tSubReqTimeout time.Duration var e2tSubDelReqTime time.Duration var e2tRecvMsgTimeout time.Duration +var waitRouteCleanup_ms time.Duration var e2tMaxSubReqTryCount uint64 // Initial try + retry var e2tMaxSubDelReqTryCount uint64 // Initial try + retry var readSubsFromDb string +var restDuplicateCtrl duplicateCtrl type Control struct { *xapp.RMRClient - e2ap *E2ap - registry *Registry - tracker *Tracker - db Sdlnterface - //subscriber *xapp.Subscriber + e2ap *E2ap + registry *Registry + tracker *Tracker + db Sdlnterface CntRecvMsg uint64 ResetTestFlag bool Counters map[string]xapp.Counter + LoggerLevel uint32 } type RMRMeid struct { @@ -97,7 +99,6 @@ func init() { func NewControl() *Control { - ReadConfigParameters() transport := httptransport.New(viper.GetString("rtmgr.HostAddr")+":"+viper.GetString("rtmgr.port"), viper.GetString("rtmgr.baseUrl"), []string{"http"}) rtmgrClient := RtmgrClient{rtClient: rtmgrclient.New(transport, strfmt.Default)} @@ -108,26 +109,27 @@ func NewControl() *Control { tracker := new(Tracker) tracker.Init() - //subscriber := xapp.NewSubscriber(viper.GetString("subscription.host"), viper.GetInt("subscription.timeout")) - c := &Control{e2ap: new(E2ap), - registry: registry, - tracker: tracker, - db: CreateSdl(), - //subscriber: subscriber, - Counters: xapp.Metric.RegisterCounterGroup(GetMetricsOpts(), "SUBMGR"), + registry: registry, + tracker: tracker, + db: CreateSdl(), + Counters: xapp.Metric.RegisterCounterGroup(GetMetricsOpts(), "SUBMGR"), + LoggerLevel: 3, } + c.ReadConfigParameters("") // Register REST handler for testing support xapp.Resource.InjectRoute("/ric/v1/test/{testId}", c.TestRestHandler, "POST") + xapp.Resource.InjectRoute("/ric/v1/symptomdata", c.SymptomDataHandler, "GET") - go xapp.Subscription.Listen(c.SubscriptionHandler, c.QueryHandler, c.SubscriptionDeleteHandler) - //go c.subscriber.Listen(c.SubscriptionHandler, c.QueryHandler) + go xapp.Subscription.Listen(c.SubscriptionHandler, c.QueryHandler, c.SubscriptionDeleteHandlerCB) if readSubsFromDb == "false" { return c } + restDuplicateCtrl.Init() + // Read subscriptions from db xapp.Logger.Info("Reading subscriptions from db") subIds, register, err := c.ReadAllSubscriptionsFromSdl() @@ -141,10 +143,15 @@ func NewControl() *Control { return c } +func (c *Control) SymptomDataHandler(w http.ResponseWriter, r *http.Request) { + subscriptions, _ := c.registry.QueryHandler() + xapp.Resource.SendSymptomDataJson(w, r, subscriptions, "platform/subscriptions.json") +} + //------------------------------------------------------------------- // //------------------------------------------------------------------- -func ReadConfigParameters() { +func (c *Control) ReadConfigParameters(f string) { // viper.GetDuration returns nanoseconds e2tSubReqTimeout = viper.GetDuration("controls.e2tSubReqTimeout_ms") * 1000000 @@ -162,11 +169,21 @@ func ReadConfigParameters() { e2tRecvMsgTimeout = 2000 * 1000000 } xapp.Logger.Info("e2tRecvMsgTimeout %v", e2tRecvMsgTimeout) + + // Internal cfg parameter, used to define a wait time for RMR route clean-up. None default + // value 100ms used currently only in unittests. + waitRouteCleanup_ms = viper.GetDuration("controls.waitRouteCleanup_ms") * 1000000 + if waitRouteCleanup_ms == 0 { + waitRouteCleanup_ms = 5000 * 1000000 + } + xapp.Logger.Info("waitRouteCleanup %v", waitRouteCleanup_ms) + e2tMaxSubReqTryCount = viper.GetUint64("controls.e2tMaxSubReqTryCount") if e2tMaxSubReqTryCount == 0 { e2tMaxSubReqTryCount = 1 } xapp.Logger.Info("e2tMaxSubReqTryCount %v", e2tMaxSubReqTryCount) + e2tMaxSubDelReqTryCount = viper.GetUint64("controls.e2tMaxSubDelReqTryCount") if e2tMaxSubDelReqTryCount == 0 { e2tMaxSubDelReqTryCount = 1 @@ -178,6 +195,10 @@ func ReadConfigParameters() { readSubsFromDb = "true" } xapp.Logger.Info("readSubsFromDb %v", readSubsFromDb) + c.LoggerLevel = viper.GetUint32("logger.level") + if c.LoggerLevel == 0 { + c.LoggerLevel = 3 + } } //------------------------------------------------------------------- @@ -203,36 +224,328 @@ func (c *Control) ReadyCB(data interface{}) { func (c *Control) Run() { xapp.SetReadyCB(c.ReadyCB, nil) + xapp.AddConfigChangeListener(c.ReadConfigParameters) xapp.Run(c) } //------------------------------------------------------------------- // //------------------------------------------------------------------- -func (c *Control) SubscriptionHandler(stype models.SubscriptionType, params interface{}) (*models.SubscriptionResponse, error) { - /* - switch p := params.(type) { - case *models.ReportParams: - trans := c.tracker.NewXappTransaction(NewRmrEndpoint(p.ClientEndpoint),"" , 0, &xapp.RMRMeid{RanName: p.Meid}) - if trans == nil { - xapp.Logger.Error("XAPP-SubReq: %s", idstring(fmt.Errorf("transaction not created"), params)) - return - } - defer trans.Release() - case *models.ControlParams: - case *models.PolicyParams: - } - */ - return &models.SubscriptionResponse{}, fmt.Errorf("Subscription rest interface not implemented") -} - -func (c *Control) SubscriptionDeleteHandler(s string) error { +func (c *Control) SubscriptionHandler(params interface{}) (*models.SubscriptionResponse, error) { + + c.CntRecvMsg++ + c.UpdateCounter(cRestSubReqFromXapp) + + subResp := models.SubscriptionResponse{} + p := params.(*models.SubscriptionParams) + + if c.LoggerLevel > 2 { + c.PrintRESTSubscriptionRequest(p) + } + + if p.ClientEndpoint == nil { + xapp.Logger.Error("ClientEndpoint == nil") + c.UpdateCounter(cRestSubFailToXapp) + return nil, fmt.Errorf("") + } + + _, xAppRmrEndpoint, err := ConstructEndpointAddresses(*p.ClientEndpoint) + if err != nil { + xapp.Logger.Error("%s", err.Error()) + c.UpdateCounter(cRestSubFailToXapp) + return nil, err + } + var restSubId string + var restSubscription *RESTSubscription + if p.SubscriptionID == "" { + restSubId = ksuid.New().String() + restSubscription, err = c.registry.CreateRESTSubscription(&restSubId, &xAppRmrEndpoint, p.Meid) + if err != nil { + xapp.Logger.Error("%s", err.Error()) + c.UpdateCounter(cRestSubFailToXapp) + return nil, err + } + + } else { + restSubId = p.SubscriptionID + restSubscription, err = c.registry.GetRESTSubscription(restSubId, false) + if err != nil { + xapp.Logger.Error("%s", err.Error()) + c.UpdateCounter(cRestSubFailToXapp) + return nil, err + } + } + + subResp.SubscriptionID = &restSubId + subReqList := e2ap.SubscriptionRequestList{} + err = c.e2ap.FillSubscriptionReqMsgs(params, &subReqList, restSubscription) + if err != nil { + xapp.Logger.Error("%s", err.Error()) + c.registry.DeleteRESTSubscription(&restSubId) + c.UpdateCounter(cRestSubFailToXapp) + return nil, err + } + + err, duplicate, md5sum := restDuplicateCtrl.IsDuplicateToOngoingTransaction(restSubId, params) + + if err != nil { + // We were unable to detect whether this request was duplicate or not, proceed + xapp.Logger.Info("%s - proceeding with the request", err.Error()) + } else { + if duplicate { + if *p.SubscriptionDetails[0].ActionToBeSetupList[0].ActionType == "report" { + xapp.Logger.Info("Retransmission blocker dropped for report typer of request") + c.UpdateCounter(cRestSubRespToXapp) + return &subResp, nil + } + } + restSubscription.Md5sumOngoing = md5sum + } + + go c.processSubscriptionRequests(restSubscription, &subReqList, p.ClientEndpoint, p.Meid, &restSubId, xAppRmrEndpoint) + + c.UpdateCounter(cRestSubRespToXapp) + return &subResp, nil +} + +func (c *Control) sendUnsuccesfullResponseNotification(restSubId *string, restSubscription *RESTSubscription, xAppEventInstanceID int64, err error, + clientEndpoint *models.SubscriptionParamsClientEndpoint, trans *TransactionXapp) { + + // Send notification to xApp that prosessing of a Subscription Request has failed. + e2EventInstanceID := (int64)(0) + errorCause := err.Error() + resp := &models.SubscriptionResponse{ + SubscriptionID: restSubId, + SubscriptionInstances: []*models.SubscriptionInstance{ + &models.SubscriptionInstance{E2EventInstanceID: &e2EventInstanceID, + ErrorCause: &errorCause, + XappEventInstanceID: &xAppEventInstanceID}, + }, + } + // Mark REST subscription request processed. + restSubscription.SetProcessed() + if trans != nil { + xapp.Logger.Info("Sending unsuccessful REST notification (cause %s) to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s", + errorCause, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans)) + } else { + xapp.Logger.Info("Sending unsuccessful REST notification (cause %s) to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v", + errorCause, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID) + } + + c.UpdateCounter(cRestSubFailNotifToXapp) + xapp.Subscription.Notify(resp, *clientEndpoint) +} + +func (c *Control) sendSuccesfullResponseNotification(restSubId *string, restSubscription *RESTSubscription, xAppEventInstanceID int64, e2EventInstanceID int64, + clientEndpoint *models.SubscriptionParamsClientEndpoint, trans *TransactionXapp) { + + // Store successfully processed InstanceId for deletion + restSubscription.AddE2InstanceId((uint32)(e2EventInstanceID)) + restSubscription.AddXappIdToE2Id(xAppEventInstanceID, e2EventInstanceID) + + // Send notification to xApp that a Subscription Request has been processed. + resp := &models.SubscriptionResponse{ + SubscriptionID: restSubId, + SubscriptionInstances: []*models.SubscriptionInstance{ + &models.SubscriptionInstance{E2EventInstanceID: &e2EventInstanceID, + ErrorCause: nil, + XappEventInstanceID: &xAppEventInstanceID}, + }, + } + // Mark REST subscription request processesd. + restSubscription.SetProcessed() + xapp.Logger.Info("Sending successful REST notification to endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s", + clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans)) + + c.UpdateCounter(cRestSubNotifToXapp) + xapp.Subscription.Notify(resp, *clientEndpoint) +} + +//------------------------------------------------------------------- +// +//------------------------------------------------------------------- + +func (c *Control) processSubscriptionRequests(restSubscription *RESTSubscription, subReqList *e2ap.SubscriptionRequestList, + clientEndpoint *models.SubscriptionParamsClientEndpoint, meid *string, restSubId *string, xAppRmrEndpoint string) { + + xapp.Logger.Info("Subscription Request count=%v ", len(subReqList.E2APSubscriptionRequests)) + + var xAppEventInstanceID int64 + var e2EventInstanceID int64 + + defer restDuplicateCtrl.TransactionComplete(restSubscription.Md5sumOngoing) + + for index := 0; index < len(subReqList.E2APSubscriptionRequests); index++ { + subReqMsg := subReqList.E2APSubscriptionRequests[index] + xAppEventInstanceID = (int64)(subReqMsg.RequestId.Id) + + trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(xAppRmrEndpoint), *restSubId, subReqMsg.RequestId, &xapp.RMRMeid{RanName: *meid}) + if trans == nil { + // Send notification to xApp that prosessing of a Subscription Request has failed. + err := fmt.Errorf("Tracking failure") + c.sendUnsuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, err, clientEndpoint, trans) + continue + } + + xapp.Logger.Info("Handle SubscriptionRequest index=%v, %s", index, idstring(nil, trans)) + + subRespMsg, err := c.handleSubscriptionRequest(trans, &subReqMsg, meid, restSubId) + if err != nil { + c.sendUnsuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, err, clientEndpoint, trans) + } else { + e2EventInstanceID = (int64)(subRespMsg.RequestId.InstanceId) + xapp.Logger.Info("SubscriptionRequest index=%v processed successfully. endpoint=%v:%v, XappEventInstanceID=%v, E2EventInstanceID=%v, %s", + index, clientEndpoint.Host, *clientEndpoint.HTTPPort, xAppEventInstanceID, e2EventInstanceID, idstring(nil, trans)) + c.sendSuccesfullResponseNotification(restSubId, restSubscription, xAppEventInstanceID, e2EventInstanceID, clientEndpoint, trans) + } + trans.Release() + } +} + +//------------------------------------------------------------------- +// +//------------------------------------------------------------------ +func (c *Control) handleSubscriptionRequest(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest, meid *string, + restSubId *string) (*e2ap.E2APSubscriptionResponse, error) { + + err := c.tracker.Track(trans) + if err != nil { + xapp.Logger.Error("XAPP-SubReq Tracking error: %s", idstring(err, trans)) + err = fmt.Errorf("Tracking failure") + return nil, err + } + + subs, err := c.registry.AssignToSubscription(trans, subReqMsg, c.ResetTestFlag, c) + if err != nil { + xapp.Logger.Error("XAPP-SubReq Assign error: %s", idstring(err, trans)) + return nil, err + } + + // + // Wake subs request + // + go c.handleSubscriptionCreate(subs, trans) + event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side + + err = nil + if event != nil { + switch themsg := event.(type) { + case *e2ap.E2APSubscriptionResponse: + trans.Release() + return themsg, nil + case *e2ap.E2APSubscriptionFailure: + err = fmt.Errorf("E2 SubscriptionFailure received") + return nil, err + default: + err = fmt.Errorf("unexpected E2 subscription response received") + break + } + } else { + err = fmt.Errorf("E2 subscription response timeout") + } + + xapp.Logger.Error("XAPP-SubReq E2 subscription failed %s", idstring(err, trans, subs)) + c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c) + return nil, err +} + +//------------------------------------------------------------------- +// +//------------------------------------------------------------------- +func (c *Control) SubscriptionDeleteHandlerCB(restSubId string) error { + + c.CntRecvMsg++ + c.UpdateCounter(cRestSubDelReqFromXapp) + + xapp.Logger.Info("SubscriptionDeleteRequest from XAPP") + + restSubscription, err := c.registry.GetRESTSubscription(restSubId, true) + if err != nil { + xapp.Logger.Error("%s", err.Error()) + if restSubscription == nil { + // Subscription was not found + return nil + } else { + if restSubscription.SubReqOngoing == true { + err := fmt.Errorf("Handling of the REST Subscription Request still ongoing %s", restSubId) + xapp.Logger.Error("%s", err.Error()) + return err + } else if restSubscription.SubDelReqOngoing == true { + // Previous request for same restSubId still ongoing + return nil + } + } + } + + xAppRmrEndPoint := restSubscription.xAppRmrEndPoint + go func() { + for _, instanceId := range restSubscription.InstanceIds { + xAppEventInstanceID, err := c.SubscriptionDeleteHandler(&restSubId, &xAppRmrEndPoint, &restSubscription.Meid, instanceId) + + if err != nil { + xapp.Logger.Error("%s", err.Error()) + //return err + } + xapp.Logger.Info("Deleteting instanceId = %v", instanceId) + restSubscription.DeleteXappIdToE2Id(xAppEventInstanceID) + restSubscription.DeleteE2InstanceId(instanceId) + } + c.registry.DeleteRESTSubscription(&restSubId) + }() + + c.UpdateCounter(cRestSubDelRespToXapp) + return nil } +//------------------------------------------------------------------- +// +//------------------------------------------------------------------- +func (c *Control) SubscriptionDeleteHandler(restSubId *string, endPoint *string, meid *string, instanceId uint32) (int64, error) { + + var xAppEventInstanceID int64 + subs, err := c.registry.GetSubscriptionFirstMatch([]uint32{instanceId}) + if err != nil { + xapp.Logger.Info("Subscription Delete Handler subscription for restSubId=%v, E2EventInstanceID=%v not found %s", + restSubId, instanceId, idstring(err, nil)) + return xAppEventInstanceID, nil + } + + xAppEventInstanceID = int64(subs.ReqId.Id) + trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(*endPoint), *restSubId, e2ap.RequestId{subs.ReqId.Id, 0}, &xapp.RMRMeid{RanName: *meid}) + if trans == nil { + err := fmt.Errorf("XAPP-SubDelReq transaction not created. restSubId %s, endPoint %s, meid %s, instanceId %v", *restSubId, *endPoint, *meid, instanceId) + xapp.Logger.Error("%s", err.Error()) + } + defer trans.Release() + + err = c.tracker.Track(trans) + if err != nil { + err := fmt.Errorf("XAPP-SubDelReq %s:", idstring(err, trans)) + xapp.Logger.Error("%s", err.Error()) + return xAppEventInstanceID, &time.ParseError{} + } + // + // Wake subs delete + // + go c.handleSubscriptionDelete(subs, trans) + trans.WaitEvent(0) //blocked wait as timeout is handled in subs side + + xapp.Logger.Debug("XAPP-SubDelReq: Handling event %s ", idstring(nil, trans, subs)) + + c.registry.RemoveFromSubscription(subs, trans, waitRouteCleanup_ms, c) + + return xAppEventInstanceID, nil +} + +//------------------------------------------------------------------- +// +//------------------------------------------------------------------- func (c *Control) QueryHandler() (models.SubscriptionList, error) { xapp.Logger.Info("QueryHandler() called") + c.CntRecvMsg++ + return c.registry.QueryHandler() } @@ -283,7 +596,11 @@ func (c *Control) rmrSendToE2T(desc string, subs *Subscription, trans *Transacti params.Payload = trans.Payload.Buf params.Mbuf = nil xapp.Logger.Info("MSG to E2T: %s %s %s", desc, trans.String(), params.String()) - return c.SendWithRetry(params, false, 5) + err = c.SendWithRetry(params, false, 5) + if err != nil { + xapp.Logger.Error("rmrSendToE2T: Send failed: %+v", err) + } + return err } func (c *Control) rmrSendToXapp(desc string, subs *Subscription, trans *TransactionXapp) (err error) { @@ -298,7 +615,11 @@ func (c *Control) rmrSendToXapp(desc string, subs *Subscription, trans *Transact params.Payload = trans.Payload.Buf params.Mbuf = nil xapp.Logger.Info("MSG to XAPP: %s %s %s", desc, trans.String(), params.String()) - return c.SendWithRetry(params, false, 5) + err = c.SendWithRetry(params, false, 5) + if err != nil { + xapp.Logger.Error("rmrSendToXapp: Send failed: %+v", err) + } + return err } func (c *Control) Consume(msg *xapp.RMRParams) (err error) { @@ -492,7 +813,7 @@ func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *Tran case *e2ap.E2APSubscriptionFailure: removeSubscriptionFromDb = true subRfMsg, valid = subs.SetCachedResponse(event, false) - xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans)) + xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans)) c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans) case *SubmgrRestartTestEvent: // This simulates that no response has been received and after restart subscriptions are restored from db @@ -511,7 +832,7 @@ func (c *Control) handleSubscriptionCreate(subs *Subscription, parentTrans *Tran //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete) if valid == false { - c.registry.RemoveFromSubscription(subs, parentTrans, 5*time.Second, c) + c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanup_ms, c) } c.UpdateSubscriptionInDB(subs, removeSubscriptionFromDb) @@ -543,7 +864,7 @@ func (c *Control) handleSubscriptionDelete(subs *Subscription, parentTrans *Tran //Now RemoveFromSubscription in here to avoid race conditions (mostly concerns delete) // If parallel deletes ongoing both might pass earlier sendE2TSubscriptionDeleteRequest(...) if // RemoveFromSubscription locates in caller side (now in handleXAPPSubscriptionDeleteRequest(...)) - c.registry.RemoveFromSubscription(subs, parentTrans, 5*time.Second, c) + c.registry.RemoveFromSubscription(subs, parentTrans, waitRouteCleanup_ms, c) c.registry.UpdateSubscriptionToDb(subs, c) parentTrans.SendEvent(nil, 0) } @@ -555,9 +876,11 @@ func (c *Control) sendE2TSubscriptionRequest(subs *Subscription, trans *Transact var err error var event interface{} = nil var timedOut bool = false + const ricRequestorId = 123 subReqMsg := subs.SubReqMsg subReqMsg.RequestId = subs.GetReqId().RequestId + subReqMsg.RequestId.Id = ricRequestorId trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionRequest(subReqMsg) if err != nil { xapp.Logger.Error("SUBS-SubReq: %s", idstring(err, trans, subs, parentTrans)) @@ -566,6 +889,7 @@ func (c *Control) sendE2TSubscriptionRequest(subs *Subscription, trans *Transact // Write uncompleted subscrition in db. If no response for subscrition it need to be re-processed (deleted) after restart c.WriteSubscriptionToDb(subs) + for retries := uint64(0); retries < e2tMaxSubReqTryCount; retries++ { desc := fmt.Sprintf("(retry %d)", retries) if retries == 0 { @@ -598,9 +922,11 @@ func (c *Control) sendE2TSubscriptionDeleteRequest(subs *Subscription, trans *Tr var err error var event interface{} var timedOut bool + const ricRequestorId = 123 subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{} subDelReqMsg.RequestId = subs.GetReqId().RequestId + subDelReqMsg.RequestId.Id = ricRequestorId subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg) if err != nil { @@ -633,6 +959,7 @@ func (c *Control) sendE2TSubscriptionDeleteRequest(subs *Subscription, trans *Tr func (c *Control) handleE2TSubscriptionResponse(params *xapp.RMRParams) { xapp.Logger.Info("MSG from E2T: %s", params.String()) c.UpdateCounter(cSubRespFromE2) + subRespMsg, err := c.e2ap.UnpackSubscriptionResponse(params.Payload) if err != nil { xapp.Logger.Error("MSG-SubResp %s", idstring(err, params)) @@ -755,14 +1082,14 @@ func typeofSubsMessage(v interface{}) string { return "NIL" } switch v.(type) { - case *e2ap.E2APSubscriptionRequest: - return "SubReq" + //case *e2ap.E2APSubscriptionRequest: + // return "SubReq" case *e2ap.E2APSubscriptionResponse: return "SubResp" case *e2ap.E2APSubscriptionFailure: return "SubFail" - case *e2ap.E2APSubscriptionDeleteRequest: - return "SubDelReq" + //case *e2ap.E2APSubscriptionDeleteRequest: + // return "SubDelReq" case *e2ap.E2APSubscriptionDeleteResponse: return "SubDelResp" case *e2ap.E2APSubscriptionDeleteFailure: @@ -813,11 +1140,13 @@ func (c *Control) RemoveSubscriptionFromDb(subs *Subscription) { func (c *Control) SendSubscriptionDeleteReq(subs *Subscription) { + const ricRequestorId = 123 xapp.Logger.Debug("Sending subscription delete due to restart. subId = %v", subs.ReqId.InstanceId) // Send delete for every endpoint in the subscription subDelReqMsg := &e2ap.E2APSubscriptionDeleteRequest{} subDelReqMsg.RequestId = subs.GetReqId().RequestId + subDelReqMsg.RequestId.Id = ricRequestorId subDelReqMsg.FunctionId = subs.SubReqMsg.FunctionId mType, payload, err := c.e2ap.PackSubscriptionDeleteRequest(subDelReqMsg) if err != nil { @@ -834,13 +1163,61 @@ func (c *Control) SendSubscriptionDeleteReq(subs *Subscription) { params.PayloadLen = len(payload.Buf) params.Payload = payload.Buf params.Mbuf = nil - - if params == nil { - xapp.Logger.Error("SendSubscriptionDeleteReq() params == nil") - return - } - subs.DeleteFromDb = true c.handleXAPPSubscriptionDeleteRequest(params) } } + +func (c *Control) PrintRESTSubscriptionRequest(p *models.SubscriptionParams) { + + fmt.Println("CRESTSubscriptionRequest") + + if p.SubscriptionID != "" { + fmt.Println(" SubscriptionID = ", p.SubscriptionID) + } else { + fmt.Println(" SubscriptionID = ''") + } + + fmt.Printf(" ClientEndpoint.Host = %s\n", p.ClientEndpoint.Host) + + if p.ClientEndpoint.HTTPPort != nil { + fmt.Printf(" ClientEndpoint.HTTPPort = %v\n", *p.ClientEndpoint.HTTPPort) + } else { + fmt.Println(" ClientEndpoint.HTTPPort = nil") + } + + if p.ClientEndpoint.RMRPort != nil { + fmt.Printf(" ClientEndpoint.RMRPort = %v\n", *p.ClientEndpoint.RMRPort) + } else { + fmt.Println(" ClientEndpoint.RMRPort = nil") + } + + if p.Meid != nil { + fmt.Printf(" Meid = %s\n", *p.Meid) + } else { + fmt.Println(" Meid = nil") + } + + for _, subscriptionDetail := range p.SubscriptionDetails { + if p.RANFunctionID != nil { + fmt.Printf(" RANFunctionID = %v\n", *p.RANFunctionID) + } else { + fmt.Println(" RANFunctionID = nil") + } + fmt.Printf(" SubscriptionDetail.XappEventInstanceID = %v\n", *subscriptionDetail.XappEventInstanceID) + fmt.Printf(" SubscriptionDetail.EventTriggers = %v\n", subscriptionDetail.EventTriggers) + + for _, actionToBeSetup := range subscriptionDetail.ActionToBeSetupList { + fmt.Printf(" SubscriptionDetail.ActionToBeSetup.ActionID = %v\n", *actionToBeSetup.ActionID) + fmt.Printf(" SubscriptionDetail.ActionToBeSetup.ActionType = %s\n", *actionToBeSetup.ActionType) + fmt.Printf(" SubscriptionDetail.ActionToBeSetup.ActionDefinition = %v\n", actionToBeSetup.ActionDefinition) + + if actionToBeSetup.SubsequentAction != nil { + fmt.Printf(" SubscriptionDetail.ActionToBeSetup.SubsequentAction.SubsequentActionType = %s\n", *actionToBeSetup.SubsequentAction.SubsequentActionType) + fmt.Printf(" SubscriptionDetail.ActionToBeSetup..SubsequentAction.TimeToWait = %s\n", *actionToBeSetup.SubsequentAction.TimeToWait) + } else { + fmt.Println(" SubscriptionDetail.ActionToBeSetup.SubsequentAction = nil") + } + } + } +}