add ignore

serv.go

@@ -9,11 +9,11 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/confluentinc/confluent-kafka-go/kafka"
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 	"github.com/gorilla/websocket"
 	uuid "github.com/satori/go.uuid"
+	kafka "github.com/segmentio/kafka-go"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -76,23 +76,31 @@ func (c *Client) readPump(topic string) {
 		"bootstrap servers": *bootstrapServers,
 		"group":             group}).Info("create kafka client")
 
-	consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
-		"bootstrap.servers":  *bootstrapServers,
-		"group.id":           group,
-		"session.timeout.ms": 6000,
-		"auto.offset.reset":  "latest"})
+	// consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
+	// 	"bootstrap.servers":  *bootstrapServers,
+	// 	"group.id":           group,
+	// 	"session.timeout.ms": 6000,
+	// 	"auto.offset.reset":  "latest"})
+	reader := kafka.NewReader(kafka.ReaderConfig{
+		Brokers:   []string{*bootstrapServers},
+		Topic:     topic,
+		Partition: 0,
+		MinBytes:  10e1,
+		MaxBytes:  10e6, // 10MB
+	})
+	reader.SetOffset(-1) //lastoffset
 
-	if err != nil {
-		// fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
-		log.WithError(err).Fatal("Failed to create consumer")
-		os.Exit(1)
-	}
+	// if err != nil {
+	// 	// fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
+	// 	log.WithError(err).Fatal("Failed to create consumer")
+	// 	os.Exit(1)
+	// }
 
 	log.WithFields(log.Fields{
-		"Consumer": consumer,
+		"reader": reader,
 	}).Info("Created ")
 
-	consumer.SubscribeTopics([]string{topic}, nil)
+	// consumer.SubscribeTopics([]string{topic}, nil)
 
 	run := true
 
@@ -103,28 +111,37 @@ func (c *Client) readPump(topic string) {
 			run = false
 			log.Fatal("break ")
 		default:
-			ev := consumer.Poll(100)
-			if ev == nil {
-				continue
-			}
+			// ev := consumer.Poll(100)
+			// if ev == nil {
+			// 	continue
+			// }
 
-			switch e := ev.(type) {
-			case *kafka.Message:
-				log.WithFields(log.Fields{"partition": e.TopicPartition, "message": string(e.Value)}).Info("got message")
-				c.hub.broadcast <- e.Value
-				if e.Headers != nil {
-					log.WithFields(log.Fields{"header": e.Headers}).Info(" with header")
+			// switch e := ev.(type) {
+			// case *kafka.Message:
+			// 	log.WithFields(log.Fields{"partition": e.TopicPartition, "message": string(e.Value)}).Info("got message")
+			// 	c.hub.broadcast <- e.Value
+			// 	if e.Headers != nil {
+			// 		log.WithFields(log.Fields{"header": e.Headers}).Info(" with header")
+			// 	}
+			// case kafka.Error:
+			// 	// Errors should generally be considered as informational, the client will try to automatically recover
+			// 	log.WithFields(log.Fields{"error": e}).Error("got error")
+			// default:
+			// 	log.WithFields(log.Fields{"event": e}).Info("ignored")
+			// }
+			for {
+				m, err := reader.ReadMessage(context.Background())
+				if err != nil {
+					log.WithFields(log.Fields{"error": err}).Error("read error")
 				}
-			case kafka.Error:
-				// Errors should generally be considered as informational, the client will try to automatically recover
-				log.WithFields(log.Fields{"error": e}).Error("got error")
-			default:
-				log.WithFields(log.Fields{"event": e}).Info("ignored")
+				// fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
+				log.WithFields(log.Fields{"offset": m.Offset, "message": string(m.Value)}).Info("got message")
+				c.hub.broadcast <- m.Value
 			}
 		}
 	}
 
-	consumer.Close()
+	// consumer.Close()
 }
 
 // writePump pumps messages from the hub to the websocket connection.