Fixed incorrect message error in the stream API

The message package is passed to the stream object and
used to evaluate correct reply message type

Change-Id: I2c9844d6447d024af1693205efd5721e2f89f22d
Signed-off-by: Vladimir Lavor <vlavor@cisco.com>

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
 	}