overflow_cloud/probe_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:35 +09:00
parent d9f54c3ad4
commit 5d6cf0e870
2 changed files with 25 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
servlet/auth-servlet.go
View File

@ -27,7 +27,7 @@ type AuthServlet interface {
type AuthServlets struct { type AuthServlets struct {
ogrs.RPCServlets ogrs.RPCServlets
connections sync.Map sessions sync.Map
} }
func (s *AuthServlets) Init(serverCtx server.ServerCtx) error { func (s *AuthServlets) Init(serverCtx server.ServerCtx) error {
@ -132,7 +132,7 @@ func (s *AuthServlets) OnConnect(servletCtx server.ServletCtx, conn socket.Conn)
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))
} }
} }
@ -141,7 +141,7 @@ func (s *AuthServlets) 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))
} }
} }
@ -156,13 +156,13 @@ func (s *AuthServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeChan
switch msg.TargetType { switch msg.TargetType {
case ogs.PROBE: case ogs.PROBE:
for _, targetID := range msg.Targets { for _, targetID := range msg.Targets {
_connections := s.getProbeConnections(targetID) _sessions := s.getAuthSessions(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
@ -175,34 +175,16 @@ func (s *AuthServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeChan
} }
} }
func (s *AuthServlets) getProbeConnections(targetID string) []*connection { func (s *AuthServlets) getAuthSessions(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 {
return nil
}
func releaseConnection(_connection *connection) {
_connection.targetID = ""
_connection.servletCtx = nil
connectionPool.Put(_connection)
} }

40
servlet/probe-servlet.go
View File

@ -113,7 +113,7 @@ func (s *ProbeServlets) OnConnect(servletCtx server.ServletCtx, conn socket.Conn
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.connections.Store(sessionID.(string), ogrs.RetainSession(targetID.(string), servletCtx))
} }
} }
@ -137,13 +137,13 @@ func (s *ProbeServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeCha
switch msg.TargetType { switch msg.TargetType {
case ogs.PROBE: case ogs.PROBE:
for _, targetID := range msg.Targets { for _, targetID := range msg.Targets {
_connections := s.getProbeConnections(targetID) _sessions := s.getProbeSessions(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
@ -155,34 +155,16 @@ func (s *ProbeServlets) handleSubscribe(serverCtx server.ServerCtx, subscribeCha
} }
} }
func (s *ProbeServlets) getProbeConnections(targetID string) []*connection { func (s *ProbeServlets) getProbeSessions(targetID string) []*ogrs.Session {
var connections []*connection var sessions []*ogrs.Session
s.connections.Range(func(k, v interface{}) bool { s.connections.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 {
// return nil
//}
//
//func releaseConnection(_connection *connection) {
// _connection.targetID = ""
// _connection.servletCtx = nil
//
// connectionPool.Put(_connection)
//}