--- /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