Backward compatible with old xApp descriptor

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

diff --git a/pkg/xapp/rmr.go b/pkg/xapp/rmr.go
index 530f223..ef8aea5 100755 (executable)
--- a/pkg/xapp/rmr.go
+++ b/pkg/xapp/rmr.go
@@ -128,26 +128,23 @@ func (params *RMRParams) String() string {
 //
 //-----------------------------------------------------------------------------
 type RMRClientParams struct {
-       protPort   string
-       maxSize    int
-       threadType int
-       statDesc   string
-       lowLatency bool
-       fastAck    bool
+       StatDesc string
+       RmrData  PortData
 }
 
 func (params *RMRClientParams) String() string {
-       return fmt.Sprintf("protPort=%s maxSize=%d threadType=%d statDesc=%s lowLatency=%t fastAck=%t",
-               params.protPort, params.maxSize, params.threadType, params.statDesc, params.lowLatency, params.fastAck)
+       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(params.protPort)
-       m := C.int(params.maxSize)
-       c := C.int(params.threadType)
+       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 {
@@ -156,30 +153,37 @@ func NewRMRClientWithParams(params *RMRClientParams) *RMRClient {
 
        Logger.Info("new rmrClient with parameters: %s", params.String())
 
-       if params.lowLatency {
+       if params.RmrData.LowLatency {
                C.rmr_set_low_latency(ctx)
        }
-       if params.fastAck {
+       if params.RmrData.FastAck {
                C.rmr_set_fack(ctx)
        }
 
        return &RMRClient{
-               protPort:  params.protPort,
-               context:   ctx,
-               consumers: make([]MessageConsumer, 0),
-               stat:      Metric.RegisterCounterGroup(RMRCounterOpts, params.statDesc),
+               context:           ctx,
+               consumers:         make([]MessageConsumer, 0),
+               stat:              Metric.RegisterCounterGroup(RMRCounterOpts, params.StatDesc),
+               maxRetryOnFailure: params.RmrData.MaxRetryOnFailure,
        }
 }
 
 func NewRMRClient() *RMRClient {
+       p := GetPortData("rmr-data")
+       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{
-                       protPort:   viper.GetString("rmr.protPort"),
-                       maxSize:    viper.GetInt("rmr.maxSize"),
-                       threadType: viper.GetInt("rmr.threadType"),
-                       statDesc:   "RMR",
-                       lowLatency: viper.GetBool("rmr.lowLatency"),
-                       fastAck:    viper.GetBool("rmr.fastAck"),
+                       RmrData:  p,
+                       StatDesc: "RMR",
                })
 }
 
@@ -409,8 +413,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()
@@ -432,12 +435,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)
@@ -451,15 +453,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) {
@@ -570,11 +578,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
 }