1 // Copyright (c) 2017 Cisco and/or its affiliates.
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at:
7 // http://www.apache.org/licenses/LICENSE-2.0
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
22 logger "github.com/Sirupsen/logrus"
24 "git.fd.io/govpp.git/api"
25 "git.fd.io/govpp.git/core/bin_api/vpe"
28 // watchRequests watches for requests on the request API channel and forwards them as messages to VPP.
29 func (c *Connection) watchRequests(ch *api.Channel, chMeta *channelMetadata) {
32 case req, ok := <-ch.ReqChan:
33 // new request on the request channel
35 // after closing the request channel, release API channel and return
36 c.releaseAPIChannel(ch, chMeta)
39 c.processRequest(ch, chMeta, req)
41 case req := <-ch.NotifSubsChan:
42 // new request on the notification subscribe channel
43 c.processNotifSubscribeRequest(ch, req)
48 // processRequest processes a single request received on the request channel.
49 func (c *Connection) processRequest(ch *api.Channel, chMeta *channelMetadata, req *api.VppRequest) error {
50 // check whether we are connected to VPP
51 if atomic.LoadUint32(&c.connected) == 0 {
52 error := errors.New("not connected to VPP, ignoring the request")
54 sendReply(ch, &api.VppReply{Error: error})
58 // retrieve message ID
59 msgID, err := c.GetMessageID(req.Message)
61 error := fmt.Errorf("unable to retrieve message ID: %v", err)
62 log.WithFields(logger.Fields{
63 "msg_name": req.Message.GetMessageName(),
64 "msg_crc": req.Message.GetCrcString(),
66 sendReply(ch, &api.VppReply{Error: error})
70 // encode the message into binary
71 data, err := c.codec.EncodeMsg(req.Message, msgID)
73 error := fmt.Errorf("unable to encode the messge: %v", err)
74 log.WithFields(logger.Fields{
78 sendReply(ch, &api.VppReply{Error: error})
83 log.WithFields(logger.Fields{
86 "msg_size": len(data),
87 }).Debug("Sending a message to VPP.")
90 // expect multipart response
91 atomic.StoreUint32(&chMeta.multipart, 1)
94 // send the request to VPP
95 c.vpp.SendMsg(chMeta.id, data)
98 // send a control ping to determine end of the multipart response
99 ping := &vpe.ControlPing{}
100 pingData, _ := c.codec.EncodeMsg(ping, c.pingReqID)
102 log.WithFields(logger.Fields{
103 "context": chMeta.id,
104 "msg_id": c.pingReqID,
105 "msg_size": len(pingData),
106 }).Debug("Sending a control ping to VPP.")
108 c.vpp.SendMsg(chMeta.id, pingData)
114 // msgCallback is called whenever any binary API message comes from VPP.
115 func msgCallback(context uint32, msgID uint16, data []byte) {
117 defer connLock.RUnlock()
120 log.Warn("Already disconnected, ignoring the message.")
124 log.WithFields(logger.Fields{
127 "msg_size": len(data),
128 }).Debug("Received a message from VPP.")
130 if context == 0 || conn.isNotificationMessage(msgID) {
131 // process the message as a notification
132 conn.sendNotifications(msgID, data)
136 // match ch according to the context
137 conn.channelsLock.RLock()
138 ch, ok := conn.channels[context]
139 conn.channelsLock.RUnlock()
142 log.WithFields(logger.Fields{
145 }).Error("Context ID not known, ignoring the message.")
149 chMeta := ch.Metadata().(*channelMetadata)
150 lastReplyReceived := false
151 // if this is a control ping reply and multipart request is being processed, treat this as a last part of the reply
152 if msgID == conn.pingReplyID && atomic.CompareAndSwapUint32(&chMeta.multipart, 1, 0) {
153 lastReplyReceived = true
156 // send the data to the channel
157 sendReply(ch, &api.VppReply{
160 LastReplyReceived: lastReplyReceived,
164 // sendReply sends the reply into the go channel, if it cannot be completed without blocking, otherwise
165 // it logs the error and do not send the message.
166 func sendReply(ch *api.Channel, reply *api.VppReply) {
168 case ch.ReplyChan <- reply:
169 // reply sent successfully
171 // unable to write into the channel without blocking
172 log.WithFields(logger.Fields{
174 "msg_id": reply.MessageID,
175 }).Warn("Unable to send the reply, reciever end not ready.")
179 // GetMessageID returns message identifier of given API message.
180 func (c *Connection) GetMessageID(msg api.Message) (uint16, error) {
182 return 0, errors.New("nil connection passed in")
184 return c.messageNameToID(msg.GetMessageName(), msg.GetCrcString())
187 // messageNameToID returns message ID of a message identified by its name and CRC.
188 func (c *Connection) messageNameToID(msgName string, msgCrc string) (uint16, error) {
189 // try to get the ID from the map
191 id, ok := c.msgIDs[msgName+msgCrc]
192 c.msgIDsLock.RUnlock()
197 // get the ID using VPP API
198 id, err := c.vpp.GetMsgID(msgName, msgCrc)
200 error := fmt.Errorf("unable to retrieve message ID: %v", err)
201 log.WithFields(logger.Fields{
204 }).Errorf("unable to retrieve message ID: %v", err)
209 c.msgIDs[msgName+msgCrc] = id
210 c.msgIDsLock.Unlock()