chris/styx
chris/styx
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Moving SendToKafka function to broker package

Browse source
This commit is contained in:
Christopher Talib 2020-01-17 13:51:11 +01:00
parent e9c065bcc8
commit 06f42fc472
3 changed files with 13 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
broker/main.go
View file

@ -2,9 +2,11 @@ package broker
import (
"context"
"fmt"
"time"
"github.com/segmentio/kafka-go"
"gitlab.dcso.lolcat/LABS/styx/models"
)
func SetUpKafkaConnecter() (*kafka.Conn, error) {
@ -38,6 +40,12 @@ func ProduceEvent(conn *kafka.Conn, message string) error {
return nil
}
func consumeEvent() {
func SendEventToKafka(conn *kafka.Conn, node models.Node) error {
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
_, err := conn.WriteMessages(kafka.Message{Value: []byte(fmt.Sprintf("%v", node))})
if err != nil {
panic(err)
}
return nil
}

2
main.go
View file

@ -28,7 +28,7 @@ func main() {
}
fmt.Println(node)
err = models.SendEventToKafka(Conn, *node)
err = broker.SendEventToKafka(Conn, *node)
if err != nil {
panic(err)
}

25
models/main.go
View file

@ -2,13 +2,10 @@ package models
import (
"encoding/json"
"fmt"
"io/ioutil"
"os"
"time"
"github.com/google/uuid"
"github.com/segmentio/kafka-go"
"github.com/sirupsen/logrus"
)
@ -50,10 +47,9 @@ func SaveData(flag string, data CertStreamStruct) {
// res := CertStreamStruct{}
// json.Unmarshal(data, &res)
node := &Node{
ID: flag + "--" + uuid.New().String(),
Type: flag,
Data: data,
node, err := BuildNode(flag, data)
if err != nil {
logrus.Error(err)
}
// edge := &Edge{
@ -87,11 +83,6 @@ func SaveData(flag string, data CertStreamStruct) {
}
func BuildNode(flag string, data CertStreamStruct) (*Node, error) {
err := fileExists(nodesFilename)
if err != nil {
return nil, err
}
return &Node{
ID: flag + "--" + uuid.New().String(),
Type: flag,
@ -100,16 +91,6 @@ func BuildNode(flag string, data CertStreamStruct) (*Node, error) {
}
func SendEventToKafka(conn *kafka.Conn, node Node) error {
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
_, err := conn.WriteMessages(kafka.Message{Value: []byte(fmt.Sprintf("%v", node))})
if err != nil {
panic(err)
}
return nil
}
// Helpers
func fileExists(filename string) error {