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         "strings"
  25         "testing"
  26         "time"
  27 )
  28
  29 //-----------------------------------------------------------------------------
  30 //
  31 //-----------------------------------------------------------------------------
  32 type RmrStubControl struct {
  33         RmrControl
  34         xapptweaks.RmrWrapper
  35         RecvChan chan *xapptweaks.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, rtfile string, port string, stat string, initMsg int) {
  71         tc.InitMsg = initMsg
  72         tc.Active = false
  73         tc.RecvChan = make(chan *xapptweaks.RMRParams)
  74         tc.RmrControl.Init(desc, rtfile, port)
  75         tc.RmrWrapper.Init()
  76
  77         tc.Rmr = xapp.NewRMRClientWithParams("tcp:"+port, 4096, 1, stat)
  78         tc.Rmr.SetReadyCB(tc.ReadyCB, nil)
  79         go tc.Rmr.Start(tc)
  80
  81         tc.WaitCB()
  82         allRmrStubs = append(allRmrStubs, tc)
  83 }
  84
  85 func (tc *RmrStubControl) Consume(params *xapp.RMRParams) (err error) {
  86         defer tc.Rmr.Free(params.Mbuf)
  87         msg := xapptweaks.NewParams(params)
  88         tc.CntRecvMsg++
  89
  90         if msg.Mtype == tc.InitMsg {
  91                 tc.Logger.Info("Testing message ignore %s", msg.String())
  92                 tc.SetActive()
  93                 return
  94         }
  95
  96         if tc.IsCheckXid() == true && strings.Contains(msg.Xid, tc.GetDesc()) == false {
  97                 tc.Logger.Info("Ignore %s", msg.String())
  98                 return
  99         }
 100
 101         tc.Logger.Info("Consume %s", msg.String())
 102         tc.PushMsg(msg)
 103         return
 104 }
 105
 106 func (tc *RmrStubControl) PushMsg(msg *xapptweaks.RMRParams) {
 107         tc.Logger.Debug("RmrStubControl PushMsg ... msg(%d) waiting", msg.Mtype)
 108         tc.RecvChan <- msg
 109         tc.Logger.Debug("RmrStubControl PushMsg ... done")
 110 }
 111
 112 func (tc *RmrStubControl) WaitMsg(secs time.Duration) *xapptweaks.RMRParams {
 113         tc.Logger.Debug("RmrStubControl WaitMsg ... waiting")
 114         if secs == 0 {
 115                 msg := <-tc.RecvChan
 116                 tc.Logger.Debug("RmrStubControl WaitMsg ... msg(%d) done", msg.Mtype)
 117                 return msg
 118         }
 119         select {
 120         case msg := <-tc.RecvChan:
 121                 tc.Logger.Debug("RmrStubControl WaitMsg ... msg(%d) done", msg.Mtype)
 122                 return msg
 123         case <-time.After(secs * time.Second):
 124                 tc.Logger.Debug("RmrStubControl WaitMsg ... timeout")
 125                 return nil
 126         }
 127         tc.Logger.Debug("RmrStubControl WaitMsg ... error")
 128         return nil
 129 }
 130
 131 var allRmrStubs []*RmrStubControl
 132
 133 //-----------------------------------------------------------------------------
 134 //
 135 //-----------------------------------------------------------------------------
 136
 137 func RmrStubControlWaitAlive(seconds int, mtype int, rmr xapptweaks.XAppWrapperIf) bool {
 138
 139         var dummyBuf []byte = make([]byte, 100)
 140
 141         params := xapptweaks.NewParams(nil)
 142         params.Mtype = mtype
 143         params.SubId = -1
 144         params.Payload = dummyBuf
 145         params.PayloadLen = 100
 146         params.Meid = &xapp.RMRMeid{RanName: "TESTPING"}
 147         params.Xid = "TESTPING"
 148         params.Mbuf = nil
 149
 150         if len(allRmrStubs) == 0 {
 151                 rmr.GetLogger().Info("No rmr stubs so no need to wait those to be alive")
 152                 return true
 153         }
 154         status := false
 155         i := 1
 156         for ; i <= seconds*2 && status == false; i++ {
 157
 158                 rmr.GetLogger().Info("SEND TESTPING: %s", params.String())
 159                 rmr.RmrSend(params, 0)
 160
 161                 status = true
 162                 for _, val := range allRmrStubs {
 163                         if val.IsActive() == false {
 164                                 status = false
 165                                 break
 166                         }
 167                 }
 168                 if status == true {
 169                         break
 170                 }
 171                 rmr.GetLogger().Info("Sleep 0.5 secs and try routes again")
 172                 time.Sleep(500 * time.Millisecond)
 173         }
 174
 175         if status == false {
 176                 rmr.GetLogger().Error("Could not initialize routes")
 177         }
 178         return status
 179 }