"bytes"
"crypto/md5"
"fmt"
- "github.com/spf13/viper"
"strings"
"time"
"unsafe"
+
+ "github.com/spf13/viper"
)
var RMRCounterOpts = []CounterOpts{
//
//-----------------------------------------------------------------------------
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 {
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("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{
- 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",
})
}
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 {
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()
}
// 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)
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) {
}
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
}