From 4d2510c99d8232173d34be107126a5280558e2db Mon Sep 17 00:00:00 2001
From: fengzhiqiang <fengzhq@it2000.com.cn>
Date: Mon, 7 Jan 2019 11:45:13 +0800
Subject: [PATCH] add ignore

---
 .gitignore         |  1 +
 Dockerfile         |  5 ++++
 build-docker.sh    |  2 ++
 build-linux-bin.sh |  2 ++
 serv.go            | 75 ++++++++++++++++++++++++++++------------------
 5 files changed, 56 insertions(+), 29 deletions(-)
 create mode 100644 .gitignore
 create mode 100644 Dockerfile
 create mode 100644 build-docker.sh
 create mode 100644 build-linux-bin.sh

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..fcc1549
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+./proxy
\ No newline at end of file
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..7701765
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,5 @@
+FROM alpine:latest
+ADD ./proxy /app/ 
+WORKDIR /app
+EXPOSE 9000
+ENTRYPOINT ["/app/proxy","-addr",":9000"]
\ No newline at end of file
diff --git a/build-docker.sh b/build-docker.sh
new file mode 100644
index 0000000..2027d72
--- /dev/null
+++ b/build-docker.sh
@@ -0,0 +1,2 @@
+#!/bin/zsh
+docker build -t kafka-websocket-go:0.0.1 .
\ No newline at end of file
diff --git a/build-linux-bin.sh b/build-linux-bin.sh
new file mode 100644
index 0000000..6f9198a
--- /dev/null
+++ b/build-linux-bin.sh
@@ -0,0 +1,2 @@
+#!/bin/zsh
+env GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o proxy .
\ No newline at end of file
diff --git a/serv.go b/serv.go
index e3d3fd8..f85f328 100644
--- a/serv.go
+++ b/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.