tools/xapp_mock/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
  18 package rmr
  19
  20 // #cgo LDFLAGS: -L/usr/local/lib -lrmr_nng -lnng
  21 // #include <rmr/rmr.h>
  22 // #include <stdlib.h>
  23 import "C"
  24 import (
  25         "fmt"
  26         "github.com/pkg/errors"
  27         "time"
  28         "unsafe"
  29 )
  30
  31 func (*Context) Init(port string, maxMsgSize int, maxRetries int, flags int) *Messenger {
  32         pp := C.CString(port)
  33         defer C.free(unsafe.Pointer(pp))
  34         ctx := NewContext(maxMsgSize, maxRetries, flags, C.rmr_init(pp, C.int(maxMsgSize), C.int(flags)))
  35         start := time.Now()
  36         for !ctx.IsReady() {
  37                 time.Sleep(time.Second)
  38                 if time.Since(start) >= time.Minute {
  39                         start = time.Now()
  40                 }
  41         }
  42         // Configure the rmr to make rounds of attempts to send a message before notifying the application that it should retry.
  43         // Each round is about 1000 attempts with a short sleep between each round.
  44         C.rmr_set_stimeout(ctx.RmrCtx, C.int(0))
  45         r := Messenger(ctx)
  46         return &r
  47 }
  48
  49 func (ctx *Context) SendMsg(msg *MBuf) (*MBuf, error) {
  50
  51         allocatedCMBuf, err := ctx.getAllocatedCRmrMBuf(msg, ctx.MaxMsgSize)
  52         if err != nil{
  53                 return nil, err
  54         }
  55         if state := allocatedCMBuf.state; state != RMR_OK {
  56                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
  57                 return nil, errors.New(errorMessage)
  58         }
  59         defer C.rmr_free_msg(allocatedCMBuf)
  60
  61         for i:=0; i < ctx.MaxRetries; i++ {
  62                 currCMBuf := C.rmr_send_msg(ctx.RmrCtx, allocatedCMBuf)
  63                 if state := currCMBuf.state; state != RMR_OK {
  64                         if state != RMR_ERR_RETRY {
  65                                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
  66                                 return nil, errors.New(errorMessage)
  67                         }
  68                         time.Sleep(100*time.Millisecond)
  69                         continue
  70                 }
  71                 return convertToMBuf(currCMBuf)
  72         }
  73
  74         return nil, errors.New(fmt.Sprintf("#rmrCgoApi.SendMsg - Too many retries"))
  75 }
  76
  77 func (ctx *Context) RecvMsg() (*MBuf, error) {
  78         allocatedCMBuf, err :=C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize))
  79         if err != nil{
  80                 return nil, err
  81         }
  82         if state := allocatedCMBuf.state;state != RMR_OK {
  83                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
  84                 return nil, errors.New(errorMessage)
  85         }
  86         defer C.rmr_free_msg(allocatedCMBuf)
  87
  88         currCMBuf := C.rmr_rcv_msg(ctx.RmrCtx, allocatedCMBuf)
  89         if state := currCMBuf.state; state != RMR_OK {
  90                 errorMessage := fmt.Sprintf("#rmrCgoApi.RecvMsg - Failed to receive message. state: %v - %s", state, states[int(state)])
  91                 return nil, errors.New(errorMessage)
  92         }
  93
  94         return convertToMBuf(currCMBuf)
  95 }
  96
  97
  98 func (ctx *Context) IsReady() bool {
  99         return int(C.rmr_ready(ctx.RmrCtx)) != 0
 100 }
 101
 102 func (ctx *Context) Close() {
 103         C.rmr_close(ctx.RmrCtx)
 104 }