blob: afad2ac8fecfa871310b0453813812f162b26a3e [file] [log] [blame]
khenaidoobf6e7bb2018-08-14 22:27:29 -04001/*
2 * Copyright 2018-present Open Networking Foundation
3
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7
8 * http://www.apache.org/licenses/LICENSE-2.0
9
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
khenaidooabad44c2018-08-03 16:58:35 -040016package kafka
17
18import (
19 "context"
20 "errors"
21 "fmt"
khenaidooabad44c2018-08-03 16:58:35 -040022 "github.com/golang/protobuf/proto"
23 "github.com/golang/protobuf/ptypes"
24 "github.com/golang/protobuf/ptypes/any"
25 "github.com/google/uuid"
26 "github.com/opencord/voltha-go/common/log"
William Kurkiandaa6bb22019-03-07 12:26:28 -050027 ic "github.com/opencord/voltha-protos/go/inter_container"
khenaidooabad44c2018-08-03 16:58:35 -040028 "reflect"
khenaidoo19374072018-12-11 11:05:15 -050029 "strings"
khenaidooabad44c2018-08-03 16:58:35 -040030 "sync"
31 "time"
32)
33
34// Initialize the logger - gets the default until the main function setup the logger
35func init() {
khenaidooca301322019-01-09 23:06:32 -050036 log.AddPackage(log.JSON, log.DebugLevel, nil)
khenaidooabad44c2018-08-03 16:58:35 -040037}
38
39const (
khenaidoo43c82122018-11-22 18:38:28 -050040 DefaultMaxRetries = 3
khenaidoo6d055132019-02-12 16:51:19 -050041 DefaultRequestTimeout = 10000 // 10000 milliseconds - to handle a wider latency range
khenaidooabad44c2018-08-03 16:58:35 -040042)
43
khenaidoo297cd252019-02-07 22:10:23 -050044const (
45 TransactionKey = "transactionID"
khenaidoo54e0ddf2019-02-27 16:21:33 -050046 FromTopic = "fromTopic"
khenaidoo297cd252019-02-07 22:10:23 -050047)
48
khenaidoo43c82122018-11-22 18:38:28 -050049// requestHandlerChannel represents an interface associated with a channel. Whenever, an event is
50// obtained from that channel, this interface is invoked. This is used to handle
51// async requests into the Core via the kafka messaging bus
52type requestHandlerChannel struct {
53 requesthandlerInterface interface{}
khenaidoo79232702018-12-04 11:00:41 -050054 ch <-chan *ic.InterContainerMessage
khenaidooabad44c2018-08-03 16:58:35 -040055}
56
khenaidoo43c82122018-11-22 18:38:28 -050057// transactionChannel represents a combination of a topic and a channel onto which a response received
58// on the kafka bus will be sent to
59type transactionChannel struct {
60 topic *Topic
khenaidoo79232702018-12-04 11:00:41 -050061 ch chan *ic.InterContainerMessage
khenaidoo43c82122018-11-22 18:38:28 -050062}
63
64// InterContainerProxy represents the messaging proxy
65type InterContainerProxy struct {
66 kafkaHost string
67 kafkaPort int
68 DefaultTopic *Topic
69 defaultRequestHandlerInterface interface{}
khenaidoo79232702018-12-04 11:00:41 -050070 deviceDiscoveryTopic *Topic
khenaidoo43c82122018-11-22 18:38:28 -050071 kafkaClient Client
72 doneCh chan int
73
74 // This map is used to map a topic to an interface and channel. When a request is received
75 // on that channel (registered to the topic) then that interface is invoked.
76 topicToRequestHandlerChannelMap map[string]*requestHandlerChannel
77 lockTopicRequestHandlerChannelMap sync.RWMutex
78
79 // This map is used to map a channel to a response topic. This channel handles all responses on that
khenaidoo4c1a5bf2018-11-29 15:53:42 -050080 // channel for that topic and forward them to the appropriate consumers channel, using the
khenaidoo43c82122018-11-22 18:38:28 -050081 // transactionIdToChannelMap.
khenaidoo79232702018-12-04 11:00:41 -050082 topicToResponseChannelMap map[string]<-chan *ic.InterContainerMessage
khenaidoo43c82122018-11-22 18:38:28 -050083 lockTopicResponseChannelMap sync.RWMutex
84
khenaidoo4c1a5bf2018-11-29 15:53:42 -050085 // This map is used to map a transaction to a consumers channel. This is used whenever a request has been
khenaidoo43c82122018-11-22 18:38:28 -050086 // sent out and we are waiting for a response.
87 transactionIdToChannelMap map[string]*transactionChannel
khenaidooabad44c2018-08-03 16:58:35 -040088 lockTransactionIdToChannelMap sync.RWMutex
89}
90
khenaidoo43c82122018-11-22 18:38:28 -050091type InterContainerProxyOption func(*InterContainerProxy)
khenaidooabad44c2018-08-03 16:58:35 -040092
khenaidoo43c82122018-11-22 18:38:28 -050093func InterContainerHost(host string) InterContainerProxyOption {
94 return func(args *InterContainerProxy) {
95 args.kafkaHost = host
khenaidooabad44c2018-08-03 16:58:35 -040096 }
97}
98
khenaidoo43c82122018-11-22 18:38:28 -050099func InterContainerPort(port int) InterContainerProxyOption {
100 return func(args *InterContainerProxy) {
101 args.kafkaPort = port
khenaidooabad44c2018-08-03 16:58:35 -0400102 }
103}
104
khenaidoo43c82122018-11-22 18:38:28 -0500105func DefaultTopic(topic *Topic) InterContainerProxyOption {
106 return func(args *InterContainerProxy) {
khenaidooabad44c2018-08-03 16:58:35 -0400107 args.DefaultTopic = topic
108 }
109}
110
khenaidoo79232702018-12-04 11:00:41 -0500111func DeviceDiscoveryTopic(topic *Topic) InterContainerProxyOption {
112 return func(args *InterContainerProxy) {
113 args.deviceDiscoveryTopic = topic
114 }
115}
116
khenaidoo43c82122018-11-22 18:38:28 -0500117func RequestHandlerInterface(handler interface{}) InterContainerProxyOption {
118 return func(args *InterContainerProxy) {
119 args.defaultRequestHandlerInterface = handler
khenaidooabad44c2018-08-03 16:58:35 -0400120 }
121}
122
khenaidoo43c82122018-11-22 18:38:28 -0500123func MsgClient(client Client) InterContainerProxyOption {
124 return func(args *InterContainerProxy) {
125 args.kafkaClient = client
126 }
127}
128
129func NewInterContainerProxy(opts ...InterContainerProxyOption) (*InterContainerProxy, error) {
130 proxy := &InterContainerProxy{
131 kafkaHost: DefaultKafkaHost,
132 kafkaPort: DefaultKafkaPort,
khenaidooabad44c2018-08-03 16:58:35 -0400133 }
134
135 for _, option := range opts {
136 option(proxy)
137 }
138
139 // Create the locks for all the maps
khenaidoo43c82122018-11-22 18:38:28 -0500140 proxy.lockTopicRequestHandlerChannelMap = sync.RWMutex{}
khenaidooabad44c2018-08-03 16:58:35 -0400141 proxy.lockTransactionIdToChannelMap = sync.RWMutex{}
khenaidoo43c82122018-11-22 18:38:28 -0500142 proxy.lockTopicResponseChannelMap = sync.RWMutex{}
khenaidooabad44c2018-08-03 16:58:35 -0400143
144 return proxy, nil
145}
146
khenaidoo43c82122018-11-22 18:38:28 -0500147func (kp *InterContainerProxy) Start() error {
khenaidooabad44c2018-08-03 16:58:35 -0400148 log.Info("Starting-Proxy")
149
khenaidoo43c82122018-11-22 18:38:28 -0500150 // Kafka MsgClient should already have been created. If not, output fatal error
151 if kp.kafkaClient == nil {
152 log.Fatal("kafka-client-not-set")
153 }
154
khenaidooabad44c2018-08-03 16:58:35 -0400155 // Create the Done channel
156 kp.doneCh = make(chan int, 1)
157
khenaidoo43c82122018-11-22 18:38:28 -0500158 // Start the kafka client
khenaidoo4c1a5bf2018-11-29 15:53:42 -0500159 if err := kp.kafkaClient.Start(); err != nil {
khenaidoo43c82122018-11-22 18:38:28 -0500160 log.Errorw("Cannot-create-kafka-proxy", log.Fields{"error": err})
khenaidooabad44c2018-08-03 16:58:35 -0400161 return err
162 }
163
khenaidoo43c82122018-11-22 18:38:28 -0500164 // Create the topic to response channel map
khenaidoo79232702018-12-04 11:00:41 -0500165 kp.topicToResponseChannelMap = make(map[string]<-chan *ic.InterContainerMessage)
khenaidoo43c82122018-11-22 18:38:28 -0500166 //
khenaidooabad44c2018-08-03 16:58:35 -0400167 // Create the transactionId to Channel Map
khenaidoo43c82122018-11-22 18:38:28 -0500168 kp.transactionIdToChannelMap = make(map[string]*transactionChannel)
169
170 // Create the topic to request channel map
171 kp.topicToRequestHandlerChannelMap = make(map[string]*requestHandlerChannel)
khenaidooabad44c2018-08-03 16:58:35 -0400172
173 return nil
174}
175
khenaidoo43c82122018-11-22 18:38:28 -0500176func (kp *InterContainerProxy) Stop() {
177 log.Info("stopping-intercontainer-proxy")
178 kp.doneCh <- 1
179 // TODO : Perform cleanup
khenaidooca301322019-01-09 23:06:32 -0500180 kp.kafkaClient.Stop()
khenaidoo43c82122018-11-22 18:38:28 -0500181 //kp.deleteAllTopicRequestHandlerChannelMap()
182 //kp.deleteAllTopicResponseChannelMap()
183 //kp.deleteAllTransactionIdToChannelMap()
khenaidooabad44c2018-08-03 16:58:35 -0400184}
185
khenaidoo79232702018-12-04 11:00:41 -0500186// DeviceDiscovered publish the discovered device onto the kafka messaging bus
khenaidoo19374072018-12-11 11:05:15 -0500187func (kp *InterContainerProxy) DeviceDiscovered(deviceId string, deviceType string, parentId string, publisher string) error {
khenaidoo79232702018-12-04 11:00:41 -0500188 log.Debugw("sending-device-discovery-msg", log.Fields{"deviceId": deviceId})
189 // Simple validation
190 if deviceId == "" || deviceType == "" {
191 log.Errorw("invalid-parameters", log.Fields{"id": deviceId, "type": deviceType})
192 return errors.New("invalid-parameters")
193 }
194 // Create the device discovery message
195 header := &ic.Header{
196 Id: uuid.New().String(),
197 Type: ic.MessageType_DEVICE_DISCOVERED,
198 FromTopic: kp.DefaultTopic.Name,
199 ToTopic: kp.deviceDiscoveryTopic.Name,
200 Timestamp: time.Now().UnixNano(),
201 }
202 body := &ic.DeviceDiscovered{
203 Id: deviceId,
204 DeviceType: deviceType,
205 ParentId: parentId,
khenaidood2b6df92018-12-13 16:37:20 -0500206 Publisher: publisher,
khenaidoo79232702018-12-04 11:00:41 -0500207 }
208
209 var marshalledData *any.Any
210 var err error
211 if marshalledData, err = ptypes.MarshalAny(body); err != nil {
212 log.Errorw("cannot-marshal-request", log.Fields{"error": err})
213 return err
214 }
215 msg := &ic.InterContainerMessage{
216 Header: header,
217 Body: marshalledData,
218 }
219
220 // Send the message
221 if err := kp.kafkaClient.Send(msg, kp.deviceDiscoveryTopic); err != nil {
222 log.Errorw("cannot-send-device-discovery-message", log.Fields{"error": err})
223 return err
224 }
225 return nil
226}
227
khenaidoo43c82122018-11-22 18:38:28 -0500228// InvokeRPC is used to send a request to a given topic
229func (kp *InterContainerProxy) InvokeRPC(ctx context.Context, rpc string, toTopic *Topic, replyToTopic *Topic,
khenaidoobdcb8e02019-03-06 16:28:56 -0500230 waitForResponse bool, key string, kvArgs ...*KVArg) (bool, *any.Any) {
khenaidoo43c82122018-11-22 18:38:28 -0500231
232 // If a replyToTopic is provided then we use it, otherwise just use the default toTopic. The replyToTopic is
233 // typically the device ID.
234 responseTopic := replyToTopic
235 if responseTopic == nil {
236 responseTopic = kp.DefaultTopic
237 }
238
khenaidooabad44c2018-08-03 16:58:35 -0400239 // Encode the request
khenaidoobdcb8e02019-03-06 16:28:56 -0500240 protoRequest, err := encodeRequest(rpc, toTopic, responseTopic, key, kvArgs...)
khenaidooabad44c2018-08-03 16:58:35 -0400241 if err != nil {
242 log.Warnw("cannot-format-request", log.Fields{"rpc": rpc, "error": err})
243 return false, nil
244 }
245
246 // Subscribe for response, if needed, before sending request
khenaidoo79232702018-12-04 11:00:41 -0500247 var ch <-chan *ic.InterContainerMessage
khenaidooabad44c2018-08-03 16:58:35 -0400248 if waitForResponse {
249 var err error
khenaidoo43c82122018-11-22 18:38:28 -0500250 if ch, err = kp.subscribeForResponse(*responseTopic, protoRequest.Header.Id); err != nil {
251 log.Errorw("failed-to-subscribe-for-response", log.Fields{"error": err, "toTopic": toTopic.Name})
khenaidooabad44c2018-08-03 16:58:35 -0400252 }
253 }
254
khenaidoo43c82122018-11-22 18:38:28 -0500255 // Send request - if the topic is formatted with a device Id then we will send the request using a
256 // specific key, hence ensuring a single partition is used to publish the request. This ensures that the
257 // subscriber on that topic will receive the request in the order it was sent. The key used is the deviceId.
khenaidoobdcb8e02019-03-06 16:28:56 -0500258 //key := GetDeviceIdFromTopic(*toTopic)
khenaidoo1ce37ad2019-03-24 22:07:24 -0400259 log.Debugw("sending-msg", log.Fields{"rpc": rpc, "toTopic": toTopic, "replyTopic": responseTopic, "key": key, "xId": protoRequest.Header.Id})
khenaidoo8f474192019-04-03 17:20:44 -0400260 go kp.kafkaClient.Send(protoRequest, toTopic, key)
khenaidooabad44c2018-08-03 16:58:35 -0400261
262 if waitForResponse {
khenaidoob9203542018-09-17 22:56:37 -0400263 // Create a child context based on the parent context, if any
khenaidooabad44c2018-08-03 16:58:35 -0400264 var cancel context.CancelFunc
khenaidoob9203542018-09-17 22:56:37 -0400265 childCtx := context.Background()
khenaidooabad44c2018-08-03 16:58:35 -0400266 if ctx == nil {
267 ctx, cancel = context.WithTimeout(context.Background(), DefaultRequestTimeout*time.Millisecond)
khenaidoob9203542018-09-17 22:56:37 -0400268 } else {
269 childCtx, cancel = context.WithTimeout(ctx, DefaultRequestTimeout*time.Millisecond)
khenaidooabad44c2018-08-03 16:58:35 -0400270 }
khenaidoob9203542018-09-17 22:56:37 -0400271 defer cancel()
khenaidooabad44c2018-08-03 16:58:35 -0400272
273 // Wait for response as well as timeout or cancellation
274 // Remove the subscription for a response on return
275 defer kp.unSubscribeForResponse(protoRequest.Header.Id)
276 select {
khenaidoo3dfc8bc2019-01-10 16:48:25 -0500277 case msg, ok := <-ch:
278 if !ok {
279 log.Warnw("channel-closed", log.Fields{"rpc": rpc, "replyTopic": replyToTopic.Name})
280 protoError := &ic.Error{Reason: "channel-closed"}
281 var marshalledArg *any.Any
282 if marshalledArg, err = ptypes.MarshalAny(protoError); err != nil {
283 return false, nil // Should never happen
284 }
285 return false, marshalledArg
286 }
khenaidoo43c82122018-11-22 18:38:28 -0500287 log.Debugw("received-response", log.Fields{"rpc": rpc, "msgHeader": msg.Header})
khenaidoo79232702018-12-04 11:00:41 -0500288 var responseBody *ic.InterContainerResponseBody
khenaidooabad44c2018-08-03 16:58:35 -0400289 var err error
290 if responseBody, err = decodeResponse(msg); err != nil {
291 log.Errorw("decode-response-error", log.Fields{"error": err})
292 }
293 return responseBody.Success, responseBody.Result
294 case <-ctx.Done():
295 log.Debugw("context-cancelled", log.Fields{"rpc": rpc, "ctx": ctx.Err()})
296 // pack the error as proto any type
khenaidoo79232702018-12-04 11:00:41 -0500297 protoError := &ic.Error{Reason: ctx.Err().Error()}
khenaidooabad44c2018-08-03 16:58:35 -0400298 var marshalledArg *any.Any
299 if marshalledArg, err = ptypes.MarshalAny(protoError); err != nil {
300 return false, nil // Should never happen
301 }
302 return false, marshalledArg
khenaidoob9203542018-09-17 22:56:37 -0400303 case <-childCtx.Done():
304 log.Debugw("context-cancelled", log.Fields{"rpc": rpc, "ctx": childCtx.Err()})
305 // pack the error as proto any type
khenaidoo79232702018-12-04 11:00:41 -0500306 protoError := &ic.Error{Reason: childCtx.Err().Error()}
khenaidoob9203542018-09-17 22:56:37 -0400307 var marshalledArg *any.Any
308 if marshalledArg, err = ptypes.MarshalAny(protoError); err != nil {
309 return false, nil // Should never happen
310 }
311 return false, marshalledArg
khenaidooabad44c2018-08-03 16:58:35 -0400312 case <-kp.doneCh:
khenaidoo43c82122018-11-22 18:38:28 -0500313 log.Infow("received-exit-signal", log.Fields{"toTopic": toTopic.Name, "rpc": rpc})
khenaidooabad44c2018-08-03 16:58:35 -0400314 return true, nil
315 }
316 }
317 return true, nil
318}
319
khenaidoo43c82122018-11-22 18:38:28 -0500320// SubscribeWithRequestHandlerInterface allows a caller to assign a target object to be invoked automatically
khenaidooabad44c2018-08-03 16:58:35 -0400321// when a message is received on a given topic
khenaidoo43c82122018-11-22 18:38:28 -0500322func (kp *InterContainerProxy) SubscribeWithRequestHandlerInterface(topic Topic, handler interface{}) error {
khenaidooabad44c2018-08-03 16:58:35 -0400323
324 // Subscribe to receive messages for that topic
khenaidoo79232702018-12-04 11:00:41 -0500325 var ch <-chan *ic.InterContainerMessage
khenaidooabad44c2018-08-03 16:58:35 -0400326 var err error
khenaidoo4c1a5bf2018-11-29 15:53:42 -0500327 if ch, err = kp.kafkaClient.Subscribe(&topic); err != nil {
khenaidoo43c82122018-11-22 18:38:28 -0500328 //if ch, err = kp.Subscribe(topic); err != nil {
khenaidooabad44c2018-08-03 16:58:35 -0400329 log.Errorw("failed-to-subscribe", log.Fields{"error": err, "topic": topic.Name})
330 }
khenaidoo43c82122018-11-22 18:38:28 -0500331
332 kp.defaultRequestHandlerInterface = handler
333 kp.addToTopicRequestHandlerChannelMap(topic.Name, &requestHandlerChannel{requesthandlerInterface: handler, ch: ch})
khenaidooabad44c2018-08-03 16:58:35 -0400334 // Launch a go routine to receive and process kafka messages
khenaidoo54e0ddf2019-02-27 16:21:33 -0500335 go kp.waitForMessages(ch, topic, handler)
khenaidooabad44c2018-08-03 16:58:35 -0400336
337 return nil
338}
339
khenaidoo43c82122018-11-22 18:38:28 -0500340// SubscribeWithDefaultRequestHandler allows a caller to add a topic to an existing target object to be invoked automatically
341// when a message is received on a given topic. So far there is only 1 target registered per microservice
khenaidoo731697e2019-01-29 16:03:29 -0500342func (kp *InterContainerProxy) SubscribeWithDefaultRequestHandler(topic Topic, initialOffset int64) error {
khenaidoo43c82122018-11-22 18:38:28 -0500343 // Subscribe to receive messages for that topic
khenaidoo79232702018-12-04 11:00:41 -0500344 var ch <-chan *ic.InterContainerMessage
khenaidoo43c82122018-11-22 18:38:28 -0500345 var err error
khenaidoo54e0ddf2019-02-27 16:21:33 -0500346 if ch, err = kp.kafkaClient.Subscribe(&topic, &KVArg{Key: Offset, Value: initialOffset}); err != nil {
khenaidoo43c82122018-11-22 18:38:28 -0500347 log.Errorw("failed-to-subscribe", log.Fields{"error": err, "topic": topic.Name})
khenaidooca301322019-01-09 23:06:32 -0500348 return err
khenaidoo43c82122018-11-22 18:38:28 -0500349 }
350 kp.addToTopicRequestHandlerChannelMap(topic.Name, &requestHandlerChannel{requesthandlerInterface: kp.defaultRequestHandlerInterface, ch: ch})
351
352 // Launch a go routine to receive and process kafka messages
khenaidoo54e0ddf2019-02-27 16:21:33 -0500353 go kp.waitForMessages(ch, topic, kp.defaultRequestHandlerInterface)
khenaidoo43c82122018-11-22 18:38:28 -0500354
khenaidooabad44c2018-08-03 16:58:35 -0400355 return nil
356}
357
khenaidoo43c82122018-11-22 18:38:28 -0500358func (kp *InterContainerProxy) UnSubscribeFromRequestHandler(topic Topic) error {
359 return kp.deleteFromTopicRequestHandlerChannelMap(topic.Name)
360}
361
khenaidoo4c1a5bf2018-11-29 15:53:42 -0500362// setupTopicResponseChannelMap sets up single consumers channel that will act as a broadcast channel for all
khenaidoo43c82122018-11-22 18:38:28 -0500363// responses from that topic.
khenaidoo79232702018-12-04 11:00:41 -0500364func (kp *InterContainerProxy) setupTopicResponseChannelMap(topic string, arg <-chan *ic.InterContainerMessage) {
khenaidoo43c82122018-11-22 18:38:28 -0500365 kp.lockTopicResponseChannelMap.Lock()
366 defer kp.lockTopicResponseChannelMap.Unlock()
367 if _, exist := kp.topicToResponseChannelMap[topic]; !exist {
368 kp.topicToResponseChannelMap[topic] = arg
khenaidooabad44c2018-08-03 16:58:35 -0400369 }
370}
371
khenaidoo43c82122018-11-22 18:38:28 -0500372func (kp *InterContainerProxy) isTopicSubscribedForResponse(topic string) bool {
khenaidoo1ce37ad2019-03-24 22:07:24 -0400373 kp.lockTopicResponseChannelMap.RLock()
374 defer kp.lockTopicResponseChannelMap.RUnlock()
khenaidoo43c82122018-11-22 18:38:28 -0500375 _, exist := kp.topicToResponseChannelMap[topic]
376 return exist
377}
378
379func (kp *InterContainerProxy) deleteFromTopicResponseChannelMap(topic string) error {
380 kp.lockTopicResponseChannelMap.Lock()
381 defer kp.lockTopicResponseChannelMap.Unlock()
382 if _, exist := kp.topicToResponseChannelMap[topic]; exist {
383 // Unsubscribe to this topic first - this will close the subscribed channel
384 var err error
385 if err = kp.kafkaClient.UnSubscribe(&Topic{Name: topic}, kp.topicToResponseChannelMap[topic]); err != nil {
386 log.Errorw("unsubscribing-error", log.Fields{"topic": topic})
387 }
388 delete(kp.topicToResponseChannelMap, topic)
389 return err
390 } else {
391 return errors.New(fmt.Sprintf("%s-Topic-not-found", topic))
khenaidooabad44c2018-08-03 16:58:35 -0400392 }
393}
394
khenaidoo43c82122018-11-22 18:38:28 -0500395func (kp *InterContainerProxy) deleteAllTopicResponseChannelMap() error {
396 kp.lockTopicResponseChannelMap.Lock()
397 defer kp.lockTopicResponseChannelMap.Unlock()
398 var err error
399 for topic, _ := range kp.topicToResponseChannelMap {
400 // Unsubscribe to this topic first - this will close the subscribed channel
401 if err = kp.kafkaClient.UnSubscribe(&Topic{Name: topic}, kp.topicToResponseChannelMap[topic]); err != nil {
402 log.Errorw("unsubscribing-error", log.Fields{"topic": topic, "error": err})
403 }
404 delete(kp.topicToResponseChannelMap, topic)
khenaidooabad44c2018-08-03 16:58:35 -0400405 }
khenaidoo43c82122018-11-22 18:38:28 -0500406 return err
khenaidooabad44c2018-08-03 16:58:35 -0400407}
408
khenaidoo43c82122018-11-22 18:38:28 -0500409func (kp *InterContainerProxy) addToTopicRequestHandlerChannelMap(topic string, arg *requestHandlerChannel) {
410 kp.lockTopicRequestHandlerChannelMap.Lock()
411 defer kp.lockTopicRequestHandlerChannelMap.Unlock()
412 if _, exist := kp.topicToRequestHandlerChannelMap[topic]; !exist {
413 kp.topicToRequestHandlerChannelMap[topic] = arg
khenaidooabad44c2018-08-03 16:58:35 -0400414 }
khenaidooabad44c2018-08-03 16:58:35 -0400415}
416
khenaidoo43c82122018-11-22 18:38:28 -0500417func (kp *InterContainerProxy) deleteFromTopicRequestHandlerChannelMap(topic string) error {
418 kp.lockTopicRequestHandlerChannelMap.Lock()
419 defer kp.lockTopicRequestHandlerChannelMap.Unlock()
420 if _, exist := kp.topicToRequestHandlerChannelMap[topic]; exist {
421 // Close the kafka client client first by unsubscribing to this topic
422 kp.kafkaClient.UnSubscribe(&Topic{Name: topic}, kp.topicToRequestHandlerChannelMap[topic].ch)
423 delete(kp.topicToRequestHandlerChannelMap, topic)
khenaidooabad44c2018-08-03 16:58:35 -0400424 return nil
khenaidoo43c82122018-11-22 18:38:28 -0500425 } else {
426 return errors.New(fmt.Sprintf("%s-Topic-not-found", topic))
khenaidooabad44c2018-08-03 16:58:35 -0400427 }
khenaidooabad44c2018-08-03 16:58:35 -0400428}
429
khenaidoo43c82122018-11-22 18:38:28 -0500430func (kp *InterContainerProxy) deleteAllTopicRequestHandlerChannelMap() error {
431 kp.lockTopicRequestHandlerChannelMap.Lock()
432 defer kp.lockTopicRequestHandlerChannelMap.Unlock()
433 var err error
434 for topic, _ := range kp.topicToRequestHandlerChannelMap {
435 // Close the kafka client client first by unsubscribing to this topic
436 if err = kp.kafkaClient.UnSubscribe(&Topic{Name: topic}, kp.topicToRequestHandlerChannelMap[topic].ch); err != nil {
437 log.Errorw("unsubscribing-error", log.Fields{"topic": topic, "error": err})
438 }
439 delete(kp.topicToRequestHandlerChannelMap, topic)
440 }
441 return err
442}
443
khenaidoo79232702018-12-04 11:00:41 -0500444func (kp *InterContainerProxy) addToTransactionIdToChannelMap(id string, topic *Topic, arg chan *ic.InterContainerMessage) {
khenaidooabad44c2018-08-03 16:58:35 -0400445 kp.lockTransactionIdToChannelMap.Lock()
446 defer kp.lockTransactionIdToChannelMap.Unlock()
447 if _, exist := kp.transactionIdToChannelMap[id]; !exist {
khenaidoo43c82122018-11-22 18:38:28 -0500448 kp.transactionIdToChannelMap[id] = &transactionChannel{topic: topic, ch: arg}
khenaidooabad44c2018-08-03 16:58:35 -0400449 }
450}
451
khenaidoo43c82122018-11-22 18:38:28 -0500452func (kp *InterContainerProxy) deleteFromTransactionIdToChannelMap(id string) {
khenaidooabad44c2018-08-03 16:58:35 -0400453 kp.lockTransactionIdToChannelMap.Lock()
454 defer kp.lockTransactionIdToChannelMap.Unlock()
khenaidoo43c82122018-11-22 18:38:28 -0500455 if transChannel, exist := kp.transactionIdToChannelMap[id]; exist {
456 // Close the channel first
457 close(transChannel.ch)
khenaidooabad44c2018-08-03 16:58:35 -0400458 delete(kp.transactionIdToChannelMap, id)
459 }
460}
461
khenaidoo43c82122018-11-22 18:38:28 -0500462func (kp *InterContainerProxy) deleteTopicTransactionIdToChannelMap(id string) {
463 kp.lockTransactionIdToChannelMap.Lock()
464 defer kp.lockTransactionIdToChannelMap.Unlock()
465 for key, value := range kp.transactionIdToChannelMap {
466 if value.topic.Name == id {
467 close(value.ch)
468 delete(kp.transactionIdToChannelMap, key)
khenaidooabad44c2018-08-03 16:58:35 -0400469 }
470 }
khenaidooabad44c2018-08-03 16:58:35 -0400471}
472
khenaidoo43c82122018-11-22 18:38:28 -0500473func (kp *InterContainerProxy) deleteAllTransactionIdToChannelMap() {
474 kp.lockTransactionIdToChannelMap.Lock()
475 defer kp.lockTransactionIdToChannelMap.Unlock()
476 for key, value := range kp.transactionIdToChannelMap {
477 close(value.ch)
478 delete(kp.transactionIdToChannelMap, key)
khenaidooabad44c2018-08-03 16:58:35 -0400479 }
khenaidooabad44c2018-08-03 16:58:35 -0400480}
481
khenaidoo43c82122018-11-22 18:38:28 -0500482func (kp *InterContainerProxy) DeleteTopic(topic Topic) error {
483 // If we have any consumers on that topic we need to close them
khenaidoo3dfc8bc2019-01-10 16:48:25 -0500484 if err := kp.deleteFromTopicResponseChannelMap(topic.Name); err != nil {
485 log.Errorw("delete-from-topic-responsechannelmap-failed", log.Fields{"error": err})
486 }
487 if err := kp.deleteFromTopicRequestHandlerChannelMap(topic.Name); err != nil {
488 log.Errorw("delete-from-topic-requesthandlerchannelmap-failed", log.Fields{"error": err})
489 }
khenaidoo43c82122018-11-22 18:38:28 -0500490 kp.deleteTopicTransactionIdToChannelMap(topic.Name)
khenaidoo3dfc8bc2019-01-10 16:48:25 -0500491
khenaidoo43c82122018-11-22 18:38:28 -0500492 return kp.kafkaClient.DeleteTopic(&topic)
493}
494
495func encodeReturnedValue(returnedVal interface{}) (*any.Any, error) {
khenaidooabad44c2018-08-03 16:58:35 -0400496 // Encode the response argument - needs to be a proto message
497 if returnedVal == nil {
498 return nil, nil
499 }
500 protoValue, ok := returnedVal.(proto.Message)
501 if !ok {
502 log.Warnw("response-value-not-proto-message", log.Fields{"error": ok, "returnVal": returnedVal})
503 err := errors.New("response-value-not-proto-message")
504 return nil, err
505 }
506
507 // Marshal the returned value, if any
508 var marshalledReturnedVal *any.Any
509 var err error
510 if marshalledReturnedVal, err = ptypes.MarshalAny(protoValue); err != nil {
511 log.Warnw("cannot-marshal-returned-val", log.Fields{"error": err})
512 return nil, err
513 }
514 return marshalledReturnedVal, nil
515}
516
khenaidoo79232702018-12-04 11:00:41 -0500517func encodeDefaultFailedResponse(request *ic.InterContainerMessage) *ic.InterContainerMessage {
518 responseHeader := &ic.Header{
khenaidooabad44c2018-08-03 16:58:35 -0400519 Id: request.Header.Id,
khenaidoo79232702018-12-04 11:00:41 -0500520 Type: ic.MessageType_RESPONSE,
khenaidooabad44c2018-08-03 16:58:35 -0400521 FromTopic: request.Header.ToTopic,
522 ToTopic: request.Header.FromTopic,
523 Timestamp: time.Now().Unix(),
524 }
khenaidoo79232702018-12-04 11:00:41 -0500525 responseBody := &ic.InterContainerResponseBody{
khenaidooabad44c2018-08-03 16:58:35 -0400526 Success: false,
527 Result: nil,
528 }
529 var marshalledResponseBody *any.Any
530 var err error
531 // Error should never happen here
532 if marshalledResponseBody, err = ptypes.MarshalAny(responseBody); err != nil {
533 log.Warnw("cannot-marshal-failed-response-body", log.Fields{"error": err})
534 }
535
khenaidoo79232702018-12-04 11:00:41 -0500536 return &ic.InterContainerMessage{
khenaidooabad44c2018-08-03 16:58:35 -0400537 Header: responseHeader,
538 Body: marshalledResponseBody,
539 }
540
541}
542
543//formatRequest formats a request to send over kafka and returns an InterContainerMessage message on success
544//or an error on failure
khenaidoo79232702018-12-04 11:00:41 -0500545func encodeResponse(request *ic.InterContainerMessage, success bool, returnedValues ...interface{}) (*ic.InterContainerMessage, error) {
khenaidoo43c82122018-11-22 18:38:28 -0500546 //log.Debugw("encodeResponse", log.Fields{"success": success, "returnedValues": returnedValues})
khenaidoo79232702018-12-04 11:00:41 -0500547 responseHeader := &ic.Header{
khenaidooabad44c2018-08-03 16:58:35 -0400548 Id: request.Header.Id,
khenaidoo79232702018-12-04 11:00:41 -0500549 Type: ic.MessageType_RESPONSE,
khenaidooabad44c2018-08-03 16:58:35 -0400550 FromTopic: request.Header.ToTopic,
551 ToTopic: request.Header.FromTopic,
khenaidoo2c6a0992019-04-29 13:46:56 -0400552 KeyTopic: request.Header.KeyTopic,
khenaidoo8f474192019-04-03 17:20:44 -0400553 Timestamp: time.Now().UnixNano(),
khenaidooabad44c2018-08-03 16:58:35 -0400554 }
555
556 // Go over all returned values
557 var marshalledReturnedVal *any.Any
558 var err error
559 for _, returnVal := range returnedValues {
khenaidoo43c82122018-11-22 18:38:28 -0500560 if marshalledReturnedVal, err = encodeReturnedValue(returnVal); err != nil {
khenaidooabad44c2018-08-03 16:58:35 -0400561 log.Warnw("cannot-marshal-response-body", log.Fields{"error": err})
562 }
563 break // for now we support only 1 returned value - (excluding the error)
564 }
565
khenaidoo79232702018-12-04 11:00:41 -0500566 responseBody := &ic.InterContainerResponseBody{
khenaidooabad44c2018-08-03 16:58:35 -0400567 Success: success,
568 Result: marshalledReturnedVal,
569 }
570
571 // Marshal the response body
572 var marshalledResponseBody *any.Any
573 if marshalledResponseBody, err = ptypes.MarshalAny(responseBody); err != nil {
574 log.Warnw("cannot-marshal-response-body", log.Fields{"error": err})
575 return nil, err
576 }
577
khenaidoo79232702018-12-04 11:00:41 -0500578 return &ic.InterContainerMessage{
khenaidooabad44c2018-08-03 16:58:35 -0400579 Header: responseHeader,
580 Body: marshalledResponseBody,
581 }, nil
582}
583
584func CallFuncByName(myClass interface{}, funcName string, params ...interface{}) (out []reflect.Value, err error) {
585 myClassValue := reflect.ValueOf(myClass)
khenaidoo19374072018-12-11 11:05:15 -0500586 // Capitalize the first letter in the funcName to workaround the first capital letters required to
587 // invoke a function from a different package
588 funcName = strings.Title(funcName)
khenaidooabad44c2018-08-03 16:58:35 -0400589 m := myClassValue.MethodByName(funcName)
590 if !m.IsValid() {
khenaidoo43c82122018-11-22 18:38:28 -0500591 return make([]reflect.Value, 0), fmt.Errorf("method-not-found \"%s\"", funcName)
khenaidooabad44c2018-08-03 16:58:35 -0400592 }
593 in := make([]reflect.Value, len(params))
594 for i, param := range params {
595 in[i] = reflect.ValueOf(param)
596 }
597 out = m.Call(in)
598 return
599}
600
khenaidoo297cd252019-02-07 22:10:23 -0500601func (kp *InterContainerProxy) addTransactionId(transactionId string, currentArgs []*ic.Argument) []*ic.Argument {
602 arg := &KVArg{
603 Key: TransactionKey,
604 Value: &ic.StrType{Val: transactionId},
605 }
606
607 var marshalledArg *any.Any
608 var err error
609 if marshalledArg, err = ptypes.MarshalAny(&ic.StrType{Val: transactionId}); err != nil {
610 log.Warnw("cannot-add-transactionId", log.Fields{"error": err})
611 return currentArgs
612 }
613 protoArg := &ic.Argument{
614 Key: arg.Key,
615 Value: marshalledArg,
616 }
617 return append(currentArgs, protoArg)
618}
619
khenaidoo54e0ddf2019-02-27 16:21:33 -0500620func (kp *InterContainerProxy) addFromTopic(fromTopic string, currentArgs []*ic.Argument) []*ic.Argument {
621 var marshalledArg *any.Any
622 var err error
623 if marshalledArg, err = ptypes.MarshalAny(&ic.StrType{Val: fromTopic}); err != nil {
624 log.Warnw("cannot-add-transactionId", log.Fields{"error": err})
625 return currentArgs
626 }
627 protoArg := &ic.Argument{
628 Key: FromTopic,
629 Value: marshalledArg,
630 }
631 return append(currentArgs, protoArg)
632}
633
634func (kp *InterContainerProxy) handleMessage(msg *ic.InterContainerMessage, targetInterface interface{}) {
khenaidooabad44c2018-08-03 16:58:35 -0400635
khenaidoo43c82122018-11-22 18:38:28 -0500636 // First extract the header to know whether this is a request - responses are handled by a different handler
khenaidoo79232702018-12-04 11:00:41 -0500637 if msg.Header.Type == ic.MessageType_REQUEST {
khenaidooabad44c2018-08-03 16:58:35 -0400638 var out []reflect.Value
639 var err error
640
641 // Get the request body
khenaidoo79232702018-12-04 11:00:41 -0500642 requestBody := &ic.InterContainerRequestBody{}
khenaidooabad44c2018-08-03 16:58:35 -0400643 if err = ptypes.UnmarshalAny(msg.Body, requestBody); err != nil {
644 log.Warnw("cannot-unmarshal-request", log.Fields{"error": err})
645 } else {
khenaidoo43c82122018-11-22 18:38:28 -0500646 log.Debugw("received-request", log.Fields{"rpc": requestBody.Rpc, "header": msg.Header})
khenaidooabad44c2018-08-03 16:58:35 -0400647 // let the callee unpack the arguments as its the only one that knows the real proto type
khenaidoo297cd252019-02-07 22:10:23 -0500648 // Augment the requestBody with the message Id as it will be used in scenarios where cores
649 // are set in pairs and competing
650 requestBody.Args = kp.addTransactionId(msg.Header.Id, requestBody.Args)
khenaidoo54e0ddf2019-02-27 16:21:33 -0500651
652 // Augment the requestBody with the From topic name as it will be used in scenarios where a container
653 // needs to send an unsollicited message to the currently requested container
654 requestBody.Args = kp.addFromTopic(msg.Header.FromTopic, requestBody.Args)
655
khenaidooabad44c2018-08-03 16:58:35 -0400656 out, err = CallFuncByName(targetInterface, requestBody.Rpc, requestBody.Args)
657 if err != nil {
658 log.Warn(err)
659 }
660 }
661 // Response required?
662 if requestBody.ResponseRequired {
663 // If we already have an error before then just return that
khenaidoo79232702018-12-04 11:00:41 -0500664 var returnError *ic.Error
khenaidooabad44c2018-08-03 16:58:35 -0400665 var returnedValues []interface{}
666 var success bool
667 if err != nil {
khenaidoo79232702018-12-04 11:00:41 -0500668 returnError = &ic.Error{Reason: err.Error()}
khenaidooabad44c2018-08-03 16:58:35 -0400669 returnedValues = make([]interface{}, 1)
670 returnedValues[0] = returnError
671 } else {
khenaidoob9203542018-09-17 22:56:37 -0400672 returnedValues = make([]interface{}, 0)
673 // Check for errors first
674 lastIndex := len(out) - 1
675 if out[lastIndex].Interface() != nil { // Error
676 if goError, ok := out[lastIndex].Interface().(error); ok {
khenaidoo79232702018-12-04 11:00:41 -0500677 returnError = &ic.Error{Reason: goError.Error()}
khenaidoob9203542018-09-17 22:56:37 -0400678 returnedValues = append(returnedValues, returnError)
679 } else { // Should never happen
khenaidoo79232702018-12-04 11:00:41 -0500680 returnError = &ic.Error{Reason: "incorrect-error-returns"}
khenaidoob9203542018-09-17 22:56:37 -0400681 returnedValues = append(returnedValues, returnError)
682 }
khenaidoo54e0ddf2019-02-27 16:21:33 -0500683 } else if len(out) == 2 && reflect.ValueOf(out[0].Interface()).IsValid() && reflect.ValueOf(out[0].Interface()).IsNil() {
khenaidoo297cd252019-02-07 22:10:23 -0500684 return // Ignore case - when core is in competing mode
khenaidoob9203542018-09-17 22:56:37 -0400685 } else { // Non-error case
686 success = true
687 for idx, val := range out {
khenaidoo43c82122018-11-22 18:38:28 -0500688 //log.Debugw("returned-api-response-loop", log.Fields{"idx": idx, "val": val.Interface()})
khenaidoob9203542018-09-17 22:56:37 -0400689 if idx != lastIndex {
690 returnedValues = append(returnedValues, val.Interface())
khenaidooabad44c2018-08-03 16:58:35 -0400691 }
khenaidooabad44c2018-08-03 16:58:35 -0400692 }
693 }
694 }
695
khenaidoo79232702018-12-04 11:00:41 -0500696 var icm *ic.InterContainerMessage
khenaidooabad44c2018-08-03 16:58:35 -0400697 if icm, err = encodeResponse(msg, success, returnedValues...); err != nil {
khenaidoob9203542018-09-17 22:56:37 -0400698 log.Warnw("error-encoding-response-returning-failure-result", log.Fields{"error": err})
khenaidooabad44c2018-08-03 16:58:35 -0400699 icm = encodeDefaultFailedResponse(msg)
700 }
khenaidoo43c82122018-11-22 18:38:28 -0500701 // To preserve ordering of messages, all messages to a given topic are sent to the same partition
702 // by providing a message key. The key is encoded in the topic name. If the deviceId is not
703 // present then the key will be empty, hence all messages for a given topic will be sent to all
704 // partitions.
705 replyTopic := &Topic{Name: msg.Header.FromTopic}
khenaidoobdcb8e02019-03-06 16:28:56 -0500706 key := msg.Header.KeyTopic
khenaidoo90847922018-12-03 14:47:51 -0500707 log.Debugw("sending-response-to-kafka", log.Fields{"rpc": requestBody.Rpc, "header": icm.Header, "key": key})
khenaidoo43c82122018-11-22 18:38:28 -0500708 // TODO: handle error response.
khenaidoo2c6a0992019-04-29 13:46:56 -0400709 go kp.kafkaClient.Send(icm, replyTopic, key)
khenaidooabad44c2018-08-03 16:58:35 -0400710 }
khenaidoo54e0ddf2019-02-27 16:21:33 -0500711 } else if msg.Header.Type == ic.MessageType_RESPONSE {
khenaidoo1ce37ad2019-03-24 22:07:24 -0400712 log.Debugw("response-received", log.Fields{"msg-header": msg.Header})
khenaidoo54e0ddf2019-02-27 16:21:33 -0500713 go kp.dispatchResponse(msg)
714 } else {
khenaidoo1ce37ad2019-03-24 22:07:24 -0400715 log.Warnw("unsupported-message-received", log.Fields{"msg-header": msg.Header})
khenaidooabad44c2018-08-03 16:58:35 -0400716 }
717}
718
khenaidoo54e0ddf2019-02-27 16:21:33 -0500719func (kp *InterContainerProxy) waitForMessages(ch <-chan *ic.InterContainerMessage, topic Topic, targetInterface interface{}) {
khenaidooabad44c2018-08-03 16:58:35 -0400720 // Wait for messages
721 for msg := range ch {
khenaidoo43c82122018-11-22 18:38:28 -0500722 //log.Debugw("request-received", log.Fields{"msg": msg, "topic": topic.Name, "target": targetInterface})
khenaidoo54e0ddf2019-02-27 16:21:33 -0500723 go kp.handleMessage(msg, targetInterface)
khenaidooabad44c2018-08-03 16:58:35 -0400724 }
725}
726
khenaidoo79232702018-12-04 11:00:41 -0500727func (kp *InterContainerProxy) dispatchResponse(msg *ic.InterContainerMessage) {
khenaidoo1ce37ad2019-03-24 22:07:24 -0400728 kp.lockTransactionIdToChannelMap.RLock()
729 defer kp.lockTransactionIdToChannelMap.RUnlock()
khenaidooabad44c2018-08-03 16:58:35 -0400730 if _, exist := kp.transactionIdToChannelMap[msg.Header.Id]; !exist {
731 log.Debugw("no-waiting-channel", log.Fields{"transaction": msg.Header.Id})
732 return
733 }
khenaidoo43c82122018-11-22 18:38:28 -0500734 kp.transactionIdToChannelMap[msg.Header.Id].ch <- msg
khenaidooabad44c2018-08-03 16:58:35 -0400735}
736
khenaidooabad44c2018-08-03 16:58:35 -0400737// subscribeForResponse allows a caller to subscribe to a given topic when waiting for a response.
738// This method is built to prevent all subscribers to receive all messages as is the case of the Subscribe
739// API. There is one response channel waiting for kafka messages before dispatching the message to the
740// corresponding waiting channel
khenaidoo79232702018-12-04 11:00:41 -0500741func (kp *InterContainerProxy) subscribeForResponse(topic Topic, trnsId string) (chan *ic.InterContainerMessage, error) {
khenaidoob9203542018-09-17 22:56:37 -0400742 log.Debugw("subscribeForResponse", log.Fields{"topic": topic.Name, "trnsid": trnsId})
khenaidooabad44c2018-08-03 16:58:35 -0400743
khenaidoo4c1a5bf2018-11-29 15:53:42 -0500744 // Create a specific channel for this consumers. We cannot use the channel from the kafkaclient as it will
khenaidoo43c82122018-11-22 18:38:28 -0500745 // broadcast any message for this topic to all channels waiting on it.
khenaidoo79232702018-12-04 11:00:41 -0500746 ch := make(chan *ic.InterContainerMessage)
khenaidoo43c82122018-11-22 18:38:28 -0500747 kp.addToTransactionIdToChannelMap(trnsId, &topic, ch)
khenaidooabad44c2018-08-03 16:58:35 -0400748
749 return ch, nil
750}
751
khenaidoo43c82122018-11-22 18:38:28 -0500752func (kp *InterContainerProxy) unSubscribeForResponse(trnsId string) error {
khenaidooabad44c2018-08-03 16:58:35 -0400753 log.Debugw("unsubscribe-for-response", log.Fields{"trnsId": trnsId})
khenaidoo7ff26c72019-01-16 14:55:48 -0500754 kp.deleteFromTransactionIdToChannelMap(trnsId)
khenaidooabad44c2018-08-03 16:58:35 -0400755 return nil
756}
757
758//formatRequest formats a request to send over kafka and returns an InterContainerMessage message on success
759//or an error on failure
khenaidoobdcb8e02019-03-06 16:28:56 -0500760func encodeRequest(rpc string, toTopic *Topic, replyTopic *Topic, key string, kvArgs ...*KVArg) (*ic.InterContainerMessage, error) {
khenaidoo79232702018-12-04 11:00:41 -0500761 requestHeader := &ic.Header{
khenaidooabad44c2018-08-03 16:58:35 -0400762 Id: uuid.New().String(),
khenaidoo79232702018-12-04 11:00:41 -0500763 Type: ic.MessageType_REQUEST,
khenaidooabad44c2018-08-03 16:58:35 -0400764 FromTopic: replyTopic.Name,
765 ToTopic: toTopic.Name,
khenaidoo2c6a0992019-04-29 13:46:56 -0400766 KeyTopic: key,
khenaidoo8f474192019-04-03 17:20:44 -0400767 Timestamp: time.Now().UnixNano(),
khenaidooabad44c2018-08-03 16:58:35 -0400768 }
khenaidoo79232702018-12-04 11:00:41 -0500769 requestBody := &ic.InterContainerRequestBody{
khenaidooabad44c2018-08-03 16:58:35 -0400770 Rpc: rpc,
771 ResponseRequired: true,
772 ReplyToTopic: replyTopic.Name,
773 }
774
775 for _, arg := range kvArgs {
khenaidoo2c6f1672018-09-20 23:14:41 -0400776 if arg == nil {
777 // In case the caller sends an array with empty args
778 continue
779 }
khenaidooabad44c2018-08-03 16:58:35 -0400780 var marshalledArg *any.Any
781 var err error
782 // ascertain the value interface type is a proto.Message
783 protoValue, ok := arg.Value.(proto.Message)
784 if !ok {
785 log.Warnw("argument-value-not-proto-message", log.Fields{"error": ok, "Value": arg.Value})
786 err := errors.New("argument-value-not-proto-message")
787 return nil, err
788 }
789 if marshalledArg, err = ptypes.MarshalAny(protoValue); err != nil {
790 log.Warnw("cannot-marshal-request", log.Fields{"error": err})
791 return nil, err
792 }
khenaidoo79232702018-12-04 11:00:41 -0500793 protoArg := &ic.Argument{
khenaidooabad44c2018-08-03 16:58:35 -0400794 Key: arg.Key,
795 Value: marshalledArg,
796 }
797 requestBody.Args = append(requestBody.Args, protoArg)
798 }
799
800 var marshalledData *any.Any
801 var err error
802 if marshalledData, err = ptypes.MarshalAny(requestBody); err != nil {
803 log.Warnw("cannot-marshal-request", log.Fields{"error": err})
804 return nil, err
805 }
khenaidoo79232702018-12-04 11:00:41 -0500806 request := &ic.InterContainerMessage{
khenaidooabad44c2018-08-03 16:58:35 -0400807 Header: requestHeader,
808 Body: marshalledData,
809 }
810 return request, nil
811}
812
khenaidoo79232702018-12-04 11:00:41 -0500813func decodeResponse(response *ic.InterContainerMessage) (*ic.InterContainerResponseBody, error) {
khenaidooabad44c2018-08-03 16:58:35 -0400814 // Extract the message body
khenaidoo79232702018-12-04 11:00:41 -0500815 responseBody := ic.InterContainerResponseBody{}
khenaidooabad44c2018-08-03 16:58:35 -0400816 if err := ptypes.UnmarshalAny(response.Body, &responseBody); err != nil {
817 log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
818 return nil, err
819 }
khenaidoo43c82122018-11-22 18:38:28 -0500820 //log.Debugw("response-decoded-successfully", log.Fields{"response-status": &responseBody.Success})
khenaidooabad44c2018-08-03 16:58:35 -0400821
822 return &responseBody, nil
823
824}