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/xapp-frame/pkg/xapp"
  23         "strconv"
  24         "strings"
  25         "testing"
  26         "time"
  27 )
  28
  29 //-----------------------------------------------------------------------------
  30 //
  31 //-----------------------------------------------------------------------------
  32 type RmrStubControl struct {
  33         RmrControl
  34         *xapp.RMRClient
  35         RecvChan chan *xapp.RMRParams
  36         Active   bool
  37         InitMsg  int
  38         CheckXid bool
  39 }
  40
  41 func (tc *RmrStubControl) SetActive() {
  42         tc.Active = true
  43 }
  44
  45 func (tc *RmrStubControl) IsActive() bool {
  46         return tc.Active
  47 }
  48
  49 func (tc *RmrStubControl) SetCheckXid(val bool) {
  50         tc.CheckXid = val
  51 }
  52
  53 func (tc *RmrStubControl) IsCheckXid() bool {
  54         return tc.CheckXid
  55 }
  56
  57 func (tc *RmrStubControl) IsChanEmpty() bool {
  58         if len(tc.RecvChan) > 0 {
  59                 return false
  60         }
  61         return true
  62 }
  63
  64 func (tc *RmrStubControl) TestMsgChanEmpty(t *testing.T) {
  65         if tc.IsChanEmpty() == false {
  66                 tc.TestError(t, "message channel not empty")
  67         }
  68 }
  69
  70 func (tc *RmrStubControl) Init(desc string, srcId RmrSrcId, rtgSvc RmrRtgSvc, stat string, initMsg int) {
  71         tc.InitMsg = initMsg
  72         tc.Active = false
  73         tc.RecvChan = make(chan *xapp.RMRParams)
  74         tc.RmrControl.Init(desc, srcId, rtgSvc)
  75
  76         tc.RMRClient = xapp.NewRMRClientWithParams(&xapp.RMRClientParams{
  77                 ProtPort:   "tcp:" + strconv.FormatUint(uint64(srcId.Port), 10),
  78                 MaxSize:    65534,
  79                 ThreadType: 0,
  80                 StatDesc:   stat,
  81                 LowLatency: false,
  82                 FastAck:    false,
  83         })
  84
  85         tc.RMRClient.SetReadyCB(tc.ReadyCB, nil)
  86         go tc.RMRClient.Start(tc)
  87
  88         tc.WaitCB()
  89         allRmrStubs = append(allRmrStubs, tc)
  90 }
  91
  92 func (tc *RmrStubControl) Consume(msg *xapp.RMRParams) (err error) {
  93         defer tc.RMRClient.Free(msg.Mbuf)
  94
  95         cPay := append(msg.Payload[:0:0], msg.Payload...)
  96         msg.Payload = cPay
  97         msg.PayloadLen = len(cPay)
  98
  99         if msg.Mtype == tc.InitMsg {
 100                 tc.Info("Testing message ignore %s", msg.String())
 101                 tc.SetActive()
 102                 return
 103         }
 104
 105         if tc.IsCheckXid() == true && strings.Contains(msg.Xid, tc.GetDesc()) == false {
 106                 tc.Info("Ignore %s", msg.String())
 107                 return
 108         }
 109
 110         tc.Info("Consume %s", msg.String())
 111         tc.PushMsg(msg)
 112         return
 113 }
 114
 115 func (tc *RmrStubControl) PushMsg(msg *xapp.RMRParams) {
 116         tc.Debug("RmrStubControl PushMsg ... msg(%d) waiting", msg.Mtype)
 117         tc.RecvChan <- msg
 118         tc.Debug("RmrStubControl PushMsg ... done")
 119 }
 120
 121 func (tc *RmrStubControl) WaitMsg(secs time.Duration) *xapp.RMRParams {
 122         tc.Debug("RmrStubControl WaitMsg ... waiting")
 123         if secs == 0 {
 124                 msg := <-tc.RecvChan
 125                 tc.Debug("RmrStubControl WaitMsg ... msg(%d) done", msg.Mtype)
 126                 return msg
 127         }
 128         select {
 129         case msg := <-tc.RecvChan:
 130                 tc.Debug("RmrStubControl WaitMsg ... msg(%d) done", msg.Mtype)
 131                 return msg
 132         case <-time.After(secs * time.Second):
 133                 tc.Debug("RmrStubControl WaitMsg ... timeout")
 134                 return nil
 135         }
 136         tc.Debug("RmrStubControl WaitMsg ... error")
 137         return nil
 138 }
 139
 140 var allRmrStubs []*RmrStubControl
 141
 142 //-----------------------------------------------------------------------------
 143 //
 144 //-----------------------------------------------------------------------------
 145
 146 func RmrStubControlWaitAlive(seconds int, mtype int, rmr *xapp.RMRClient, tent *TestWrapper) bool {
 147
 148         var dummyBuf []byte = make([]byte, 100)
 149
 150         params := &xapp.RMRParams{}
 151         params.Mtype = mtype
 152         params.SubId = -1
 153         params.Payload = dummyBuf
 154         params.PayloadLen = 100
 155         params.Meid = &xapp.RMRMeid{RanName: "TESTPING"}
 156         params.Xid = "TESTPING"
 157         params.Mbuf = nil
 158
 159         if len(allRmrStubs) == 0 {
 160                 tent.Info("No rmr stubs so no need to wait those to be alive")
 161                 return true
 162         }
 163         status := false
 164         i := 1
 165         for ; i <= seconds*2 && status == false; i++ {
 166
 167                 tent.Info("SEND TESTPING: %s", params.String())
 168                 rmr.SendWithRetry(params, false, 0)
 169
 170                 status = true
 171                 for _, val := range allRmrStubs {
 172                         if val.IsActive() == false {
 173                                 status = false
 174                                 break
 175                         }
 176                 }
 177                 if status == true {
 178                         break
 179                 }
 180                 tent.Info("Sleep 0.5 secs and try routes again")
 181                 time.Sleep(500 * time.Millisecond)
 182         }
 183
 184         if status == false {
 185                 tent.Error("Could not initialize routes")
 186         }
 187         return status
 188 }