ing

glide.yaml

@@ -9,3 +9,8 @@ import:
   subpackages:
   - server
   - servlet
+- package: git.loafle.net/overflow/gateway
+- package: github.com/dgrijalva/jwt-go
+  version: ^3.2.0
+- package: github.com/satori/go.uuid
+  version: ^1.2.0

main.go

@@ -1,22 +1,22 @@
 package main
 
 import (
-	"git.loafle.net/commons/logging-go"
-	"git.loafle.net/overflow/member_gateway_rpc/server"
+	"context"
+	"flag"
 	"log"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
-	"context"
-	"flag"
-	"git.loafle.net/overflow/member_gateway_rpc/config"
+
 	"git.loafle.net/commons/configuration-go"
+	"git.loafle.net/commons/logging-go"
+	"git.loafle.net/overflow/member_gateway_rpc/config"
+	"git.loafle.net/overflow/member_gateway_rpc/server"
 )
 
-
 var (
-	configDir     *string
+	configDir *string
 )
 
 func init() {
@@ -34,7 +34,6 @@ func main() {
 		logging.Logger().Panic(err)
 	}
 
-
 	s := server.New()
 
 	go func() {

server/server-handler.go

@@ -1,7 +1,9 @@
 package server
 
 import (
+	cs "git.loafle.net/commons/server-go"
 	ogrs "git.loafle.net/overflow/gateway_rpc/server"
+	"git.loafle.net/overflow/member_gateway_rpc/subscribe"
 )
 
 type ServerHandler interface {
@@ -10,4 +12,36 @@ type ServerHandler interface {
 
 type ServerHandlers struct {
 	ogrs.ServerHandlers
-}
+}
+
+func (sh *ServerHandlers) Init(serverCtx cs.ServerCtx) error {
+	if err := sh.ServerHandlers.Init(serverCtx); nil != err {
+		return err
+	}
+
+	subscribe.InitPackage()
+
+	return nil
+}
+
+func (sh *ServerHandlers) OnStart(serverCtx cs.ServerCtx) error {
+	if err := sh.ServerHandlers.OnStart(serverCtx); nil != err {
+		return err
+	}
+
+	subscribe.StartPackage()
+
+	return nil
+}
+
+func (sh *ServerHandlers) OnStop(serverCtx cs.ServerCtx) {
+	subscribe.StopPackage()
+
+	sh.ServerHandlers.OnStop(serverCtx)
+}
+
+func (sh *ServerHandlers) Destroy(serverCtx cs.ServerCtx) {
+	subscribe.DestroyPackage()
+
+	sh.ServerHandlers.Destroy(serverCtx)
+}

server/server.go

@@ -1,19 +1,18 @@
 package server
 
 import (
-	"git.loafle.net/commons/server-go/socket/web"
+	csgw "git.loafle.net/commons/server-go/socket/web"
 	"git.loafle.net/overflow/member_gateway_rpc/servlet"
 )
 
-func New() *web.Server {
-	rs := &servlet.RPCServlets{}
+func New() *csgw.Server {
+	ws := &servlet.WebappServlets{}
 	sh := &ServerHandlers{}
-	sh.RegisterServlet("/webapp", rs)
+	sh.RegisterServlet("/webapp", ws)
 
-	s := &web.Server{
+	s := &csgw.Server{
 		ServerHandler: sh,
 	}
 
 	return s
 }
-

servlet/rpc-servlet.go

@@ -1,12 +0,0 @@
-package servlet
-
-import (
-	ogrs "git.loafle.net/overflow/gateway_rpc/servlet"
-)
-type RPCServlet interface {
-	ogrs.RPCServlet
-}
-
-type RPCServlets struct {
-	ogrs.RPCServlets
-}

servlet/webapp-servlet.go

@@ -0,0 +1,163 @@
+package servlet
+
+import (
+	"crypto/rsa"
+	"fmt"
+	"io/ioutil"
+
+	logging "git.loafle.net/commons/logging-go"
+	"git.loafle.net/commons/server-go"
+	"git.loafle.net/commons/server-go/socket"
+	ogs "git.loafle.net/overflow/gateway/subscribe"
+	ogrs "git.loafle.net/overflow/gateway_rpc/servlet"
+	"git.loafle.net/overflow/member_gateway_rpc/subscribe"
+	"github.com/dgrijalva/jwt-go"
+	uuid "github.com/satori/go.uuid"
+	"github.com/valyala/fasthttp"
+)
+
+var (
+	verifyKey *rsa.PublicKey
+	signKey   *rsa.PrivateKey
+)
+
+func init() {
+	signBytes, err := ioutil.ReadFile("overFlow-private.key")
+	if nil != err {
+		panic(err)
+	}
+
+	signKey, err = jwt.ParseRSAPrivateKeyFromPEM(signBytes)
+	if nil != err {
+		panic(err)
+	}
+
+	verifyBytes, err := ioutil.ReadFile("overFlow-public.pem")
+	if nil != err {
+		panic(err)
+	}
+
+	verifyKey, err = jwt.ParseRSAPublicKeyFromPEM(verifyBytes)
+	if nil != err {
+		panic(err)
+	}
+}
+
+type WebappServlet interface {
+	ogrs.RPCServlet
+}
+
+type WebappServlets struct {
+	ogrs.RPCServlets
+}
+
+func (s *WebappServlets) Init(serverCtx server.ServerCtx) error {
+	if err := s.RPCServlets.Init(serverCtx); nil != err {
+		return err
+	}
+
+	return nil
+}
+
+func (s *WebappServlets) OnStart(serverCtx server.ServerCtx) error {
+	if err := s.RPCServlets.OnStart(serverCtx); nil != err {
+		return err
+	}
+
+	subscribeChan, err := subscribe.Subscriber.Subscribe("/webapp")
+	if nil != err {
+		return err
+	}
+	go s.handleSubscribe(serverCtx, subscribeChan)
+
+	return nil
+}
+
+func (s *WebappServlets) OnStop(serverCtx server.ServerCtx) {
+	if err := subscribe.Subscriber.Unsubscribe("/webapp"); nil != err {
+		logging.Logger().Warn(err)
+	}
+
+	s.RPCServlets.OnStop(serverCtx)
+}
+
+func (s *WebappServlets) Destroy(serverCtx server.ServerCtx) {
+
+	s.RPCServlets.Destroy(serverCtx)
+}
+
+func (s *WebappServlets) Handshake(servletCtx server.ServletCtx, ctx *fasthttp.RequestCtx) (*fasthttp.ResponseHeader, error) {
+	var ok bool
+
+	tokenString := string(ctx.QueryArgs().Peek("authToken"))
+	token, err := jwt.Parse(tokenString, func(token *jwt.Token) (interface{}, error) {
+		// Don't forget to validate the alg is what you expect:
+		if _, ok := token.Method.(*jwt.SigningMethodRSA); !ok {
+			return nil, fmt.Errorf("Webapp: Unexpected signing method: %v", token.Header["alg"])
+		}
+		// hmacSampleSecret is a []byte containing your secret, e.g. []byte("my_secret_key")
+		return verifyKey, nil
+	})
+
+	if nil != err {
+		return nil, nil
+	}
+
+	var claims jwt.MapClaims
+	if claims, ok = token.Claims.(jwt.MapClaims); !ok || !token.Valid {
+		return nil, fmt.Errorf("Token is not valid %v", token)
+	}
+
+	userEmail := claims["sub"].(string)
+	sessionID := uuid.NewV4().String()
+
+	logging.Logger().Debugf("%s, %s", userEmail, sessionID)
+
+	// servletCtx.SetAttribute(oogw.ClientTypeKey, ogs.MEMBER)
+	// servletCtx.SetAttribute(oogw.TargetIDKey, userEmail)
+
+	return nil, nil
+}
+
+func (s *WebappServlets) OnConnect(servletCtx server.ServletCtx, conn socket.Conn) {
+	//
+}
+
+func (s *WebappServlets) OnDisconnect(servletCtx server.ServletCtx) {
+	//
+}
+
+func (s *WebappServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeChan <-chan *ogs.Message) {
+	// for {
+	// 	select {
+	// 	case msg, ok := <-subscribeChan:
+	// 		switch msg.TargetType {
+	// 		case ogs.MEMBER:
+	// 			for _, uid := range msg.Targets {
+	// 				sockets := sh.getMemberSockets(uid)
+	// 				if nil == sockets {
+	// 					continue
+	// 				}
+
+	// 				for _, soc := range sockets {
+	// 					if err := soc.WriteMessageUseWriter(websocket.TextMessage, message.Message); nil != err {
+	// 						logging.Logger().Errorf("OnSubscribe: write error %v", err)
+	// 					}
+	// 				}
+	// 			}
+
+	// 		case oos.MEMBER_SESSION:
+	// 			for _, sid := range message.Targets {
+	// 				if soc := sh.GetSocket(sid); nil != soc {
+	// 					if err := soc.WriteMessageUseWriter(websocket.TextMessage, message.Message); nil != err {
+	// 						logging.Logger().Errorf("OnSubscribe: write error %v", err)
+	// 					}
+	// 				}
+	// 			}
+	// 		default:
+	// 			logging.Logger().Warnf("[WEBAPP] SubscriberHandler: Unknown TargetType %s", message.TargetType)
+	// 		}
+	// 	}
+	// }
+
+}

subscribe/subscribe.go

@@ -0,0 +1,32 @@
+package subscribe
+
+import (
+	"git.loafle.net/commons/logging-go"
+	oger "git.loafle.net/overflow/gateway/external/redis"
+	ogsr "git.loafle.net/overflow/gateway/subscribe/redis"
+)
+
+var Subscriber *ogsr.Subscribers
+
+func InitPackage() {
+	Subscriber = &ogsr.Subscribers{}
+}
+
+func StartPackage() {
+	Subscriber.Conn = oger.Pool.Get()
+	if err := Subscriber.Start(); nil != err {
+		logging.Logger().Panic(err)
+	}
+}
+
+func StopPackage() {
+	if nil != Subscriber {
+		if err := Subscriber.Stop(); nil != err {
+			logging.Logger().Error(err)
+		}
+	}
+}
+
+func DestroyPackage() {
+	Subscriber = nil
+}