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 }