RSM/rmrcgo/rmr_c_go_api.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 rmrcgo
  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         "strings"
  28         "time"
  29         "unsafe"
  30
  31         "rsm/logger"
  32 )
  33
  34 func (*Context) Init(readyIntervalSec int, port string, maxMsgSize int, flags int, logger *logger.Logger) RmrMessenger {
  35         pp := C.CString(port)
  36         defer C.free(unsafe.Pointer(pp))
  37         logger.Debugf("#rmr_c_go_api.Init - Going to initiate RMR router")
  38         ctx := NewContext(maxMsgSize, flags, C.rmr_init(pp, C.int(maxMsgSize), C.int(flags)), logger)
  39         start := time.Now()
  40         ticker := time.NewTicker(time.Duration(readyIntervalSec) * time.Second)
  41         for ; !ctx.IsReady(); <-ticker.C {
  42                 if time.Since(start) >= time.Minute {
  43                         logger.Debugf("#rmr_c_go_api.Init - Routing table is not ready")
  44                         start = time.Now()
  45                 }
  46         }
  47         logger.Infof("#rmr_c_go_api.Init - RMR router has been initiated")
  48
  49         // Configure the rmr to make rounds of attempts to send a message before notifying the application that it should retry.
  50         // Each round is about 1000 attempts with a short sleep between each round.
  51         C.rmr_set_stimeout(ctx.RmrCtx, C.int(1000))
  52         return ctx
  53 }
  54
  55 func (ctx *Context) SendMsg(msg *MBuf) (*MBuf, error) {
  56         ctx.checkContextInitialized()
  57         ctx.Logger.Debugf("#rmr_c_go_api.SendMsg - Going to send message. MBuf: %v", *msg)
  58         allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, ctx.MaxMsgSize)
  59         defer C.rmr_free_msg(allocatedCMBuf)
  60         state := allocatedCMBuf.state
  61         if state != RMR_OK {
  62                 errorMessage := fmt.Sprintf("#rmr_c_go_api.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
  63                 return nil, errors.New(errorMessage)
  64         }
  65
  66         transactionId := string(*msg.XAction)
  67         tmpTid := strings.TrimSpace(transactionId)
  68         ctx.Logger.Infof("[RSM -> RMR] #rmr_c_go_api.SendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid)
  69
  70         currCMBuf := C.rmr_send_msg(ctx.RmrCtx, allocatedCMBuf)
  71         state = currCMBuf.state
  72
  73         if ctx.Logger.DebugEnabled() {
  74                 ctx.Logger.Debugf("#rmr_c_go_api.SendMsg - The current message  state: %v, message buffer:%v", state, currCMBuf)
  75         }
  76
  77         if state != RMR_OK {
  78                 errorMessage := fmt.Sprintf("#rmr_c_go_api.SendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
  79                 return nil, errors.New(errorMessage)
  80         }
  81
  82         ctx.Logger.Debugf("#rmr_c_go_api.SendMsg - The message has been sent successfully ")
  83         return convertToMBuf(currCMBuf), nil
  84 }
  85
  86 func (ctx *Context) RecvMsg() (*MBuf, error) {
  87         ctx.checkContextInitialized()
  88         ctx.Logger.Debugf("#rmr_c_go_api.RecvMsg - Going to receive message")
  89         allocatedCMBuf := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize))
  90         defer C.rmr_free_msg(allocatedCMBuf)
  91
  92         currCMBuf := C.rmr_rcv_msg(ctx.RmrCtx, allocatedCMBuf)
  93         state := currCMBuf.state
  94
  95         if state != RMR_OK {
  96                 errorMessage := fmt.Sprintf("#rmr_c_go_api.RecvMsg - Failed to receive message. state: %v - %s", state, states[int(state)])
  97                 ctx.Logger.Errorf(errorMessage)
  98                 return nil, errors.New(errorMessage)
  99         }
 100
 101         mbuf := convertToMBuf(currCMBuf)
 102         transactionId := string(*mbuf.XAction)
 103         tmpTid := strings.TrimSpace(transactionId)
 104         ctx.Logger.Infof("[RMR ->RSM] #rmr_c_go_api.RecvMsg - message %v has been received for transaction id: %s", *mbuf, tmpTid)
 105         return mbuf, nil
 106 }
 107
 108 func (ctx *Context) IsReady() bool {
 109         ctx.Logger.Debugf("#rmr_c_go_api.IsReady - Going to check if routing table is initialized")
 110         return int(C.rmr_ready(ctx.RmrCtx)) != 0
 111 }
 112
 113 func (ctx *Context) Close() {
 114         ctx.Logger.Debugf("#rmr_c_go_api.Close - Going to close RMR context")
 115         C.rmr_close(ctx.RmrCtx)
 116         time.Sleep(100 * time.Millisecond)
 117 }