if err != nil {
retval += filler + "err(" + err.Error() + ")"
filler = " "
-
}
return retval
}
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)}
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,
//subscriber: subscriber,
Counters: xapp.Metric.RegisterCounterGroup(GetMetricsOpts(), "SUBMGR"),
}
+ c.ReadConfigParameters("")
// Register REST handler for testing support
xapp.Resource.InjectRoute("/ric/v1/test/{testId}", c.TestRestHandler, "POST")
go xapp.Subscription.Listen(c.SubscriptionHandler, c.QueryHandler, c.SubscriptionDeleteHandler)
- //go c.subscriber.Listen(c.SubscriptionHandler, c.QueryHandler)
if readSubsFromDb == "false" {
return c
//-------------------------------------------------------------------
//
//-------------------------------------------------------------------
-func ReadConfigParameters() {
+func (c *Control) ReadConfigParameters(f string) {
// viper.GetDuration returns nanoseconds
e2tSubReqTimeout = viper.GetDuration("controls.e2tSubReqTimeout_ms") * 1000000
func (c *Control) Run() {
xapp.SetReadyCB(c.ReadyCB, nil)
+ xapp.AddConfigChangeListener(c.ReadConfigParameters)
xapp.Run(c)
}
}
func (c *Control) QueryHandler() (models.SubscriptionList, error) {
+ xapp.Logger.Info("QueryHandler() called")
+
return c.registry.QueryHandler()
}
func (c *Control) TestRestHandler(w http.ResponseWriter, r *http.Request) {
-
xapp.Logger.Info("TestRestHandler() called")
pathParams := mux.Vars(r)
return
}
- trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subReqMsg.RequestId.InstanceId, params.Meid)
+ trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subReqMsg.RequestId, params.Meid)
if trans == nil {
xapp.Logger.Error("XAPP-SubReq: %s", idstring(fmt.Errorf("transaction not created"), params))
return
}
defer trans.Release()
- err = c.tracker.Track(trans)
- if err != nil {
+ if err = c.tracker.Track(trans); err != nil {
xapp.Logger.Error("XAPP-SubReq: %s", idstring(err, trans))
return
}
return
}
- //
- // Wake subs request
- //
+ c.wakeSubscriptionRequest(subs, trans)
+}
+
+//-------------------------------------------------------------------
+// Wake Subscription Request to E2node
+//------------------------------------------------------------------
+func (c *Control) wakeSubscriptionRequest(subs *Subscription, trans *TransactionXapp) {
+
go c.handleSubscriptionCreate(subs, trans)
event, _ := trans.WaitEvent(0) //blocked wait as timeout is handled in subs side
- err = nil
+ var err error
if event != nil {
switch themsg := event.(type) {
case *e2ap.E2APSubscriptionResponse:
+ themsg.RequestId.Id = trans.RequestId.Id
trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionResponse(themsg)
if err == nil {
trans.Release()
return
}
case *e2ap.E2APSubscriptionFailure:
+ themsg.RequestId.Id = trans.RequestId.Id
trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionFailure(themsg)
if err == nil {
c.UpdateCounter(cSubFailToXapp)
return
}
- trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subDelReqMsg.RequestId.InstanceId, params.Meid)
+ trans := c.tracker.NewXappTransaction(xapp.NewRmrEndpoint(params.Src), params.Xid, subDelReqMsg.RequestId, params.Meid)
if trans == nil {
xapp.Logger.Error("XAPP-SubDelReq: %s", idstring(fmt.Errorf("transaction not created"), params))
return
// Whatever is received success, fail or timeout, send successful delete response
subDelRespMsg := &e2ap.E2APSubscriptionDeleteResponse{}
- subDelRespMsg.RequestId = subs.GetReqId().RequestId
+ subDelRespMsg.RequestId.Id = trans.RequestId.Id
+ subDelRespMsg.RequestId.InstanceId = subs.GetReqId().RequestId.InstanceId
subDelRespMsg.FunctionId = subs.SubReqMsg.FunctionId
trans.Mtype, trans.Payload, err = c.e2ap.PackSubscriptionDeleteResponse(subDelRespMsg)
if err == nil {
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))
- c.sendE2TSubscriptionDeleteRequest(subs, trans, parentTrans)
+ xapp.Logger.Info("SUBS-SubReq: internal delete due event(%s) %s", typeofSubsMessage(event), idstring(nil, trans, subs, parentTrans))
case *SubmgrRestartTestEvent:
// This simulates that no response has been received and after restart subscriptions are restored from db
xapp.Logger.Debug("Test restart flag is active. Dropping this transaction to test restart case")