+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+func (c *Control) SubscriptionHandler(stype models.SubscriptionType, params interface{}) (models.SubscriptionResult, 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.SubscriptionResult{}, fmt.Errorf("Subscription rest interface not implemented")
+}
+
+func (c *Control) QueryHandler() (models.SubscriptionList, error) {
+ return c.registry.QueryHandler()
+}
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+
+func (c *Control) rmrSendToE2T(desc string, subs *Subscription, trans *TransactionSubs) (err error) {
+ params := xapptweaks.NewParams(nil)
+ params.Mtype = trans.GetMtype()
+ params.SubId = int(subs.GetReqId().Seq)
+ params.Xid = ""
+ params.Meid = subs.GetMeid()
+ params.Src = ""
+ params.PayloadLen = len(trans.Payload.Buf)
+ params.Payload = trans.Payload.Buf
+ params.Mbuf = nil
+ xapp.Logger.Info("MSG to E2T: %s %s %s", desc, trans.String(), params.String())
+ return c.RmrSend(params, 5)
+}
+
+func (c *Control) rmrSendToXapp(desc string, subs *Subscription, trans *TransactionXapp) (err error) {
+
+ params := xapptweaks.NewParams(nil)
+ params.Mtype = trans.GetMtype()
+ params.SubId = int(subs.GetReqId().Seq)
+ params.Xid = trans.GetXid()
+ params.Meid = trans.GetMeid()
+ params.Src = ""
+ params.PayloadLen = len(trans.Payload.Buf)
+ params.Payload = trans.Payload.Buf
+ params.Mbuf = nil
+ xapp.Logger.Info("MSG to XAPP: %s %s %s", desc, trans.String(), params.String())
+ return c.RmrSend(params, 5)
+}
+
+func (c *Control) Consume(params *xapp.RMRParams) (err error) {
+ msg := xapptweaks.NewParams(params)
+ if c.Rmr == nil {
+ err = fmt.Errorf("Rmr object nil can handle %s", msg.String())
+ xapp.Logger.Error("%s", err.Error())
+ return
+ }
+ c.CntRecvMsg++
+
+ defer c.Rmr.Free(msg.Mbuf)
+
+ switch msg.Mtype {
+ case xapp.RIC_SUB_REQ:
+ go c.handleXAPPSubscriptionRequest(msg)
+ case xapp.RIC_SUB_RESP:
+ go c.handleE2TSubscriptionResponse(msg)
+ case xapp.RIC_SUB_FAILURE:
+ go c.handleE2TSubscriptionFailure(msg)
+ case xapp.RIC_SUB_DEL_REQ:
+ go c.handleXAPPSubscriptionDeleteRequest(msg)
+ case xapp.RIC_SUB_DEL_RESP:
+ go c.handleE2TSubscriptionDeleteResponse(msg)
+ case xapp.RIC_SUB_DEL_FAILURE:
+ go c.handleE2TSubscriptionDeleteFailure(msg)