Code Review
/
ric-plt
/
xapp-frame.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
review
|
tree
raw
|
inline
| side by side
GetPortData() returns policies
[ric-plt/xapp-frame.git]
/
pkg
/
xapp
/
rmr.go
diff --git
a/pkg/xapp/rmr.go
b/pkg/xapp/rmr.go
index
530f223
..
9407024
100755
(executable)
--- a/
pkg/xapp/rmr.go
+++ b/
pkg/xapp/rmr.go
@@
-67,7
+67,6
@@
import (
"bytes"
"crypto/md5"
"fmt"
"bytes"
"crypto/md5"
"fmt"
- "github.com/spf13/viper"
"strings"
"time"
"unsafe"
"strings"
"time"
"unsafe"
@@
-128,26
+127,23
@@
func (params *RMRParams) String() string {
//
//-----------------------------------------------------------------------------
type RMRClientParams struct {
//
//-----------------------------------------------------------------------------
type RMRClientParams struct {
- protPort string
- maxSize int
- threadType int
- statDesc string
- lowLatency bool
- fastAck bool
+ StatDesc string
+ RmrData PortData
}
func (params *RMRClientParams) String() string {
}
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 {
}
//-----------------------------------------------------------------------------
//
//-----------------------------------------------------------------------------
func NewRMRClientWithParams(params *RMRClientParams) *RMRClient {
- p := C.CString(
params.protPort
)
- m := C.int(params.
m
axSize)
- c := C.int(params.
t
hreadType)
+ p := C.CString(
fmt.Sprintf("%d", params.RmrData.Port)
)
+ m := C.int(params.
RmrData.M
axSize)
+ c := C.int(params.
RmrData.T
hreadType)
defer C.free(unsafe.Pointer(p))
ctx := C.rmr_init(p, m, c)
if ctx == nil {
defer C.free(unsafe.Pointer(p))
ctx := C.rmr_init(p, m, c)
if ctx == nil {
@@
-156,30
+152,27
@@
func NewRMRClientWithParams(params *RMRClientParams) *RMRClient {
Logger.Info("new rmrClient with parameters: %s", params.String())
Logger.Info("new rmrClient with parameters: %s", params.String())
- if params.
l
owLatency {
+ if params.
RmrData.L
owLatency {
C.rmr_set_low_latency(ctx)
}
C.rmr_set_low_latency(ctx)
}
- if params.
f
astAck {
+ if params.
RmrData.F
astAck {
C.rmr_set_fack(ctx)
}
return &RMRClient{
C.rmr_set_fack(ctx)
}
return &RMRClient{
-
protPort: params.protPort
,
- con
text: ctx
,
-
consumers: make([]MessageConsumer, 0
),
-
stat: Metric.RegisterCounterGroup(RMRCounterOpts, params.statDesc)
,
+
context: ctx
,
+ con
sumers: make([]MessageConsumer, 0)
,
+
stat: Metric.RegisterCounterGroup(RMRCounterOpts, params.StatDesc
),
+
maxRetryOnFailure: params.RmrData.MaxRetryOnFailure
,
}
}
func NewRMRClient() *RMRClient {
}
}
func NewRMRClient() *RMRClient {
+ p := GetPortData("rmr-data")
return NewRMRClientWithParams(
&RMRClientParams{
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
+402,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 (
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()
)
m.contextMux.Lock()
@@
-432,12
+424,11
@@
func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int {
}
// Just quick retry seems to help for K8s issue
}
// 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
.m
axRetryOnFailure && 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)
m.contextMux.Lock()
if whid != 0 {
currBuffer = C.rmr_wh_send_msg(m.context, C.rmr_whid_t(whid), txBuffer)
@@
-451,15
+442,21
@@
func (m *RMRClient) SendBuf(txBuffer *C.rmr_mbuf_t, isRts bool, whid int) int {
m.contextMux.Unlock()
}
m.contextMux.Unlock()
}
- if currBuffer
.state != C.RMR_OK
{
- counterName = "TransmitError"
+ if currBuffer
== nil
{
+ m.UpdateStatCounter("TransmitError")
m.LogMBufError("SendBuf failed", currBuffer)
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)
defer m.Free(currBuffer)
-
return int(currBuffer.state)
return int(currBuffer.state)
+
}
func (m *RMRClient) SendCallMsg(params *RMRParams) (int, string) {
}
func (m *RMRClient) SendCallMsg(params *RMRParams) (int, string) {
@@
-570,11
+567,10
@@
func (m *RMRClient) GetRicMessageName(id int) (s string) {
}
func (m *RMRClient) LogMBufError(text string, mbuf *C.rmr_mbuf_t) int {
}
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
}
}