+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+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 {
+ return nil
+}
+
+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)
+ s := pathParams["testId"]
+
+ // This can be used to delete single subscription from db
+ if contains := strings.Contains(s, "deletesubid="); contains == true {
+ var splits = strings.Split(s, "=")
+ if subId, err := strconv.ParseInt(splits[1], 10, 64); err == nil {
+ xapp.Logger.Info("RemoveSubscriptionFromSdl() called. subId = %v", subId)
+ c.RemoveSubscriptionFromSdl(uint32(subId))
+ return
+ }
+ }
+
+ // This can be used to remove all subscriptions db from
+ if s == "emptydb" {
+ xapp.Logger.Info("RemoveAllSubscriptionsFromSdl() called")
+ c.RemoveAllSubscriptionsFromSdl()
+ return
+ }
+
+ // This is meant to cause submgr's restart in testing
+ if s == "restart" {
+ xapp.Logger.Info("os.Exit(1) called")
+ os.Exit(1)
+ }
+
+ xapp.Logger.Info("Unsupported rest command received %s", s)
+}
+
+//-------------------------------------------------------------------
+//
+//-------------------------------------------------------------------
+
+func (c *Control) rmrSendToE2T(desc string, subs *Subscription, trans *TransactionSubs) (err error) {
+ params := &xapp.RMRParams{}
+ params.Mtype = trans.GetMtype()
+ params.SubId = int(subs.GetReqId().InstanceId)
+ 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.SendWithRetry(params, false, 5)
+}
+
+func (c *Control) rmrSendToXapp(desc string, subs *Subscription, trans *TransactionXapp) (err error) {
+
+ params := &xapp.RMRParams{}
+ params.Mtype = trans.GetMtype()
+ params.SubId = int(subs.GetReqId().InstanceId)
+ 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.SendWithRetry(params, false, 5)
+}
+
+func (c *Control) Consume(msg *xapp.RMRParams) (err error) {
+ if c.RMRClient == nil {
+ err = fmt.Errorf("Rmr object nil can handle %s", msg.String())
+ xapp.Logger.Error("%s", err.Error())
+ return
+ }
+ c.CntRecvMsg++
+
+ defer c.RMRClient.Free(msg.Mbuf)
+
+ // xapp-frame might use direct access to c buffer and
+ // when msg.Mbuf is freed, someone might take it into use
+ // and payload data might be invalid inside message handle function
+ //
+ // subscriptions won't load system a lot so there is no
+ // real performance hit by cloning buffer into new go byte slice
+ cPay := append(msg.Payload[:0:0], msg.Payload...)
+ msg.Payload = cPay
+ msg.PayloadLen = len(cPay)
+
+ 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)