summaryrefslogtreecommitdiffstats
path: root/core
diff options
context:
space:
mode:
Diffstat (limited to 'core')
-rw-r--r--core/channel.go7
-rw-r--r--core/connection.go84
-rw-r--r--core/request_handler.go22
-rw-r--r--core/stream.go12
4 files changed, 80 insertions, 45 deletions
diff --git a/core/channel.go b/core/channel.go
index 28d0710..fbb3e59 100644
--- a/core/channel.go
+++ b/core/channel.go
@@ -45,8 +45,10 @@ type MessageCodec interface {
type MessageIdentifier interface {
// GetMessageID returns message identifier of given API message.
GetMessageID(msg api.Message) (uint16, error)
+ // GetMessagePath returns path for the given message
+ GetMessagePath(msg api.Message) string
// LookupByID looks up message name and crc by ID
- LookupByID(msgID uint16) (api.Message, error)
+ LookupByID(path string, msgID uint16) (api.Message, error)
}
// vppRequest is a request that will be sent to VPP.
@@ -329,7 +331,8 @@ func (ch *Channel) processReply(reply *vppReply, expSeqNum uint16, msg api.Messa
if reply.msgID != expMsgID {
var msgNameCrc string
- if replyMsg, err := ch.msgIdentifier.LookupByID(reply.msgID); err != nil {
+ pkgPath := ch.msgIdentifier.GetMessagePath(msg)
+ if replyMsg, err := ch.msgIdentifier.LookupByID(pkgPath, reply.msgID); err != nil {
msgNameCrc = err.Error()
} else {
msgNameCrc = getMsgNameWithCrc(replyMsg)
diff --git a/core/connection.go b/core/connection.go
index 0f54f38..f3ff964 100644
--- a/core/connection.go
+++ b/core/connection.go
@@ -17,6 +17,7 @@ package core
import (
"errors"
"fmt"
+ "path"
"reflect"
"sync"
"sync/atomic"
@@ -103,9 +104,9 @@ type Connection struct {
connChan chan ConnectionEvent // connection status events are sent to this channel
- codec MessageCodec // message codec
- msgIDs map[string]uint16 // map of message IDs indexed by message name + CRC
- msgMap map[uint16]api.Message // map of messages indexed by message ID
+ codec MessageCodec // message codec
+ msgIDs map[string]uint16 // map of message IDs indexed by message name + CRC
+ msgMapByPath map[string]map[uint16]api.Message // map of messages indexed by message ID which are indexed by path
maxChannelID uint32 // maximum used channel ID (the real limit is 2^15, 32-bit is used for atomic operations)
channelsLock sync.RWMutex // lock for the channels map
@@ -139,7 +140,7 @@ func newConnection(binapi adapter.VppAPI, attempts int, interval time.Duration)
connChan: make(chan ConnectionEvent, NotificationChanBufSize),
codec: codec.DefaultCodec,
msgIDs: make(map[string]uint16),
- msgMap: make(map[uint16]api.Message),
+ msgMapByPath: make(map[string]map[uint16]api.Message),
channels: make(map[uint16]*Channel),
subscriptions: make(map[uint16][]*subscriptionCtx),
msgControlPing: msgControlPing,
@@ -400,69 +401,74 @@ func (c *Connection) GetMessageID(msg api.Message) (uint16, error) {
if c == nil {
return 0, errors.New("nil connection passed in")
}
-
- if msgID, ok := c.msgIDs[getMsgNameWithCrc(msg)]; ok {
- return msgID, nil
- }
-
+ pkgPath := c.GetMessagePath(msg)
msgID, err := c.vppClient.GetMsgID(msg.GetMessageName(), msg.GetCrcString())
if err != nil {
return 0, err
}
-
+ if pathMsgs, pathOk := c.msgMapByPath[pkgPath]; !pathOk {
+ c.msgMapByPath[pkgPath] = make(map[uint16]api.Message)
+ c.msgMapByPath[pkgPath][msgID] = msg
+ } else if _, msgOk := pathMsgs[msgID]; !msgOk {
+ c.msgMapByPath[pkgPath][msgID] = msg
+ }
+ if _, ok := c.msgIDs[getMsgNameWithCrc(msg)]; ok {
+ return msgID, nil
+ }
c.msgIDs[getMsgNameWithCrc(msg)] = msgID
- c.msgMap[msgID] = msg
-
return msgID, nil
}
// LookupByID looks up message name and crc by ID.
-func (c *Connection) LookupByID(msgID uint16) (api.Message, error) {
+func (c *Connection) LookupByID(path string, msgID uint16) (api.Message, error) {
if c == nil {
return nil, errors.New("nil connection passed in")
}
-
- if msg, ok := c.msgMap[msgID]; ok {
+ if msg, ok := c.msgMapByPath[path][msgID]; ok {
return msg, nil
}
+ return nil, fmt.Errorf("unknown message ID %d for path '%s'", msgID, path)
+}
- return nil, fmt.Errorf("unknown message ID: %d", msgID)
+// GetMessagePath returns path for the given message
+func (c *Connection) GetMessagePath(msg api.Message) string {
+ return path.Dir(reflect.TypeOf(msg).Elem().PkgPath())
}
// retrieveMessageIDs retrieves IDs for all registered messages and stores them in map
func (c *Connection) retrieveMessageIDs() (err error) {
t := time.Now()
- msgs := api.GetRegisteredMessages()
+ msgsByPath := api.GetRegisteredMessages()
var n int
- for name, msg := range msgs {
- typ := reflect.TypeOf(msg).Elem()
- path := fmt.Sprintf("%s.%s", typ.PkgPath(), typ.Name())
+ for pkgPath, msgs := range msgsByPath {
+ for _, msg := range msgs {
+ msgID, err := c.GetMessageID(msg)
+ if err != nil {
+ if debugMsgIDs {
+ log.Debugf("retrieving message ID for %s.%s failed: %v",
+ pkgPath, msg.GetMessageName(), err)
+ }
+ continue
+ }
+ n++
+
+ if c.pingReqID == 0 && msg.GetMessageName() == c.msgControlPing.GetMessageName() {
+ c.pingReqID = msgID
+ c.msgControlPing = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
+ } else if c.pingReplyID == 0 && msg.GetMessageName() == c.msgControlPingReply.GetMessageName() {
+ c.pingReplyID = msgID
+ c.msgControlPingReply = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
+ }
- msgID, err := c.GetMessageID(msg)
- if err != nil {
if debugMsgIDs {
- log.Debugf("retrieving message ID for %s failed: %v", path, err)
+ log.Debugf("message %q (%s) has ID: %d", msg.GetMessageName(), getMsgNameWithCrc(msg), msgID)
}
- continue
- }
- n++
-
- if c.pingReqID == 0 && msg.GetMessageName() == c.msgControlPing.GetMessageName() {
- c.pingReqID = msgID
- c.msgControlPing = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
- } else if c.pingReplyID == 0 && msg.GetMessageName() == c.msgControlPingReply.GetMessageName() {
- c.pingReplyID = msgID
- c.msgControlPingReply = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
- }
-
- if debugMsgIDs {
- log.Debugf("message %q (%s) has ID: %d", name, getMsgNameWithCrc(msg), msgID)
}
+ log.WithField("took", time.Since(t)).
+ Debugf("retrieved IDs for %d messages (registered %d) from path %s", n, len(msgs), pkgPath)
}
- log.WithField("took", time.Since(t)).
- Debugf("retrieved IDs for %d messages (registered %d)", n, len(msgs))
return nil
}
diff --git a/core/request_handler.go b/core/request_handler.go
index fc704cb..f9d972a 100644
--- a/core/request_handler.go
+++ b/core/request_handler.go
@@ -210,9 +210,9 @@ func (c *Connection) msgCallback(msgID uint16, data []byte) {
return
}
- msg, ok := c.msgMap[msgID]
- if !ok {
- log.Warnf("Unknown message received, ID: %d", msgID)
+ msg, err := c.getMessageByID(msgID)
+ if err != nil {
+ log.Warnln(err)
return
}
@@ -419,3 +419,19 @@ func compareSeqNumbers(seqNum1, seqNum2 uint16) int {
}
return 1
}
+
+// Returns first message from any package where the message ID matches
+// Note: the msg is further used only for its MessageType which is not
+// affected by the message's package
+func (c *Connection) getMessageByID(msgID uint16) (msg api.Message, err error) {
+ var ok bool
+ for _, msgs := range c.msgMapByPath {
+ if msg, ok = msgs[msgID]; ok {
+ break
+ }
+ }
+ if !ok {
+ return nil, fmt.Errorf("unknown message received, ID: %d", msgID)
+ }
+ return msg, nil
+}
diff --git a/core/stream.go b/core/stream.go
index abe9d55..3d417f1 100644
--- a/core/stream.go
+++ b/core/stream.go
@@ -19,6 +19,7 @@ import (
"errors"
"fmt"
"reflect"
+ "sync"
"sync/atomic"
"time"
@@ -34,6 +35,9 @@ type Stream struct {
requestSize int
replySize int
replyTimeout time.Duration
+ // per-request context
+ pkgPath string
+ sync.Mutex
}
func (c *Connection) NewStream(ctx context.Context, options ...api.StreamOption) (api.Stream, error) {
@@ -109,6 +113,9 @@ func (s *Stream) SendMsg(msg api.Message) error {
if err := s.conn.processRequest(s.channel, req); err != nil {
return err
}
+ s.Lock()
+ s.pkgPath = s.conn.GetMessagePath(msg)
+ s.Unlock()
return nil
}
@@ -118,7 +125,10 @@ func (s *Stream) RecvMsg() (api.Message, error) {
return nil, err
}
// resolve message type
- msg, err := s.channel.msgIdentifier.LookupByID(reply.msgID)
+ s.Lock()
+ path := s.pkgPath
+ s.Unlock()
+ msg, err := s.channel.msgIdentifier.LookupByID(path, reply.msgID)
if err != nil {
return nil, err
}