[RIC-621] Add redis monitor verification to automation tests
[ric-plt/e2mgr.git] / tools / xappmock / rmr / rmrCgoApi.go
1 //
2 // Copyright 2019 AT&T Intellectual Property
3 // Copyright 2019 Nokia
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 //      http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16
17 //  This source code is part of the near-RT RIC (RAN Intelligent Controller)
18 //  platform project (RICP).
19
20
21 package rmr
22
23 // #cgo LDFLAGS: -L/usr/local/lib -lrmr_nng -lnng
24 // #include <rmr/rmr.h>
25 // #include <stdlib.h>
26 import "C"
27 import (
28         "fmt"
29         "github.com/pkg/errors"
30         "time"
31         "unsafe"
32 )
33
34 func (*Context) Init(port string, maxMsgSize int, maxRetries int, flags int) *Messenger {
35         pp := C.CString(port)
36         defer C.free(unsafe.Pointer(pp))
37         ctx := NewContext(maxMsgSize, maxRetries, flags, C.rmr_init(pp, C.int(maxMsgSize), C.int(flags)))
38         start := time.Now()
39         for !ctx.IsReady() {
40                 time.Sleep(time.Second)
41                 if time.Since(start) >= time.Minute {
42                         start = time.Now()
43                 }
44         }
45         // Configure the rmr to make rounds of attempts to send a message before notifying the application that it should retry.
46         // Each round is about 1000 attempts with a short sleep between each round.
47         C.rmr_set_stimeout(ctx.RmrCtx, C.int(0))
48         r := Messenger(ctx)
49         return &r
50 }
51
52 func (ctx *Context) SendMsg(msg *MBuf) (*MBuf, error) {
53
54         allocatedCMBuf, err := ctx.getAllocatedCRmrMBuf(msg, ctx.MaxMsgSize)
55         if err != nil {
56                 return nil, err
57         }
58         if state := allocatedCMBuf.state; state != RMR_OK {
59                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
60                 return nil, errors.New(errorMessage)
61         }
62         defer C.rmr_free_msg(allocatedCMBuf)
63
64         for i := 0; i < ctx.MaxRetries; i++ {
65                 currCMBuf := C.rmr_send_msg(ctx.RmrCtx, allocatedCMBuf)
66                 if state := currCMBuf.state; state != RMR_OK {
67                         if state != RMR_ERR_RETRY {
68                                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
69                                 return nil, errors.New(errorMessage)
70                         }
71                         time.Sleep(100 * time.Millisecond)
72                         continue
73                 }
74                 return convertToMBuf(currCMBuf)
75         }
76
77         return nil, errors.New(fmt.Sprintf("#rmrCgoApi.SendMsg - Too many retries"))
78 }
79
80 func (ctx *Context) RecvMsg() (*MBuf, error) {
81         allocatedCMBuf, err := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize))
82         if err != nil {
83                 return nil, err
84         }
85         if state := allocatedCMBuf.state; state != RMR_OK {
86                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
87                 return nil, errors.New(errorMessage)
88         }
89         defer C.rmr_free_msg(allocatedCMBuf)
90
91         currCMBuf := C.rmr_rcv_msg(ctx.RmrCtx, allocatedCMBuf)
92         if state := currCMBuf.state; state != RMR_OK {
93                 errorMessage := fmt.Sprintf("#rmrCgoApi.RecvMsg - Failed to receive message. state: %v - %s", state, states[int(state)])
94                 return nil, errors.New(errorMessage)
95         }
96
97         return convertToMBuf(currCMBuf)
98 }
99
100 func (ctx *Context) IsReady() bool {
101         return int(C.rmr_ready(ctx.RmrCtx)) != 0
102 }
103
104 func (ctx *Context) Close() {
105         C.rmr_close(ctx.RmrCtx)
106 }