@@ -3,7 +3,6 @@ package main
33import (
44 "crypto/tls"
55 "crypto/x509"
6- "encoding/json"
76 "errors"
87 "fmt"
98 "io"
@@ -12,10 +11,12 @@ import (
1211 "net"
1312 "path/filepath"
1413 "strconv"
14+ "strings"
1515 "time"
1616
1717 "github.com/Shopify/sarama"
1818 "github.com/golang/protobuf/proto"
19+ "github.com/nileshsimaria/jtimon/dialout"
1920 gnmi_dialout "github.com/nileshsimaria/jtimon/gnmi/dialout"
2021 "github.com/nileshsimaria/jtimon/gnmi/gnmi"
2122
@@ -52,25 +53,19 @@ type rpcInfoT struct {
5253 rpc string
5354 rpcId int8
5455 running bool
55- cfgChannel chan * dialOutConfigT
56- config * dialOutConfigT
56+ cfgChannel chan * dialout. DialOutRequest
57+ config * dialout. DialOutRequest
5758 jctx * JCtx // Currently inherited from device, so use this handle as read-only
5859
5960 device * deviceInfoT
6061}
6162
62- type dialOutConfigT struct {
63- Device string `json:"device"`
64- RpcType string `json:"rpc-type"`
65- Paths []PathsConfig `json:"paths"`
66- }
67-
6863func newDialOutServer (rpcs []string ) * dialoutServerT {
6964 s := & dialoutServerT {devices : map [string ]* deviceInfoT {}}
7065
7166 // Create kafka Client
7267 kafkaCfg := sarama .NewConfig ()
73- kafkaClient , err := sarama .NewClient ([] string { * kafkaBroker } , kafkaCfg )
68+ kafkaClient , err := sarama .NewClient (strings . Split ( * kafkaBroker , "," ) , kafkaCfg )
7469 if err != nil {
7570 log .Fatalf ("Not able to connect to Kafka broker at %v: %v" , * kafkaBroker , err )
7671 }
@@ -109,7 +104,7 @@ func newDialOutServer(rpcs []string) *dialoutServerT {
109104
110105func createRpc (device * deviceInfoT , name string , id int ) (* rpcInfoT , error ) {
111106 // Sarama's default channel size for consuming messages is 256
112- cfgChannel := make (chan * dialOutConfigT , 512 )
107+ cfgChannel := make (chan * dialout. DialOutRequest , 512 )
113108 return & rpcInfoT {rpc : name , rpcId : int8 (id ), cfgChannel : cfgChannel , jctx : device .jctx , device : device }, nil
114109}
115110
@@ -217,15 +212,15 @@ func (s *dialoutServerT) DialOutSubscriber(stream gnmi_dialout.Subscriber_DialOu
217212
218213 i := 0
219214 length := len (rpc .cfgChannel )
220- var dialOutCfg * dialOutConfigT
215+ var cfgReq * dialout. DialOutRequest
221216 // Drain n - 1 message to get latest config
222217 for i < (length - 1 ) {
223218 <- rpc .cfgChannel
224219 i ++
225220 }
226221 log .Printf ("Waiting for config.." )
227- dialOutCfg = <- rpc .cfgChannel
228- log .Printf ("Read config.. : %v" , * dialOutCfg )
222+ cfgReq = <- rpc .cfgChannel
223+ log .Printf ("Read config.. : %v" , * cfgReq )
229224
230225 req := & gnmi.SubscribeRequest {}
231226 req = & gnmi.SubscribeRequest {Request : & gnmi.SubscribeRequest_Subscribe {
@@ -235,7 +230,7 @@ func (s *dialoutServerT) DialOutSubscriber(stream gnmi_dialout.Subscriber_DialOu
235230 },
236231 }}
237232 subReq := req .Request .(* gnmi.SubscribeRequest_Subscribe )
238- subReq .Subscribe .Subscription , err = xPathsTognmiSubscription (dialOutCfg .Paths )
233+ subReq .Subscribe .Subscription , err = xPathsTognmiSubscription (nil , cfgReq .Paths )
239234 if err != nil {
240235 //log.Printf("Host: %v, Invalid path config: %s", cn, cfg)
241236 jLog (s .jctx , fmt .Sprintf ("Host: %v, Invalid path config: %s" , cn , "foo...." ))
@@ -247,7 +242,7 @@ func (s *dialoutServerT) DialOutSubscriber(stream gnmi_dialout.Subscriber_DialOu
247242
248243 for {
249244 select {
250- case dialOutCfg = <- rpc .cfgChannel :
245+ case cfgReq = <- rpc .cfgChannel :
251246 removeRpcFromDevice (rpc )
252247 stream .Context ().Done ()
253248 case producerError := <- (* s .dataProducer ).Errors ():
@@ -268,13 +263,16 @@ func (s *dialoutServerT) DialOutSubscriber(stream gnmi_dialout.Subscriber_DialOu
268263 continue
269264 }
270265
271- payload , err := proto .Marshal (rspFromDevice )
266+ // TODO: Vivek Take data topic from cmd line
267+ var dialOutRsp dialout.DialOutResponse
268+ dialOutRsp .Device = cn
269+ dialOutRsp .DialOutContext = cfgReq .DialOutContext
270+ dialOutRsp .Response = append (dialOutRsp .Response , rspFromDevice )
271+ payload , err := proto .Marshal (& dialOutRsp )
272272 if err != nil {
273273 log .Printf ("Marshalling failed for %s, len: %v\n " , rspString , len (rspString ))
274274 continue
275275 }
276-
277- // TODO: Vivek Take data topic from cmd line
278276 (* s .dataProducer ).Input () <- & sarama.ProducerMessage {Topic : "gnmi-data" , Key : sarama .ByteEncoder (cn ), Value : sarama .ByteEncoder (payload )}
279277 }
280278 }
@@ -295,8 +293,8 @@ func consumePartition(server *dialoutServerT, topic string, partition int32, off
295293 var tmpDeviceName string
296294 for msg := range partitionConsumer .Messages () {
297295 log .Printf ("topic: %v, partition: %v, offset: %v, msg key: %v, msg val: %v" , topic , partition , offset , string (msg .Key ), string (msg .Value ))
298- var dialOutCfg dialOutConfigT
299- err = json .Unmarshal (msg .Value , & dialOutCfg )
296+ var dialOutCfg dialout. DialOutRequest
297+ err = proto .Unmarshal (msg .Value , & dialOutCfg )
300298 log .Printf ("dialOutCfg: %v" , dialOutCfg .Paths [0 ].Path )
301299 if err != nil {
302300 jLog (jctx , fmt .Sprintf ("Unmarshalling dialout config failed, ignoring" ))
0 commit comments