2 // Copyright 2019 AT&T Intellectual Property
3 // Copyright 2019 Nokia
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
9 // http://www.apache.org/licenses/LICENSE-2.0
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.
20 // #cgo LDFLAGS: -L/usr/local/lib -lrmr_nng -lnng
21 // #include <rmr/rmr.h>
22 // #include <stdlib.h>
26 "github.com/pkg/errors"
31 func (*Context) Init(port string, maxMsgSize int, maxRetries int, flags int) *Messenger {
33 defer C.free(unsafe.Pointer(pp))
34 ctx := NewContext(maxMsgSize, maxRetries, flags, C.rmr_init(pp, C.int(maxMsgSize), C.int(flags)))
37 time.Sleep(time.Second)
38 if time.Since(start) >= time.Minute {
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))
49 func (ctx *Context) SendMsg(msg *MBuf) (*MBuf, error) {
51 allocatedCMBuf, err := ctx.getAllocatedCRmrMBuf(msg, ctx.MaxMsgSize)
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)
59 defer C.rmr_free_msg(allocatedCMBuf)
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)
68 time.Sleep(100 * time.Millisecond)
71 return convertToMBuf(currCMBuf)
74 return nil, errors.New(fmt.Sprintf("#rmrCgoApi.SendMsg - Too many retries"))
77 func (ctx *Context) RecvMsg() (*MBuf, error) {
78 allocatedCMBuf, err := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize))
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)
86 defer C.rmr_free_msg(allocatedCMBuf)
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)
94 return convertToMBuf(currCMBuf)
97 func (ctx *Context) IsReady() bool {
98 return int(C.rmr_ready(ctx.RmrCtx)) != 0
101 func (ctx *Context) Close() {
102 C.rmr_close(ctx.RmrCtx)