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         status := false
 151         i := 1
 152         for ; i <= seconds*2 && status == false; i++ {
 153
 154                 rmr.GetLogger().Info("SEND TESTPING: %s", params.String())
 155                 rmr.RmrSend(params)
 156
 157                 status = true
 158                 for _, val := range allRmrStubs {
 159                         if val.IsActive() == false {
 160                                 status = false
 161                                 break
 162                         }
 163                 }
 164                 if status == true {
 165                         break
 166                 }
 167                 rmr.GetLogger().Info("Sleep 0.5 secs and try routes again")
 168                 time.Sleep(500 * time.Millisecond)
 169         }
 170
 171         if status == false {
 172                 rmr.GetLogger().Error("Could not initialize routes")
 173         }
 174         return status
 175 }