Merge "Special version for alarm manager" into alarm_wa

[ric-plt/xapp-frame.git] / pkg / xapp / rmr.go

diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go
index e27d46e..9342828 100755 (executable)
--- a/pkg/xapp/rmr.go
+++ b/pkg/xapp/rmr.go
@@ -67,10 +67,11 @@ import (
        "bytes"
        "crypto/md5"
        "fmt"
-       "github.com/spf13/viper"
        "strings"
        "time"
        "unsafe"
+
+       "github.com/spf13/viper"
 )
 
 var RMRCounterOpts = []CounterOpts{
@@ -127,25 +128,64 @@ func (params *RMRParams) String() string {
 //-----------------------------------------------------------------------------
 //
 //-----------------------------------------------------------------------------
-func NewRMRClientWithParams(protPort string, maxSize int, threadType int, statDesc string) *RMRClient {
-       p := C.CString(protPort)
-       m := C.int(maxSize)
-       c := C.int(threadType)
+type RMRClientParams struct {
+       StatDesc string
+       RmrData  PortData
+}
+
+func (params *RMRClientParams) String() string {
+       return fmt.Sprintf("ProtPort=%d MaxSize=%d ThreadType=%d StatDesc=%s LowLatency=%t FastAck=%t Policies=%v",
+               params.RmrData.Port, params.RmrData.MaxSize, params.RmrData.ThreadType, params.StatDesc,
+               params.RmrData.LowLatency, params.RmrData.FastAck, params.RmrData.Policies)
+}
+
+//-----------------------------------------------------------------------------
+//
+//-----------------------------------------------------------------------------
+func NewRMRClientWithParams(params *RMRClientParams) *RMRClient {
+       p := C.CString(fmt.Sprintf("%d", params.RmrData.Port))
+       m := C.int(params.RmrData.MaxSize)
+       c := C.int(params.RmrData.ThreadType)
        defer C.free(unsafe.Pointer(p))
        ctx := C.rmr_init(p, m, c)
        if ctx == nil {
                Logger.Error("rmrClient: Initializing RMR context failed, bailing out!")
        }
+
+       Logger.Info("new rmrClient with parameters: %s", params.String())
+
+       if params.RmrData.LowLatency {
+               C.rmr_set_low_latency(ctx)
+       }
+       if params.RmrData.FastAck {
+               C.rmr_set_fack(ctx)
+       }
+
        return &RMRClient{
-               protPort:  protPort,
-               context:   ctx,
-               consumers: make([]MessageConsumer, 0),
-               stat:      Metric.RegisterCounterGroup(RMRCounterOpts, statDesc),
+               context:           ctx,
+               consumers:         make([]MessageConsumer, 0),
+               stat:              Metric.RegisterCounterGroup(RMRCounterOpts, params.StatDesc),
+               maxRetryOnFailure: params.RmrData.MaxRetryOnFailure,
        }
 }
 
 func NewRMRClient() *RMRClient {
-       return NewRMRClientWithParams(viper.GetString("rmr.protPort"), viper.GetInt("rmr.maxSize"), viper.GetInt("rmr.threadType"), "RMR")
+       p := GetPortData("rmrdata")
+       if p.Port == 0 || viper.IsSet("rmr.protPort") {
+               // Old xApp descriptor used, fallback to rmr section
+               fmt.Sscanf(viper.GetString("rmr.protPort"), "tcp:%d", &p.Port)
+               p.MaxSize = viper.GetInt("rmr.maxSize")
+               p.ThreadType = viper.GetInt("rmr.threadType")
+               p.LowLatency = viper.GetBool("rmr.lowLatency")
+               p.FastAck = viper.GetBool("rmr.fastAck")
+               p.MaxRetryOnFailure = viper.GetInt("rmr.maxRetryOnFailure")
+       }
+
+       return NewRMRClientWithParams(
+               &RMRClientParams{
+                       RmrData:  p,
+                       StatDesc: "RMR",
+               })
 }
 
 func (m *RMRClient) Start(c MessageConsumer) {
@@ -240,19 +280,21 @@ func (m *RMRClient) parseMessage(rxBuffer *C.rmr_mbuf_t) {
                return
        }
 
-       // Special case for multiple consumers
-       for _, c := range m.consumers {
-               cptr := unsafe.Pointer(rxBuffer.payload)
-               params.Payload = C.GoBytes(cptr, C.int(rxBuffer.len))
-               params.PayloadLen = int(rxBuffer.len)
-               params.Mtype = int(rxBuffer.mtype)
-               params.SubId = int(rxBuffer.sub_id)
+       /*
+               // Special case for multiple consumers
+               for _, c := range m.consumers {
+                       cptr := unsafe.Pointer(rxBuffer.payload)
+                       params.Payload = C.GoBytes(cptr, C.int(rxBuffer.len))
+                       params.PayloadLen = int(rxBuffer.len)
+                       params.Mtype = int(rxBuffer.mtype)
+                       params.SubId = int(rxBuffer.sub_id)
 
-               err := c.Consume(params)
-               if err != nil {
-                       Logger.Warn("rmrClient: Consumer returned error: %v", err)
+                       err := c.Consume(params)
+                       if err != nil {
+                               Logger.Warn("rmrClient: Consumer returned error: %v", err)
+                       }
                }
-       }
+       */
 }
 
 func (m *RMRClient) Allocate(size int) *C.rmr_mbuf_t {
@@ -374,8 +416,7 @@ func (m *RMRClient) Send(params *RMRParams, isRts bool) bool {
 
 func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int {
        var (
-               currBuffer  *C.rmr_mbuf_t
-               counterName string = "Transmitted"
+               currBuffer *C.rmr_mbuf_t
        )
 
        m.contextMux.Lock()
@@ -397,12 +438,11 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int {
        }
 
        // Just quick retry seems to help for K8s issue
-       maxRetryOnFailure := viper.GetInt("rmr.maxRetryOnFailure")
-       if maxRetryOnFailure == 0 {
-               maxRetryOnFailure = 5
+       if m.maxRetryOnFailure == 0 {
+               m.maxRetryOnFailure = 5
        }
 
-       for j := 0; j < maxRetryOnFailure && currBuffer != nil && currBuffer.state == C.RMR_ERR_RETRY; j++ {
+       for j := 0; j < m.maxRetryOnFailure && currBuffer != nil && currBuffer.state == C.RMR_ERR_RETRY; j++ {
                m.contextMux.Lock()
                if whid != 0 {
                        currBuffer = C.rmr_wh_send_msg(m.context, C.rmr_whid_t(whid), txBuffer)
@@ -416,15 +456,21 @@ func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int {
                m.contextMux.Unlock()
        }
 
-       if currBuffer.state != C.RMR_OK {
-               counterName = "TransmitError"
+       if currBuffer == nil {
+               m.UpdateStatCounter("TransmitError")
                m.LogMBufError("SendBuf failed", currBuffer)
+               return int(C.RMR_ERR_INITFAILED)
        }
 
-       m.UpdateStatCounter(counterName)
+       if currBuffer.state != C.RMR_OK {
+               m.UpdateStatCounter("TransmitError")
+               m.LogMBufError("SendBuf failed", currBuffer)
+       } else {
+               m.UpdateStatCounter("Transmitted")
+       }
        defer m.Free(currBuffer)
-
        return int(currBuffer.state)
+
 }
 
 func (m *RMRClient) SendCallMsg(params *RMRParams) (int, string) {
@@ -535,11 +581,10 @@ func (m *RMRClient) GetRicMessageName(id int) (s string) {
 }
 
 func (m *RMRClient) LogMBufError(text string, mbuf *C.rmr_mbuf_t) int {
-       Logger.Debug(fmt.Sprintf("rmrClient: %s -> [tp=%v] %v - %s", text, mbuf.tp_state, mbuf.state, RMRErrors[int(mbuf.state)]))
-       return int(mbuf.state)
-}
-
-// To be removed ...
-func (m *RMRClient) GetStat() (r RMRStatistics) {
-       return
+       if mbuf != nil {
+               Logger.Debug(fmt.Sprintf("rmrClient: %s -> [tp=%v] %v - %s", text, mbuf.tp_state, mbuf.state, RMRErrors[int(mbuf.state)]))
+               return int(mbuf.state)
+       }
+       Logger.Debug(fmt.Sprintf("rmrClient: %s -> mbuf nil", text))
+       return 0
 }