From c09ee3241377aae2530a73d48c4e20641d76d0ee Mon Sep 17 00:00:00 2001 From: mhalaj1 Date: Thu, 26 Aug 2021 20:15:08 +0200 Subject: Refactoring and fixes * refactor creation of new channel * add missing closing of created streams * correct documentation regarding thread safety of stream Signed-off-by: mhalaj1 Change-Id: Ic601efff298fcbdecaafab83fa236253af69de21 --- core/channel.go | 26 ++++++++++++++++++-------- 1 file changed, 18 insertions(+), 8 deletions(-) (limited to 'core/channel.go') diff --git a/core/channel.go b/core/channel.go index 4cb5761..1086c36 100644 --- a/core/channel.go +++ b/core/channel.go @@ -19,6 +19,7 @@ import ( "fmt" "reflect" "strings" + "sync/atomic" "time" "github.com/sirupsen/logrus" @@ -109,17 +110,26 @@ type Channel struct { receiveReplyTimeout time.Duration // maximum time that we wait for receiver to consume reply } -func newChannel(id uint16, conn *Connection, codec MessageCodec, identifier MessageIdentifier, reqSize, replySize int) *Channel { - return &Channel{ - id: id, - conn: conn, - msgCodec: codec, - msgIdentifier: identifier, - reqChan: make(chan *vppRequest, reqSize), - replyChan: make(chan *vppReply, replySize), +func (c *Connection) newChannel(reqChanBufSize, replyChanBufSize int) *Channel { + // create new channel + chID := uint16(atomic.AddUint32(&c.maxChannelID, 1) & 0x7fff) + channel := &Channel{ + id: chID, + conn: c, + msgCodec: c.codec, + msgIdentifier: c, + reqChan: make(chan *vppRequest, reqChanBufSize), + replyChan: make(chan *vppReply, replyChanBufSize), replyTimeout: DefaultReplyTimeout, receiveReplyTimeout: ReplyChannelTimeout, } + + // store API channel within the client + c.channelsLock.Lock() + c.channels[chID] = channel + c.channelsLock.Unlock() + + return channel } func (ch *Channel) GetID() uint16 { -- cgit 1.2.3-korg