ing

config.json

@@ -1,7 +1,11 @@
 {
   "server": {
+    "name": "Gateway APP",
+    "network": "tcp4",
     "addr": ":19090",
-    "tls": false
+    "tls": false,
+    "concurrency": 262144,
+    "maxStopWaitTime": 0
   },
   "auth": {
     "signingKey": "tWB0lUXiCwX4U3qsJZcZ10mKvEH793RHkTJDbDuZVshQTk4uNB6ck59UQ96lhsRi4XNUiEnlIbP8XYQMPabeNtERX3iyHeDcwocgUVAor1nkAajYeq1gNyJszGpMhEOT"

glide.yaml

@@ -16,3 +16,4 @@ import:
   - redis
 - package: google.golang.org/grpc
   version: v1.7.2
+- package: git.loafle.net/overflow/overflow_subscriber

server/server.go

@@ -7,11 +7,18 @@ import (
 
 func New() cwf.Server {
 	sh := newServerHandler()
-	wsh := web.NewSocketHandler()
-
-	sh.RegisterSocketHandler("/web", wsh)
+	registerWebServlet(sh)
 
 	s := cwf.New(sh)
 
 	return s
 }
+
+func registerWebServlet(sh ServerHandler) {
+	webEntry, wsh := web.NewSocketHandler()
+	wsubh := web.NewSubscriberHandler(webEntry)
+
+	sh.RegisterSocketHandler(webEntry, wsh)
+	sh.RegisterSubscriberHandler(wsubh)
+
+}

server/server_handlers.go

@@ -1,24 +1,59 @@
 package server
 
 import (
+	"fmt"
 	"net"
 
+	"git.loafle.net/commons_go/logging"
 	cwf "git.loafle.net/commons_go/websocket_fasthttp"
+	"git.loafle.net/overflow/overflow_gateway_app/conf"
+	"git.loafle.net/overflow/overflow_gateway_app/external/redis"
+	oos "git.loafle.net/overflow/overflow_subscriber"
+	oosr "git.loafle.net/overflow/overflow_subscriber/redis"
 	"github.com/valyala/fasthttp"
 )
 
-func newServerHandler() cwf.ServerHandler {
+func newServerHandler() ServerHandler {
 	sh := &ServerHandlers{}
+	sh.Name = conf.Config.Server.Name
+	sh.Concurrency = conf.Config.Server.Concurrency
+	sh.MaxStopWaitTime = conf.Config.Server.MaxStopWaitTime
+
+	sh.HandshakeTimeout = conf.Config.Websocket.HandshakeTimeout
+	sh.ReadBufferSize = conf.Config.Websocket.ReadBufferSize
+	sh.WriteBufferSize = conf.Config.Websocket.WriteBufferSize
+	sh.EnableCompression = conf.Config.Websocket.EnableCompression
 
 	return sh
 }
 
+type ServerHandler interface {
+	cwf.ServerHandler
+
+	RegisterSubscriberHandler(subscriberHandler oos.SubscriberHandler)
+}
+
 type ServerHandlers struct {
 	cwf.ServerHandlers
+
+	redisSubscriber oos.Subscriber
+	subscribers     []oos.SubscriberHandler
+}
+
+func (sh *ServerHandlers) OnStart() {
+	sh.redisSubscriber = oosr.New(redis.RedisPool.Get())
+	if err := sh.redisSubscriber.Start(); nil != err {
+		logging.Logger().Error(fmt.Sprintf("App: Redist Subscriber did not start %v", err))
+		return
+	}
+
+	for _, subscriber := range sh.subscribers {
+		sh.redisSubscriber.Subscribe(subscriber)
+	}
 }
 
 func (sh *ServerHandlers) Listen() (net.Listener, error) {
-	return net.Listen("tcp4", ":19090")
+	return net.Listen(conf.Config.Server.Network, conf.Config.Server.Addr)
 }
 
 func (sh *ServerHandlers) CheckOrigin(ctx *fasthttp.RequestCtx) bool {
@@ -36,10 +71,13 @@ func (sh *ServerHandlers) OnError(ctx *fasthttp.RequestCtx, status int, reason e
 	// no op
 }
 
-func (sh *ServerHandlers) OnStart() {
-	// no op
+func (sh *ServerHandlers) OnStop() {
+	sh.redisSubscriber.Stop()
 }
 
-func (sh *ServerHandlers) OnStop() {
-	// no op
+func (sh *ServerHandlers) RegisterSubscriberHandler(subscriberHandler oos.SubscriberHandler) {
+	if nil == sh.subscribers {
+		sh.subscribers = make([]oos.SubscriberHandler, 0)
+	}
+	sh.subscribers = append(sh.subscribers, subscriberHandler)
 }

servlet/web/socket_handlers.go

@@ -3,23 +3,39 @@ package web
 import (
 	"context"
 	"log"
+	"time"
 
 	"git.loafle.net/commons_go/rpc/gateway"
 	"git.loafle.net/commons_go/server"
 	cwf "git.loafle.net/commons_go/websocket_fasthttp"
+	"git.loafle.net/overflow/overflow_gateway_app/conf"
 	"github.com/valyala/fasthttp"
 )
 
-func NewSocketHandler() cwf.SocketHandler {
+func NewSocketHandler() (entry string, socketHandler cwf.SocketHandler) {
 	sh := &SocketHandlers{}
+
+	cfg := conf.Config.Handlers["web"]
+
+	sh.entryPath = cfg.Entry
+
+	sh.MaxMessageSize = cfg.Socket.MaxMessageSize
+	sh.WriteTimeout = cfg.Socket.WriteTimeout * time.Second
+	sh.ReadTimeout = cfg.Socket.ReadTimeout * time.Second
+	sh.PongTimeout = cfg.Socket.PongTimeout * time.Second
+	sh.PingTimeout = cfg.Socket.PingTimeout * time.Second
+	sh.PingPeriod = cfg.Socket.PingPeriod * time.Second
+
 	sh.rpcServerHandler = newRCPServerHandler()
 
-	return sh
+	return sh.entryPath, sh
 }
 
 type SocketHandlers struct {
 	cwf.SocketHandlers
 
+	entryPath string
+
 	rpcServerHandler gateway.ServerHandler
 }
 
@@ -27,6 +43,10 @@ func (sh *SocketHandlers) Handshake(ctx *fasthttp.RequestCtx) (interface{}, *fas
 	return "overflow@loafle.com", nil
 }
 
+func (sh *SocketHandlers) OnConnect(soc *cwf.Socket) {
+	socketManager.putSocket(soc.ID().(string), soc)
+}
+
 func (sh *SocketHandlers) Handle(soc *cwf.Socket, stopChan <-chan struct{}, doneChan chan<- struct{}) {
 	contentType := "json"
 	codec, err := sh.rpcServerHandler.GetCodec(contentType)
@@ -75,6 +95,10 @@ func (sh *SocketHandlers) Handle(soc *cwf.Socket, stopChan <-chan struct{}, done
 
 }
 
+func (sh *SocketHandlers) OnDisconnect(soc *cwf.Socket) {
+	socketManager.removeSocket(soc.ID().(string))
+}
+
 func (sh *SocketHandlers) Validate() {
 	sh.SocketHandlers.Validate()
 
@@ -82,3 +106,27 @@ func (sh *SocketHandlers) Validate() {
 		panic("RPCServerHandler must be specified.")
 	}
 }
+
+var socketManager webSocketManager
+
+type webSocketManager struct {
+	sockets map[string]*cwf.Socket
+}
+
+func (sm *webSocketManager) putSocket(id string, soc *cwf.Socket) {
+	if nil == sm.sockets {
+		sm.sockets = make(map[string]*cwf.Socket)
+	}
+	sm.sockets[id] = soc
+}
+
+func (sm *webSocketManager) removeSocket(id string) {
+	delete(sm.sockets, id)
+}
+
+func (sm *webSocketManager) getSocket(id string) *cwf.Socket {
+	if nil == sm.sockets {
+		return nil
+	}
+	return sm.sockets[id]
+}

servlet/web/subscriber_handlers.go

@@ -0,0 +1,29 @@
+package web
+
+import (
+	"git.loafle.net/commons_go/websocket_fasthttp/websocket"
+	oos "git.loafle.net/overflow/overflow_subscriber"
+)
+
+func NewSubscriberHandler(channel string) oos.SubscriberHandler {
+	sh := &webSubscriberHandlers{}
+	sh.Channel = channel
+
+	return sh
+}
+
+type webSubscriberHandlers struct {
+	oos.SubscriberHandlers
+}
+
+func (sh *webSubscriberHandlers) OnSubscribe(channel string, message oos.SubscribeMessage) {
+	for _, uid := range message.Targets {
+		if soc := socketManager.getSocket(uid); nil != soc {
+			soc.WriteMessage(websocket.TextMessage, message.Message)
+		}
+	}
+}
+
+func (sh *webSubscriberHandlers) Validate() {
+	sh.SubscriberHandlers.Validate()
+}