windy/kafka-proxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

60秒 ping

Browse Source
This commit is contained in:
fengzhiqiang
2019-01-09 16:46:40 +08:00
parent 76eaf270be
commit 94eefe498e
1 changed files with 16 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
serv.go
+16 -5
View File
@@ -4,6 +4,7 @@ import (
"bytes" "bytes"
"context" "context"
"flag" "flag"
"io"
"net/http" "net/http"
"os" "os"
"os/signal" "os/signal"
@@ -24,7 +25,7 @@ const (
writeWait = 10 * time.Second writeWait = 10 * time.Second
// Time allowed to read the next pong message from the peer. // 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. // Send pings to peer with this period. Must be less than pongWait.
pingPeriod = (pongWait * 9) / 10 pingPeriod = (pongWait * 9) / 10
@@ -38,11 +39,12 @@ var (
space = []byte{' '} space = []byte{' '}
addr = flag.String("addr", ":9000", "http service address") addr = flag.String("addr", ":9000", "http service address")
bootstrapServers = flag.String("bootstrap-servers", "localhost:9092", "kafka bootstrap servers") bootstrapServers = flag.String("bootstrap-servers", "localhost:9092", "kafka bootstrap servers")
logfile = flag.String("logfile", "", "log file")
// partition = flag.Int("partition", 0, "kafka tipic partion") // 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") 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{ upgrader = websocket.Upgrader{
ReadBufferSize: 2048, ReadBufferSize: 1024 * 4,
WriteBufferSize: 2048, WriteBufferSize: 1024 * 4,
} }
hubMap = make(map[string]Hub) hubMap = make(map[string]Hub)
router = mux.NewRouter() router = mux.NewRouter()
@@ -175,7 +177,7 @@ func (c *Client) writePump() {
case message, ok := <-c.send: case message, ok := <-c.send:
c.conn.SetWriteDeadline(time.Now().Add(writeWait)) c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if !ok { 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. // The hub closed the channel.
c.conn.WriteMessage(websocket.CloseMessage, []byte{}) c.conn.WriteMessage(websocket.CloseMessage, []byte{})
return return
@@ -252,7 +254,7 @@ func newHub(topic string, offset int64, partition int) *Hub {
partition: partition, partition: partition,
offset: offset, offset: offset,
reader: createKafkaReader(topic, offset, partition), reader: createKafkaReader(topic, offset, partition),
broadcast: make(chan []byte), broadcast: make(chan []byte, 512),
register: make(chan *Client), register: make(chan *Client),
unregister: make(chan *Client), unregister: make(chan *Client),
clients: make(map[*Client]bool), clients: make(map[*Client]bool),
@@ -337,6 +339,15 @@ func destroy() {
func main() { func main() {
flag.Parse() 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) loggedRouter := handlers.LoggingHandler(os.Stdout, router)
srv := &http.Server{ srv := &http.Server{
Addr: *addr, Addr: *addr,