overflow_backup_2018_0455/overflow_discovery
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2017-11-29 14:46:47 +09:00
parent 2b154aeb87
commit ec15abc42d
2 changed files with 49 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
discovery/discovery.go
View File

@ -5,10 +5,13 @@ import (
"sync" "sync"
"git.loafle.net/commons_go/logging" "git.loafle.net/commons_go/logging"
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/util/net/cidr" "git.loafle.net/commons_go/util/net/cidr"
"git.loafle.net/overflow/overflow_discovery/api/module/discovery/model" "git.loafle.net/overflow/overflow_discovery/api/module/discovery/model"
) )
var RPCServlet rpc.Servlet
var discoverer *discovery var discoverer *discovery
func DiscoveryInit() { func DiscoveryInit() {
@ -140,6 +143,7 @@ func (d *discovery) discoverService(port *model.Port, ds *model.DiscoveryService
} }
func (d *discovery) sendResult(method string, args ...interface{}) { func (d *discovery) sendResult(method string, args ...interface{}) {
go RPCServlet.Send(method, args...)
// go notify.Notifier.Notify(method, args...) // go notify.Notifier.Notify(method, args...)
} }

48
server/socket_handlers.go
View File

@ -2,8 +2,12 @@ package server
import ( import (
"net" "net"
"sync"
cRPC "git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol/json"
"git.loafle.net/commons_go/server" "git.loafle.net/commons_go/server"
"git.loafle.net/overflow/overflow_discovery/discovery"
) )
func newSocketHandler(rpcSH RPCServletHandler) SocketHandler { func newSocketHandler(rpcSH RPCServletHandler) SocketHandler {
@ -29,15 +33,38 @@ func (sh *SocketHandlers) Init(serverCTX server.ServerContext) error {
} }
func (sh *SocketHandlers) Handshake(serverCTX server.ServerContext, conn net.Conn) (id string) { func (sh *SocketHandlers) Handshake(serverCTX server.ServerContext, conn net.Conn) (id string) {
return "" return "discovery"
} }
func (sh *SocketHandlers) OnConnect(soc server.Socket) { func (sh *SocketHandlers) OnConnect(soc server.Socket) {
// no op sh.SocketHandlers.OnConnect(soc)
soc.Context().SetAttribute(cRPC.ContentTypeKey, json.Name)
} }
func (sh *SocketHandlers) Handle(soc server.Socket, stopChan <-chan struct{}, doneChan chan<- error) { func (sh *SocketHandlers) Handle(soc server.Socket, stopChan <-chan struct{}, doneChan chan<- error) {
// no op var err error
rpcServlet := retainRPCServlet(sh.rpcSH)
discovery.RPCServlet = rpcServlet
defer func() {
releaseRPCServlet(rpcServlet)
doneChan <- err
}()
rpcDoneChan := make(chan error, 1)
if err = rpcServlet.Start(soc.Context(), soc, rpcDoneChan); nil != err {
return
}
select {
case err = <-rpcDoneChan:
case <-stopChan:
rpcServlet.Stop()
}
} }
func (sh *SocketHandlers) OnDisconnect(soc server.Socket) { func (sh *SocketHandlers) OnDisconnect(soc server.Socket) {
@ -53,3 +80,18 @@ func (sh *SocketHandlers) Destroy() {
func (sh *SocketHandlers) Validate() { func (sh *SocketHandlers) Validate() {
} }
var rpcServletPool sync.Pool
func retainRPCServlet(sh RPCServletHandler) cRPC.Servlet {
v := rpcServletPool.Get()
if v == nil {
return cRPC.NewServlet(sh)
}
return v.(cRPC.Servlet)
}
func releaseRPCServlet(s cRPC.Servlet) {
rpcServletPool.Put(s)
}