diff --git a/discovery/discovery.go b/discovery/discovery.go index 1d2bd8d..9037645 100644 --- a/discovery/discovery.go +++ b/discovery/discovery.go @@ -5,10 +5,13 @@ import ( "sync" "git.loafle.net/commons_go/logging" + "git.loafle.net/commons_go/rpc" "git.loafle.net/commons_go/util/net/cidr" "git.loafle.net/overflow/overflow_discovery/api/module/discovery/model" ) +var RPCServlet rpc.Servlet + var discoverer *discovery func DiscoveryInit() { @@ -140,6 +143,7 @@ func (d *discovery) discoverService(port *model.Port, ds *model.DiscoveryService } func (d *discovery) sendResult(method string, args ...interface{}) { + go RPCServlet.Send(method, args...) // go notify.Notifier.Notify(method, args...) } diff --git a/server/socket_handlers.go b/server/socket_handlers.go index 8377cac..cf3d4c3 100644 --- a/server/socket_handlers.go +++ b/server/socket_handlers.go @@ -2,8 +2,12 @@ package server import ( "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/overflow/overflow_discovery/discovery" ) 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) { - return "" + return "discovery" } 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) { - // 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) { @@ -53,3 +80,18 @@ func (sh *SocketHandlers) Destroy() { 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) +}