Stop writer closing

ale8k · Dec 1, 2021 · eb82b66 · eb82b66
1 parent 9413178
commit eb82b66
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/internal/question_processor.go b/internal/question_processor.go
@@ -12,7 +12,7 @@ import (
 )
 
 type QuestionProcessor struct {
-	KafkaWriter  kafka.Writer
+	KafkaWriter  *kafka.Writer
 	Tags         string
 	CollectEvery time.Duration
 }
@@ -95,6 +95,7 @@ func (qp *QuestionProcessor) ProcessAskedQuestions(done chan struct{}) {
 
 	// Wrapper to process
 	processQuestionsToKafka := func() {
+
 		data, _ := runQuery()
 		parsedMessages := make(chan kafka.Message, len(data))
 		// TODO: Make batching size configurable alex

diff --git a/internal/server.go b/internal/server.go
@@ -20,7 +20,8 @@ var Server *http.Server
 func processorWrapper(topicname string, tags string) {
 	done := make(chan struct{})
 	phpQp := &QuestionProcessor{
-		KafkaWriter: kafka.Writer{
+		// Seems to close otherwise...?
+		KafkaWriter: &kafka.Writer{
 			Addr:         kafka.TCP("kafka:9092"),
 			Topic:        topicname,
 			Balancer:     &kafka.LeastBytes{},