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

ing

Browse Source
This commit is contained in:
crusader 2018-07-01 01:44:44 +09:00
parent 1db82cb967
commit b7d6a6fcb3
2 changed files with 35 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
Gopkg.lock generated
View File

@ -17,7 +17,7 @@
branch = "master" branch = "master"
name = "git.loafle.net/commons/rpc-go" name = "git.loafle.net/commons/rpc-go"
packages = ["protocol"] packages = ["protocol"]
revision = "dcc3af07239b3f6fcbae3529bcb52c522b02053d" revision = "f5810183daf8aad248da6aab13fc5785ab984be4"
[[projects]] [[projects]]
branch = "master" branch = "master"
@ -27,7 +27,7 @@
"socket", "socket",
"socket/web" "socket/web"
] ]
revision = "1cae23cf7819d56fd399346a1dc30af6a82d1a93" revision = "d1d485c711ef8c59ad51fbf17dba80158f1a8ace"
[[projects]] [[projects]]
branch = "master" branch = "master"
@ -45,7 +45,7 @@
branch = "master" branch = "master"
name = "git.loafle.net/overflow/commons-go" name = "git.loafle.net/overflow/commons-go"
packages = ["config/external"] packages = ["config/external"]
revision = "687985e7f2122d12e43a0111f13d6df9921c3d82" revision = "7e2ff26648aefbafaee2ca28c1a958332ae4c6f9"
[[projects]] [[projects]]
branch = "master" branch = "master"
@ -99,8 +99,8 @@
"models", "models",
"pkg/escape" "pkg/escape"
] ]
revision = "89e084a80fb1e0bf5e7d38038e3367f821fdf3d7" revision = "4e4e00bc5ab85a3ff5e988c91020cf0399a87026"
version = "v1.5.3" version = "v1.5.4"
[[projects]] [[projects]]
name = "github.com/klauspost/compress" name = "github.com/klauspost/compress"
@ -109,8 +109,8 @@
"gzip", "gzip",
"zlib" "zlib"
] ]
revision = "6c8db69c4b49dd4df1fff66996cf556176d0b9bf" revision = "5fb1f31b0a61e9858f12f39266e059848a5f1cea"
version = "v1.2.1" version = "v1.3.0"
[[projects]] [[projects]]
name = "github.com/klauspost/cpuid" name = "github.com/klauspost/cpuid"
@ -118,12 +118,6 @@
revision = "ae7887de9fa5d2db4eaa8174a7eff2c1ac00f2da" revision = "ae7887de9fa5d2db4eaa8174a7eff2c1ac00f2da"
version = "v1.1" version = "v1.1"
[[projects]]
name = "github.com/klauspost/crc32"
packages = ["."]
revision = "cb6bfca970f6908083f26f39a79009d608efd5cd"
version = "v1.1"
[[projects]] [[projects]]
branch = "master" branch = "master"
name = "github.com/segmentio/kafka-go" name = "github.com/segmentio/kafka-go"
@ -176,7 +170,7 @@
"internal/timeseries", "internal/timeseries",
"trace" "trace"
] ]
revision = "db08ff08e8622530d9ed3a0e8ac279f6d4c02196" revision = "4cb1c02c05b0e749b0365f61ae859a8e0cfceed9"
[[projects]] [[projects]]
name = "golang.org/x/text" name = "golang.org/x/text"
@ -203,7 +197,7 @@
branch = "master" branch = "master"
name = "google.golang.org/genproto" name = "google.golang.org/genproto"
packages = ["googleapis/rpc/status"] packages = ["googleapis/rpc/status"]
revision = "32ee49c4dd805befd833990acba36cb75042378c" revision = "ff3583edef7de132f219f0efc00e097cabcc0ec0"
[[projects]] [[projects]]
name = "google.golang.org/grpc" name = "google.golang.org/grpc"
@ -212,15 +206,16 @@
"balancer", "balancer",
"balancer/base", "balancer/base",
"balancer/roundrobin", "balancer/roundrobin",
"channelz",
"codes", "codes",
"connectivity", "connectivity",
"credentials", "credentials",
"encoding", "encoding",
"encoding/proto", "encoding/proto",
"grpclb/grpc_lb_v1/messages",
"grpclog", "grpclog",
"internal", "internal",
"internal/backoff",
"internal/channelz",
"internal/grpcrand",
"keepalive", "keepalive",
"metadata", "metadata",
"naming", "naming",
@ -233,8 +228,8 @@
"tap", "tap",
"transport" "transport"
] ]
revision = "7a6a684ca69eb4cae85ad0a484f2e531598c047b" revision = "168a6198bcb0ef175f7dacec0b8691fc141dc9b8"
version = "v1.12.2" version = "v1.13.0"
[[projects]] [[projects]]
name = "gopkg.in/yaml.v2" name = "gopkg.in/yaml.v2"

