add tcp_keepalive config for tcp option SO_KEEPALIVE

Author: Terry-Mao

config.go

@@ -33,6 +33,7 @@ type Config struct {
 	RedisMQSize         int    `json:"redis_mqsize"`
 	MaxProcs            int    `json:"max_procs"`
 	MaxSubscriberPerKey int    `json:"max_subscriber_per_key"`
+	TCPKeepAlive        int    `json:"tcp_keepalive"`
 }
 
 func InitConfig(file string) (*Config, error) {
@@ -60,6 +61,7 @@ func InitConfig(file string) (*Config, error) {
 		RedisMQSize:         20,
 		MaxSubscriberPerKey: 0, // no limit
 		MaxProcs:            runtime.NumCPU(),
+		TCPKeepAlive:        1,
 	}
 	if err = json.Unmarshal(c, cf); err != nil {
 		Log.Printf("json.Unmarshal(\"%s\", cf) failed (%s)", string(c), err.Error())

gopush.conf

@@ -15,5 +15,6 @@
   "redis_poolsize": 50,
   "redis_mqsize": 20,
   "max_procs": 4,
-  "max_subscriber_per_key": 3
+  "max_subscriber_per_key": 3,
+  "tcp_keepalive": 1
 }

listen.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+)
+
+type KeepAliveListener struct {
+	net.Listener
+}
+
+func (l *KeepAliveListener) Accept() (c net.Conn, err error) {
+	c, err = l.Listener.Accept()
+	if err != nil {
+		Log.Printf("Listener.Accept() failed (%s)", err.Error())
+		return
+	}
+
+	// set keepalive
+	if tc, ok := c.(*net.TCPConn); !ok {
+		panic("Assection type failed c.(net.TCPConn)")
+	} else {
+		err = tc.SetKeepAlive(true)
+		if err != nil {
+			Log.Printf("tc.SetKeepAlive(true) failed (%s)", err.Error())
+			return
+		}
+	}
+
+	return
+}
+
+func Listen() error {
+	addr := fmt.Sprintf("%s:%d", Conf.Addr, Conf.Port)
+	if Conf.TCPKeepAlive == 1 {
+		server := &http.Server{}
+		l, err := net.Listen("tcp", addr)
+		if err != nil {
+			Log.Printf("net.Listen(\"tcp\", \"%s\") failed (%s)", addr, err.Error())
+			return err
+		}
+
+		return server.Serve(&KeepAliveListener{Listener: l})
+	} else {
+		if err := http.ListenAndServe(addr, nil); err != nil {
+			Log.Printf("http.ListenAdServe(\"%s\") failed (%s)", addr, err.Error())
+			return err
+		}
+	}
+	// nerve here
+	return nil
+}

main.go

@@ -80,10 +80,8 @@ func main() {
 		http.HandleFunc("/pub", Publish)
 	}
 	// start listen and pending here
-	if err := http.ListenAndServe(fmt.Sprintf("%s:%d", Conf.Addr, Conf.Port), nil); err != nil {
-		Log.Printf("http.ListenAdServe(\"%s:%d\") failed (%s)", Conf.Addr, Conf.Port, err.Error())
+	if err = Listen(); err != nil {
+		Log.Printf("Listen() failed (%s)", err.Error())
 		return
 	}
-
-	Log.Printf("gopush service exit.")
 }

push.go

@@ -22,9 +22,9 @@ const (
 )
 
 var (
-	errMsg = map[int]string{}
-	pusher Pusher
-    protocolErr = errors.New("client must not send any data")
+	errMsg      = map[int]string{}
+	pusher      Pusher
+	protocolErr = errors.New("client must not send any data")
 )
 
 type Pusher interface {
@@ -109,7 +109,7 @@ func Subscribe(ws *websocket.Conn) {
 		return
 	}
 	// Generate Key
-    key = pusher.Key(key)
+	key = pusher.Key(key)
 	Log.Printf("Client (%s) subscribe to key %s", ws.Request().RemoteAddr, key)
 	// check multi cli connected
 	if Conf.MaxSubscriberPerKey > 0 {