Merge "[RICPLT-1852] Supports E2T Initialize - missing protocol and ran name in setup...
[ric-plt/e2mgr.git] / E2Manager / rmrCgo / 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 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         "e2mgr/logger"
32 )
33
34 func (*Context) Init(port string, maxMsgSize int, flags int, logger *logger.Logger) *RmrMessenger {//TODO remove pointer from interface
35         pp := C.CString(port)
36         defer C.free(unsafe.Pointer(pp))
37         logger.Debugf("#rmrCgoApi.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         //TODO use time.Ticker()
41         for !ctx.IsReady() {
42                 time.Sleep(time.Second)
43                 if time.Since(start) >= time.Minute {
44                         logger.Debugf("#rmrCgoApi.Init - Routing table is not ready")
45                         start = time.Now()
46                 }
47         }
48         logger.Infof("#rmrCgoApi.Init - RMR router has been initiated")
49
50         // Configure the rmr to make rounds of attempts to send a message before notifying the application that it should retry.
51         // Each round is about 1000 attempts with a short sleep between each round.
52         C.rmr_set_stimeout(ctx.RmrCtx, C.int(1000))
53         r := RmrMessenger(ctx)
54         return &r
55 }
56
57 func (ctx *Context) SendMsg(msg *MBuf, maxMsgSize int) (*MBuf, error) {
58         ctx.checkContextInitialized()
59         ctx.Logger.Debugf("#rmrCgoApi.SendMsg - Going to send message. MBuf: %v", *msg)
60         allocatedCMBuf := ctx.getAllocatedCRmrMBuf(ctx.Logger, msg, maxMsgSize)
61         defer C.rmr_free_msg(allocatedCMBuf)
62         state := allocatedCMBuf.state
63         if state != RMR_OK {
64                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to get allocated message. state: %v - %s", state, states[int(state)])
65                 return nil, errors.New(errorMessage)
66         }
67
68         //TODO: if debug enabled
69         transactionId := string(*msg.XAction)
70         tmpTid := strings.TrimSpace(transactionId)
71         ctx.Logger.Infof("[E2 Manager -> RMR] #rmrCgoApi.SendMsg - Going to send message %v for transaction id: %s", *msg, tmpTid)
72
73         currCMBuf := C.rmr_send_msg(ctx.RmrCtx, allocatedCMBuf)
74         state = currCMBuf.state
75         ctx.Logger.Debugf("#rmrCgoApi.SendMsg - The current message  state: %v, message buffer:%v", state, currCMBuf)
76
77         if state != RMR_OK {
78                 errorMessage := fmt.Sprintf("#rmrCgoApi.SendMsg - Failed to send message. state: %v - %s", state, states[int(state)])
79                 return nil, errors.New(errorMessage)
80         }
81
82         ctx.Logger.Debugf("#rmrCgoApi.SendMsg - The message has been sent successfully ")
83         return convertToMBuf(ctx.Logger, currCMBuf), nil
84 }
85
86 func (ctx *Context) RecvMsg() (*MBuf, error) {
87         ctx.checkContextInitialized()
88         ctx.Logger.Debugf("#rmrCgoApi.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("#rmrCgoApi.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(ctx.Logger, currCMBuf)
102         transactionId := string(*mbuf.XAction)
103         tmpTid := strings.TrimSpace(transactionId)
104         ctx.Logger.Infof("[RMR -> E2 Manager] #rmrCgoApi.RecvMsg - message %v has been received for transaction id: %s", *mbuf, tmpTid)
105         return mbuf, nil
106 }
107
108 func (ctx *Context) RtsMsg(msg *MBuf) {
109         ctx.checkContextInitialized()
110         ctx.Logger.Debugf("#rmrCgoApi.RtsMsg - Going to return message to the sender")
111         allocatedCMBuf := C.rmr_alloc_msg(ctx.RmrCtx, C.int(ctx.MaxMsgSize))
112         defer C.rmr_free_msg(allocatedCMBuf)
113         C.rmr_rts_msg(ctx.RmrCtx, allocatedCMBuf)
114 }
115
116 func (ctx *Context) IsReady() bool {
117         ctx.Logger.Debugf("#rmrCgoApi.IsReady - Going to check if routing table is initialized")
118         return int(C.rmr_ready(ctx.RmrCtx)) != 0
119 }
120
121 func (ctx *Context) Close() {
122         ctx.Logger.Debugf("#rmrCgoApi.Close - Going to close RMR context")
123         C.rmr_close(ctx.RmrCtx)
124         time.Sleep(100 * time.Millisecond)
125 }