From ec15abc42d441bf88fc26b74130aa9bb598436dd Mon Sep 17 00:00:00 2001
From: crusader <crusader@loafle.com>
Date: Wed, 29 Nov 2017 14:46:47 +0900
Subject: [PATCH] ing

---
 discovery/discovery.go    |  4 ++++
 server/socket_handlers.go | 48 ++++++++++++++++++++++++++++++++++++---
 2 files changed, 49 insertions(+), 3 deletions(-)

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)
+}