"time"
logger "github.com/sirupsen/logrus"
+
+ "git.fd.io/govpp.git/api"
)
+var ReplyChannelTimeout = time.Millisecond * 100
+
var (
ErrNotConnected = errors.New("not connected to VPP, ignoring the request")
ErrProbeTimeout = errors.New("probe reply not received within timeout period")
return
}
if err := c.processRequest(ch, req); err != nil {
- sendReplyError(ch, req, err)
+ sendReply(ch, &vppReply{
+ seqNum: req.seqNum,
+ err: fmt.Errorf("unable to process request: %w", err),
+ })
}
}
}
}
+// processRequest processes a single request received on the request channel.
+func (c *Connection) sendMessage(context uint32, msg api.Message) error {
+ // check whether we are connected to VPP
+ if atomic.LoadUint32(&c.vppConnected) == 0 {
+ return ErrNotConnected
+ }
+
+ /*log := log.WithFields(logger.Fields{
+ "context": context,
+ "msg_name": msg.GetMessageName(),
+ "msg_crc": msg.GetCrcString(),
+ })*/
+
+ // retrieve message ID
+ msgID, err := c.GetMessageID(msg)
+ if err != nil {
+ //log.WithError(err).Debugf("unable to retrieve message ID: %#v", msg)
+ return err
+ }
+
+ //log = log.WithField("msg_id", msgID)
+
+ // encode the message
+ data, err := c.codec.EncodeMsg(msg, msgID)
+ if err != nil {
+ log.WithError(err).Debugf("unable to encode message: %#v", msg)
+ return err
+ }
+
+ //log = log.WithField("msg_length", len(data))
+
+ if log.Level >= logger.DebugLevel {
+ log.Debugf("--> SEND: MSG %T %+v", msg, msg)
+ }
+
+ // send message to VPP
+ err = c.vppClient.SendMsg(context, data)
+ if err != nil {
+ log.WithError(err).Debugf("unable to send message: %#v", msg)
+ return err
+ }
+
+ return nil
+}
+
// processRequest processes a single request received on the request channel.
func (c *Connection) processRequest(ch *Channel, req *vppRequest) error {
// check whether we are connected to VPP
if atomic.LoadUint32(&c.vppConnected) == 0 {
err := ErrNotConnected
- log.Errorf("processing request failed: %v", err)
+ log.WithFields(logger.Fields{
+ "channel": ch.id,
+ "seq_num": req.seqNum,
+ "msg_name": req.msg.GetMessageName(),
+ "msg_crc": req.msg.GetCrcString(),
+ "error": err,
+ }).Warnf("Unable to process request")
return err
}
msgID, err := c.GetMessageID(req.msg)
if err != nil {
log.WithFields(logger.Fields{
+ "channel": ch.id,
"msg_name": req.msg.GetMessageName(),
"msg_crc": req.msg.GetCrcString(),
"seq_num": req.seqNum,
"error": err,
- }).Errorf("failed to retrieve message ID")
- return fmt.Errorf("unable to retrieve message ID: %v", err)
+ }).Warnf("Unable to retrieve message ID")
+ return err
}
// encode the message into binary
"channel": ch.id,
"msg_id": msgID,
"msg_name": req.msg.GetMessageName(),
+ "msg_crc": req.msg.GetCrcString(),
"seq_num": req.seqNum,
"error": err,
- }).Errorf("failed to encode message: %#v", req.msg)
- return fmt.Errorf("unable to encode the message: %v", err)
+ }).Warnf("Unable to encode message: %T %+v", req.msg, req.msg)
+ return err
}
context := packRequestContext(ch.id, req.multi, req.seqNum)
- if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
+ if log.Level >= logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
log.WithFields(logger.Fields{
"channel": ch.id,
- "context": context,
- "is_multi": req.multi,
"msg_id": msgID,
"msg_name": req.msg.GetMessageName(),
- "msg_size": len(data),
+ "msg_crc": req.msg.GetCrcString(),
"seq_num": req.seqNum,
- }).Debug(" -> Sending a message to VPP.")
+ "is_multi": req.multi,
+ "context": context,
+ "data_len": len(data),
+ }).Debugf("--> SEND MSG: %T %+v", req.msg, req.msg)
}
// send the request to VPP
err = c.vppClient.SendMsg(context, data)
if err != nil {
- err = fmt.Errorf("unable to send the message: %v", err)
log.WithFields(logger.Fields{
- "context": context,
- "msg_id": msgID,
- "seq_num": req.seqNum,
- }).Error(err)
+ "channel": ch.id,
+ "msg_id": msgID,
+ "msg_name": req.msg.GetMessageName(),
+ "msg_crc": req.msg.GetCrcString(),
+ "seq_num": req.seqNum,
+ "is_multi": req.multi,
+ "context": context,
+ "data_len": len(data),
+ "error": err,
+ }).Warnf("Unable to send message")
return err
}
if req.multi {
// send a control ping to determine end of the multipart response
- pingData, _ := c.codec.EncodeMsg(msgControlPing, c.pingReqID)
+ pingData, _ := c.codec.EncodeMsg(c.msgControlPing, c.pingReqID)
- log.WithFields(logger.Fields{
- "channel": ch.id,
- "context": context,
- "msg_id": c.pingReqID,
- "msg_size": len(pingData),
- "seq_num": req.seqNum,
- }).Debug(" -> Sending a control ping to VPP.")
+ if log.Level >= logger.DebugLevel {
+ log.WithFields(logger.Fields{
+ "channel": ch.id,
+ "msg_id": c.pingReqID,
+ "msg_name": c.msgControlPing.GetMessageName(),
+ "msg_crc": c.msgControlPing.GetCrcString(),
+ "seq_num": req.seqNum,
+ "context": context,
+ "data_len": len(pingData),
+ }).Debugf(" -> SEND MSG: %T", c.msgControlPing)
+ }
if err := c.vppClient.SendMsg(context, pingData); err != nil {
log.WithFields(logger.Fields{
"context": context,
- "msg_id": msgID,
"seq_num": req.seqNum,
- }).Warnf("unable to send control ping: %v", err)
+ "error": err,
+ }).Warnf("unable to send control ping")
}
}
// msgCallback is called whenever any binary API message comes from VPP.
func (c *Connection) msgCallback(msgID uint16, data []byte) {
- connLock.RLock()
- defer connLock.RUnlock()
-
if c == nil {
- log.Warn("Already disconnected, ignoring the message.")
+ log.WithField(
+ "msg_id", msgID,
+ ).Warn("Connection already disconnected, ignoring the message.")
return
}
- msg, ok := c.msgMap[msgID]
- if !ok {
- log.Warnf("Unknown message received, ID: %d", msgID)
+ msgType, name, crc, err := c.getMessageDataByID(msgID)
+ if err != nil {
+ log.Warnln(err)
return
}
// - replies that don't have context as first field (comes as zero)
// - events that don't have context at all (comes as non zero)
//
- context, err := c.codec.DecodeMsgContext(data, msg)
+ context, err := c.codec.DecodeMsgContext(data, msgType)
if err != nil {
- log.Errorf("decoding context failed: %v", err)
+ log.WithField("msg_id", msgID).Warnf("Unable to decode message context: %v", err)
+ return
}
chanID, isMulti, seqNum := unpackRequestContext(context)
+
if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
log.WithFields(logger.Fields{
"context": context,
"msg_id": msgID,
- "msg_name": msg.GetMessageName(),
"msg_size": len(data),
"channel": chanID,
"is_multi": isMulti,
"seq_num": seqNum,
- }).Debug(" <- Received a message from VPP.")
+ "msg_crc": crc,
+ }).Debugf("<-- govpp RECEIVE: %s", name)
}
if context == 0 || c.isNotificationMessage(msgID) {
// sendReply sends the reply into the go channel, if it cannot be completed without blocking, otherwise
// it logs the error and do not send the message.
func sendReply(ch *Channel, reply *vppReply) {
+ // first try to avoid creating timer
+ select {
+ case ch.replyChan <- reply:
+ return // reply sent ok
+ default:
+ // reply channel full
+ }
+ if ch.receiveReplyTimeout == 0 {
+ log.WithFields(logger.Fields{
+ "channel": ch.id,
+ "msg_id": reply.msgID,
+ "seq_num": reply.seqNum,
+ "err": reply.err,
+ }).Warn("Reply channel full, dropping reply.")
+ return
+ }
select {
case ch.replyChan <- reply:
- // reply sent successfully
- case <-time.After(time.Millisecond * 100):
+ return // reply sent ok
+ case <-time.After(ch.receiveReplyTimeout):
// receiver still not ready
log.WithFields(logger.Fields{
- "channel": ch,
+ "channel": ch.id,
"msg_id": reply.msgID,
"seq_num": reply.seqNum,
- }).Warn("Unable to send the reply, reciever end not ready.")
+ "err": reply.err,
+ }).Warnf("Unable to send reply (reciever end not ready in %v).", ch.receiveReplyTimeout)
}
}
-func sendReplyError(ch *Channel, req *vppRequest, err error) {
- sendReply(ch, &vppReply{seqNum: req.seqNum, err: err})
-}
-
// isNotificationMessage returns true if someone has subscribed to provided message ID.
func (c *Connection) isNotificationMessage(msgID uint16) bool {
c.subscriptionsLock.RLock()
"msg_name": sub.event.GetMessageName(),
"msg_id": msgID,
"msg_size": len(data),
- }).Errorf("Unable to decode the notification message: %v", err)
+ "error": err,
+ }).Warnf("Unable to decode the notification message")
continue
}
}
return 1
}
+
+// Returns message data based on the message ID not depending on the message path
+func (c *Connection) getMessageDataByID(msgID uint16) (typ api.MessageType, name, crc string, err error) {
+ for _, msgs := range c.msgMapByPath {
+ if msg, ok := msgs[msgID]; ok {
+ return msg.GetMessageType(), msg.GetMessageName(), msg.GetCrcString(), nil
+ }
+ }
+ return typ, name, crc, fmt.Errorf("unknown message received, ID: %d", msgID)
+}