pkg/teststub/controlRmrStub.go

   1 /*
   2 ==================================================================================
   3   Copyright (c) 2019 AT&T Intellectual Property.
   4   Copyright (c) 2019 Nokia
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10        http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17 ==================================================================================
  18 */
  19 package teststub
  20
  21 import (
  22         "gerrit.o-ran-sc.org/r/ric-plt/submgr/pkg/xapptweaks"
  23         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
  24         "strconv"
  25         "strings"
  26         "testing"
  27         "time"
  28 )
  29
  30 //-----------------------------------------------------------------------------
  31 //
  32 //-----------------------------------------------------------------------------
  33 type RmrStubControl struct {
  34         RmrControl
  35         xapptweaks.RmrWrapper
  36         RecvChan chan *xapptweaks.RMRParams
  37         Active   bool
  38         InitMsg  int
  39         CheckXid bool
  40 }
  41
  42 func (tc *RmrStubControl) SetActive() {
  43         tc.Active = true
  44 }
  45
  46 func (tc *RmrStubControl) IsActive() bool {
  47         return tc.Active
  48 }
  49
  50 func (tc *RmrStubControl) SetCheckXid(val bool) {
  51         tc.CheckXid = val
  52 }
  53
  54 func (tc *RmrStubControl) IsCheckXid() bool {
  55         return tc.CheckXid
  56 }
  57
  58 func (tc *RmrStubControl) IsChanEmpty() bool {
  59         if len(tc.RecvChan) > 0 {
  60                 return false
  61         }
  62         return true
  63 }
  64
  65 func (tc *RmrStubControl) TestMsgChanEmpty(t *testing.T) {
  66         if tc.IsChanEmpty() == false {
  67                 tc.TestError(t, "message channel not empty")
  68         }
  69 }
  70
  71 func (tc *RmrStubControl) Init(desc string, srcId RmrSrcId, rtgSvc RmrRtgSvc, stat string, initMsg int) {
  72         tc.InitMsg = initMsg
  73         tc.Active = false
  74         tc.RecvChan = make(chan *xapptweaks.RMRParams)
  75         tc.RmrControl.Init(desc, srcId, rtgSvc)
  76         tc.RmrWrapper.Init()
  77
  78         tc.Rmr = xapp.NewRMRClientWithParams("tcp:"+strconv.FormatUint(uint64(srcId.Port), 10), 65534, 1, 0, stat)
  79         tc.Rmr.SetReadyCB(tc.ReadyCB, nil)
  80         go tc.Rmr.Start(tc)
  81
  82         tc.WaitCB()
  83         allRmrStubs = append(allRmrStubs, tc)
  84 }
  85
  86 func (tc *RmrStubControl) Consume(params *xapp.RMRParams) (err error) {
  87         defer tc.Rmr.Free(params.Mbuf)
  88         msg := xapptweaks.NewParams(params)
  89         tc.CntRecvMsg++
  90
  91         cPay := append(msg.Payload[:0:0], msg.Payload...)
  92         msg.Payload = cPay
  93         msg.PayloadLen = len(cPay)
  94
  95         if msg.Mtype == tc.InitMsg {
  96                 tc.Logger.Info("Testing message ignore %s", msg.String())
  97                 tc.SetActive()
  98                 return
  99         }
 100
 101         if tc.IsCheckXid() == true && strings.Contains(msg.Xid, tc.GetDesc()) == false {
 102                 tc.Logger.Info("Ignore %s", msg.String())
 103                 return
 104         }
 105
 106         tc.Logger.Info("Consume %s", msg.String())
 107         tc.PushMsg(msg)
 108         return
 109 }
 110
 111 func (tc *RmrStubControl) PushMsg(msg *xapptweaks.RMRParams) {
 112         tc.Logger.Debug("RmrStubControl PushMsg ... msg(%d) waiting", msg.Mtype)
 113         tc.RecvChan <- msg
 114         tc.Logger.Debug("RmrStubControl PushMsg ... done")
 115 }
 116
 117 func (tc *RmrStubControl) WaitMsg(secs time.Duration) *xapptweaks.RMRParams {
 118         tc.Logger.Debug("RmrStubControl WaitMsg ... waiting")
 119         if secs == 0 {
 120                 msg := <-tc.RecvChan
 121                 tc.Logger.Debug("RmrStubControl WaitMsg ... msg(%d) done", msg.Mtype)
 122                 return msg
 123         }
 124         select {
 125         case msg := <-tc.RecvChan:
 126                 tc.Logger.Debug("RmrStubControl WaitMsg ... msg(%d) done", msg.Mtype)
 127                 return msg
 128         case <-time.After(secs * time.Second):
 129                 tc.Logger.Debug("RmrStubControl WaitMsg ... timeout")
 130                 return nil
 131         }
 132         tc.Logger.Debug("RmrStubControl WaitMsg ... error")
 133         return nil
 134 }
 135
 136 var allRmrStubs []*RmrStubControl
 137
 138 //-----------------------------------------------------------------------------
 139 //
 140 //-----------------------------------------------------------------------------
 141
 142 func RmrStubControlWaitAlive(seconds int, mtype int, rmr xapptweaks.XAppWrapperIf) bool {
 143
 144         var dummyBuf []byte = make([]byte, 100)
 145
 146         params := xapptweaks.NewParams(nil)
 147         params.Mtype = mtype
 148         params.SubId = -1
 149         params.Payload = dummyBuf
 150         params.PayloadLen = 100
 151         params.Meid = &xapp.RMRMeid{RanName: "TESTPING"}
 152         params.Xid = "TESTPING"
 153         params.Mbuf = nil
 154
 155         if len(allRmrStubs) == 0 {
 156                 rmr.GetLogger().Info("No rmr stubs so no need to wait those to be alive")
 157                 return true
 158         }
 159         status := false
 160         i := 1
 161         for ; i <= seconds*2 && status == false; i++ {
 162
 163                 rmr.GetLogger().Info("SEND TESTPING: %s", params.String())
 164                 rmr.RmrSend(params, 0)
 165
 166                 status = true
 167                 for _, val := range allRmrStubs {
 168                         if val.IsActive() == false {
 169                                 status = false
 170                                 break
 171                         }
 172                 }
 173                 if status == true {
 174                         break
 175                 }
 176                 rmr.GetLogger().Info("Sleep 0.5 secs and try routes again")
 177                 time.Sleep(500 * time.Millisecond)
 178         }
 179
 180         if status == false {
 181                 rmr.GetLogger().Error("Could not initialize routes")
 182         }
 183         return status
 184 }