From 94eefe498ee5f8f50fd65c008dbb4f74df368396 Mon Sep 17 00:00:00 2001
From: fengzhiqiang <fengzhq@it2000.com.cn>
Date: Wed, 9 Jan 2019 16:46:40 +0800
Subject: [PATCH] =?UTF-8?q?60=E7=A7=92=20ping?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 serv.go | 21 ++++++++++++++++-----
 1 file changed, 16 insertions(+), 5 deletions(-)

diff --git a/serv.go b/serv.go
index 0b4808b..d72ab9f 100644
--- a/serv.go
+++ b/serv.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"flag"
+	"io"
 	"net/http"
 	"os"
 	"os/signal"
@@ -24,7 +25,7 @@ const (
 	writeWait = 10 * time.Second
 
 	// Time allowed to read the next pong message from the peer.
-	pongWait = 30 * time.Second
+	pongWait = 60 * time.Second
 
 	// Send pings to peer with this period. Must be less than pongWait.
 	pingPeriod = (pongWait * 9) / 10
@@ -38,11 +39,12 @@ var (
 	space            = []byte{' '}
 	addr             = flag.String("addr", ":9000", "http service address")
 	bootstrapServers = flag.String("bootstrap-servers", "localhost:9092", "kafka bootstrap servers")
+	logfile          = flag.String("logfile", "", "log file")
 	// partition        = flag.Int("partition", 0, "kafka tipic partion")
 	wait     = flag.Duration("graceful-timeout", time.Second*15, "the duration for which the server gracefully wait for existing connections to finish - e.g. 15s or 1m")
 	upgrader = websocket.Upgrader{
-		ReadBufferSize:  2048,
-		WriteBufferSize: 2048,
+		ReadBufferSize:  1024 * 4,
+		WriteBufferSize: 1024 * 4,
 	}
 	hubMap = make(map[string]Hub)
 	router = mux.NewRouter()
@@ -175,7 +177,7 @@ func (c *Client) writePump() {
 		case message, ok := <-c.send:
 			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
 			if !ok {
-				log.WithField("client", c).Warn("sending close message")
+				log.WithFields(log.Fields{"client": c.conn.RemoteAddr(), "msg": string(message)}).Warn("sending websocket close message")
 				// The hub closed the channel.
 				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
 				return
@@ -252,7 +254,7 @@ func newHub(topic string, offset int64, partition int) *Hub {
 		partition:  partition,
 		offset:     offset,
 		reader:     createKafkaReader(topic, offset, partition),
-		broadcast:  make(chan []byte),
+		broadcast:  make(chan []byte, 512),
 		register:   make(chan *Client),
 		unregister: make(chan *Client),
 		clients:    make(map[*Client]bool),
@@ -337,6 +339,15 @@ func destroy() {
 
 func main() {
 	flag.Parse()
+	if *logfile != "" {
+		if logFile, err := os.OpenFile(*logfile, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0644); err == nil {
+			log.WithField("file", *logfile).Info("start with log file")
+			mw := io.MultiWriter(os.Stdout, logFile)
+			log.SetOutput(mw)
+		} else {
+			log.WithField("file", *logfile).WithError(err).Fatal("open log file")
+		}
+	}
 	loggedRouter := handlers.LoggingHandler(os.Stdout, router)
 	srv := &http.Server{
 		Addr: *addr,