+++ /dev/null
-/*
-==================================================================================
- Copyright (c) 2020 AT&T Intellectual Property.
- Copyright (c) 2020 Nokia
-
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
-==================================================================================
-*/
-package main
-
-import (
- "encoding/json"
- "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
- "os"
- "strconv"
- "time"
-)
-
-var delay int // used for the delay receiver
-var handlerID string // used for the delay receiver too
-var doQuery bool // used for the query receiver
-
-type a1Receiver struct {
- msgChan chan *xapp.RMRParams
- appReady bool
- rmrReady bool
-}
-
-type policyRequest struct {
- Operation string `json:"operation"`
- PolicyTypeID int `json:"policy_type_id"`
- PolicyInstanceID string `json:"policy_instance_id"`
- Pay interface{} `json:"payload"`
-}
-
-type policyRequestResponse struct {
- PolicyTypeID int `json:"policy_type_id"`
- PolicyInstanceID string `json:"policy_instance_id"`
- HandlerID string `json:"handler_id"`
- Status string `json:"status"`
-}
-
-type policyQuery struct {
- PolicyTypeID int `json:"policy_type_id"`
-}
-
-func (e *a1Receiver) sendMsgRetry(params *xapp.RMRParams) {
- // helper for rmr that handles retries and sleep
- retries := 0
- for { // just keep trying until it works
- if e.rmrReady { // we must wait for ready, else SendMsg will blow with a nullptr
- if ok := xapp.Rmr.SendMsg(params); ok {
- xapp.Logger.Info("Msg successfully sent after %d retries!", retries)
- return
- }
- retries++
- //xapp.Logger.Info("Query failed to send...")
- } else {
- xapp.Logger.Info("rmr not ready...")
- time.Sleep(time.Duration(1) * time.Second)
- }
- }
-}
-
-func (e *a1Receiver) handlePolicyReq(msg *xapp.RMRParams) {
-
- // unmarshal the request
- var dat policyRequest
- if err := json.Unmarshal(msg.Payload, &dat); err != nil {
- panic(err)
- }
-
- var status string
- switch dat.Operation {
- case "CREATE":
- status = "OK"
- case "DELETE":
- status = "DELETED"
- }
-
- // form the response
- res := &policyRequestResponse{
- dat.PolicyTypeID,
- dat.PolicyInstanceID,
- "test_receiver",
- status,
- }
-
- outgoing, err := json.Marshal(res)
- if err != nil {
- panic(err)
- }
-
- /*
- WARNING:
- we want to use rts here. However, the current go xapp framework rts is broken.
- */
- params := &xapp.RMRParams{
- Mtype: 20011,
- Payload: outgoing,
- }
-
- if delay > 0 {
- xapp.Logger.Info("Xapp is sleeping...")
- time.Sleep(time.Duration(delay) * time.Second) // so much work to replicate python's time.sleep(5)...
- }
-
- e.sendMsgRetry(params)
-
- xapp.Logger.Info("Policy response sent!")
-}
-
-func (e *a1Receiver) sendQuery() {
- // form the query
- res := &policyQuery{
- 1006001,
- }
- outgoing, err := json.Marshal(res)
- if err != nil {
- panic(err)
- }
- params := &xapp.RMRParams{
- Mtype: 20012,
- Payload: outgoing,
- }
-
- for {
- /* We do this in a loop here, because even when the query first works, it could be the case that
- a1 does not even have the type yet, or there are no instances yet. In this integration test,
- we just keep pounding away so that eventually a1 returns the list this int test is looking for.
- A real xapp would NOT call the query in a loop like this.
- */
- e.sendMsgRetry(params)
- xapp.Logger.Info("Query sent successfully")
- time.Sleep(time.Duration(1) * time.Second)
- }
-}
-
-func (e *a1Receiver) messageLoop() {
- for {
- xapp.Logger.Info("Waiting for message..")
-
- msg := <-e.msgChan
-
- xapp.Logger.Info("Message received!")
- defer xapp.Rmr.Free(msg.Mbuf)
-
- switch msg.Mtype {
- case 20010:
- e.handlePolicyReq(msg)
- default:
- panic("Unexpected message type!")
- }
- }
-}
-
-// Consume: This named function is a required callback for e to use the xapp interface. it is called on all received rmr messages.
-func (e *a1Receiver) Consume(rp *xapp.RMRParams) (err error) {
- e.msgChan <- rp
- return
-}
-
-func (e *a1Receiver) Run() {
- // Set MDC (read: name visible in the logs)
- xapp.Logger.SetMdc(handlerID, "0.1.0")
-
- /* from reading the xapp frame code...
- this SetReadyCB sets off a chain of events..
- it sets readycb and readycbparams at the module level in xapp.go
- nothing happens yet..
- when the xapp is ran with` xapp.Run, this callback actually gets passed into the Rmr client which is not exposed in the xapp
- Rmr.SetReadyCB(xappReadyCb, nil)
- This "primes" the rmr client with it's own readycb, which is now set to this callback function
- When the rmr client is ready, it invokes the callback
- so basically, when rmr is ready, this function is invoked
- I think the xapp frame code could have been greatly simplified by just passing this into the invocation of Run() and then just passing that into the rmr client init!
- */
- xapp.SetReadyCB(func(d interface{}) { e.rmrReady = true }, true)
-
- // start message loop. We cannot wait for e.rmrReady here since that doesn't get populated until Run() runs.
- go e.messageLoop()
-
- if doQuery {
- // we are in the query tester; kick off a loop that does that until it works
- go e.sendQuery()
- }
-
- xapp.Run(e)
-}
-
-func newA1Receiver(appReady, rmrReady bool) *a1Receiver {
- return &a1Receiver{
- msgChan: make(chan *xapp.RMRParams),
- rmrReady: rmrReady,
- appReady: appReady,
- }
-}
-
-func main() {
-
- delay = 0
- if d, ok := os.LookupEnv("TEST_RCV_SEC_DELAY"); ok {
- delay, _ = strconv.Atoi(d)
- }
-
- handlerID = "test_receiver"
- if hid, ok := os.LookupEnv("HANDLER_ID"); ok {
- handlerID = hid
- }
-
- doQuery = false
- if _, ok := os.LookupEnv("DO_QUERY"); ok {
- doQuery = true
- }
-
- newA1Receiver(true, false).Run()
-}
Code Review / ric-plt / a1.git / blobdiff