37
servlet/rpc-servlet.go
View File

@ -58,20 +58,22 @@ func (s *RPCServlets) OnDisconnect(servletCtx server.ServletCtx) {
func (s *RPCServlets) Handle(servletCtx server.ServletCtx, func (s *RPCServlets) Handle(servletCtx server.ServletCtx,
stopChan <-chan struct{}, doneChan chan<- struct{}, stopChan <-chan struct{}, doneChan chan<- struct{},
readChan <-chan []byte, writeChan chan<- []byte) { readChan <-chan css.SocketMessage, writeChan chan<- css.SocketMessage) {
defer func() { defer func() {
doneChan <- struct{}{} doneChan <- struct{}{}
}() }()
var ( var (
md metadata.MD md metadata.MD
src crp.ServerRequestCodec src crp.ServerRequestCodec
method string messageType int
params []string message []byte
grpcCtx context.Context method string
grpcReply string params []string
replyBuff []byte grpcCtx context.Context
err error grpcReply string
replyBuff []byte
err error
) )
_clientType := servletCtx.GetAttribute(og.SessionClientTypeKey) _clientType := servletCtx.GetAttribute(og.SessionClientTypeKey)
@ -90,12 +92,15 @@ func (s *RPCServlets) Handle(servletCtx server.ServletCtx,
for { for {
select { select {
case msg, ok := <-readChan: case socketMessage, ok := <-readChan:
if !ok { if !ok {
return return
} }
messageType, message = socketMessage()
// grpc exec method call // grpc exec method call
src, err = s.RPCServerCodec.NewRequest(msg) src, err = s.RPCServerCodec.NewRequest(messageType, message)
if nil != err { if nil != err {
logging.Logger().Error(err) logging.Logger().Error(err)
break break
@ -118,14 +123,14 @@ func (s *RPCServlets) Handle(servletCtx server.ServletCtx,
if !src.HasResponse() { if !src.HasResponse() {
break break
} }
replyBuff, err = src.NewResponseWithString(grpcReply, err) messageType, replyBuff, err = src.NewResponseWithString(grpcReply, err)
if nil != err { if nil != err {
logging.Logger().Error(err) logging.Logger().Error(err)
s.writeError(src, writeChan, crp.E_INTERNAL, "", err) s.writeError(src, writeChan, crp.E_INTERNAL, "", err)
break break
} }
writeChan <- replyBuff writeChan <- css.MakeSocketMessage(messageType, replyBuff)
case <-stopChan: case <-stopChan:
return return
} }
@ -134,7 +139,7 @@ func (s *RPCServlets) Handle(servletCtx server.ServletCtx,
} }
func (s *RPCServlets) writeError(src crp.ServerRequestCodec, writeChan chan<- []byte, code crp.ErrorCode, message string, data interface{}) { func (s *RPCServlets) writeError(src crp.ServerRequestCodec, writeChan chan<- css.SocketMessage, code crp.ErrorCode, message string, data interface{}) {
if !src.HasResponse() { if !src.HasResponse() {
return return
} }
@ -145,12 +150,12 @@ func (s *RPCServlets) writeError(src crp.ServerRequestCodec, writeChan chan<- []
Data: data, Data: data,
} }
buf, err := src.NewResponse(nil, pErr) messageType, buf, err := src.NewResponse(nil, pErr)
if nil != err { if nil != err {
logging.Logger().Error(err) logging.Logger().Error(err)
return return
} }
writeChan <- buf writeChan <- css.MakeSocketMessage(messageType, buf)
} }
func (s *RPCServlets) GetSessions(sessionIDs []string) []*Session { func (s *RPCServlets) GetSessions(sessionIDs []string) []*Session {