overflow_cloud/member_gateway_rpc
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
geek 2018-04-13 18:07:14 +09:00
parent 9a50b95275
commit eaa0c12de7
2 changed files with 20 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
build.sh
View File

@ -1,7 +1,7 @@
#!/bin/bash #!/bin/bash
rm ./dist rm ./dist
CGO_ENABLED=0 go build -a --installsuffix cgo --ldflags="-s" -o ./_docker/bin/member_gateway_rpc CGO_ENABLED=0 go build -a --installsuffix cgo --ldflags="-s" -o ./dist/member_gateway_rpc
docker build -t docker.loafle.net/overflow/member_gateway_rpc:1.0.0 . docker build -t docker.loafle.net/overflow/member_gateway_rpc:1.0.0 .

65
servlet/webapp-servlet.go
View File

@ -12,6 +12,8 @@ import (
ogs "git.loafle.net/overflow/gateway/subscribe" ogs "git.loafle.net/overflow/gateway/subscribe"
ogrs "git.loafle.net/overflow/gateway_rpc/servlet" ogrs "git.loafle.net/overflow/gateway_rpc/servlet"
"git.loafle.net/overflow/member_gateway_rpc/subscribe" "git.loafle.net/overflow/member_gateway_rpc/subscribe"
"github.com/dgrijalva/jwt-go" "github.com/dgrijalva/jwt-go"
"github.com/satori/go.uuid" "github.com/satori/go.uuid"
"github.com/valyala/fasthttp" "github.com/valyala/fasthttp"
@ -27,7 +29,7 @@ type WebappServlets struct {
VerifyKey *rsa.PublicKey VerifyKey *rsa.PublicKey
SignKey *rsa.PrivateKey SignKey *rsa.PrivateKey
connections sync.Map sessions sync.Map
} }
func (s *WebappServlets) Init(serverCtx server.ServerCtx) error { func (s *WebappServlets) Init(serverCtx server.ServerCtx) error {
@ -103,7 +105,7 @@ func (s *WebappServlets) OnConnect(servletCtx server.ServletCtx, conn socket.Con
sessionID := servletCtx.GetAttribute(og.SessionIDKey) sessionID := servletCtx.GetAttribute(og.SessionIDKey)
targetID := servletCtx.GetAttribute(og.SessionTargetIDKey) targetID := servletCtx.GetAttribute(og.SessionTargetIDKey)
if nil != sessionID && nil != targetID { if nil != sessionID && nil != targetID {
s.connections.Store(sessionID.(string), retainConnection(targetID.(string), servletCtx)) s.sessions.Store(sessionID.(string), ogrs.RetainSession(targetID.(string), servletCtx))
} }
} }
@ -112,7 +114,7 @@ func (s *WebappServlets) OnDisconnect(servletCtx server.ServletCtx) {
sessionID := servletCtx.GetAttribute(og.SessionIDKey) sessionID := servletCtx.GetAttribute(og.SessionIDKey)
if nil != sessionID { if nil != sessionID {
s.connections.Delete(sessionID.(string)) s.sessions.Delete(sessionID.(string))
} }
} }
@ -126,13 +128,13 @@ func (s *WebappServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeCh
switch msg.TargetType { switch msg.TargetType {
case ogs.MEMBER: case ogs.MEMBER:
for _, targetID := range msg.Targets { for _, targetID := range msg.Targets {
_connections := s.getMemberConnections(targetID) _sessions := s.getMemberSessions(targetID)
if nil == _connections || 0 == len(_connections) { if nil == _sessions || 0 == len(_sessions) {
break break
} }
for _, _connection := range _connections { for _, _session := range _sessions {
_writeChan := _connection.servletCtx.GetAttribute(og.SessionWriteChanKey) _writeChan := _session.ServletCtx.GetAttribute(og.SessionWriteChanKey)
if nil != _writeChan { if nil != _writeChan {
writeChan := _writeChan.(chan<- []byte) writeChan := _writeChan.(chan<- []byte)
writeChan <- msg.Message writeChan <- msg.Message
@ -142,13 +144,13 @@ func (s *WebappServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeCh
case ogs.MEMBER_SESSION: case ogs.MEMBER_SESSION:
for _, sessionID := range msg.Targets { for _, sessionID := range msg.Targets {
__connection, ok := s.connections.Load(sessionID) _session, ok := s.sessions.Load(sessionID)
if !ok { if !ok {
logging.Logger().Debugf("Client[%s] is not exist", sessionID) logging.Logger().Debugf("Client[%s] is not exist", sessionID)
break break
} }
_connection := __connection.(*connection) session := _session.(*ogrs.Session)
_writeChan := _connection.servletCtx.GetAttribute(og.SessionWriteChanKey) _writeChan := session.ServletCtx.GetAttribute(og.SessionWriteChanKey)
if nil != _writeChan { if nil != _writeChan {
writeChan := _writeChan.(chan<- []byte) writeChan := _writeChan.(chan<- []byte)
writeChan <- msg.Message writeChan <- msg.Message
@ -161,45 +163,16 @@ func (s *WebappServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeCh
} }
} }
func (s *WebappServlets) getMemberConnections(targetID string) []*connection { func (s *WebappServlets) getMemberSessions(targetID string) []*ogrs.Session {
var connections []*connection var sessions []*ogrs.Session
s.connections.Range(func(k, v interface{}) bool { s.sessions.Range(func(k, v interface{}) bool {
_connection := v.(*connection) session := v.(*ogrs.Session)
if _connection.targetID == targetID { if session.TargetID == targetID {
connections = append(connections, _connection) sessions = append(sessions, session)
} }
return true return true
}) })
return connections return sessions
}
type connection struct {
targetID string
servletCtx server.ServletCtx
}
var connectionPool sync.Pool
func retainConnection(targetID string, servletCtx server.ServletCtx) *connection {
v := connectionPool.Get()
var _connection *connection
if v == nil {
_connection = &connection{}
} else {
_connection = v.(*connection)
}
_connection.targetID = targetID
_connection.servletCtx = servletCtx
return _connection
}
func releaseConnection(_connection *connection) {
_connection.targetID = ""
_connection.servletCtx = nil
connectionPool.Put(_connection)
} }