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

ing

Browse Source
This commit is contained in:
crusader 2017-12-05 15:09:04 +09:00
parent 8d60284f54
commit 89fe1b886e
9 changed files with 29 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
server/rpc_servlet_handler.go
View File

@ -1,9 +0,0 @@
package server
import (
"git.loafle.net/commons_go/rpc/server"
)
type RPCServletHandler interface {
server.ServletHandler
}

22
server/rpc_servlet_handlers.go
View File

@ -1,22 +0,0 @@
package server
import (
crr "git.loafle.net/commons_go/rpc/registry"
"git.loafle.net/commons_go/rpc/server"
)
func newRPCServletHandler(rpcInvoker crr.RPCInvoker) server.ServletHandler {
sh := &RPCServletHandlers{}
sh.RPCInvoker = rpcInvoker
return sh
}
type RPCServletHandlers struct {
server.ServletHandlers
}
func (sh *RPCServletHandlers) Validate() {
sh.ServletHandlers.Validate()
}

9
server/server.go
View File

@ -1,10 +1,9 @@
package server
import (
crpj "git.loafle.net/commons_go/rpc/protocol/json"
crr "git.loafle.net/commons_go/rpc/registry"
"git.loafle.net/commons_go/server"
oopcs "git.loafle.net/overflow/overflow_probe_container/server"
"git.loafle.net/overflow/overflow_probe_container_network/service"
)
@ -12,13 +11,11 @@ func New(addr string) server.Server {
rpcRegistry := crr.NewRPCRegistry()
service.RegisterRPC(rpcRegistry)
rpcSH := newRPCServletHandler(rpcRegistry)
rpcSH.RegisterCodec(crpj.Name, crpj.NewServerCodec())
rpcSH := oopcs.NewRPCServletHandler(rpcRegistry)
socketHandler := newSocketHandler(rpcSH)
sh := newServerHandler(addr, socketHandler)
s := server.New(sh)
s := oopcs.New(sh)
return s
}

6
server/server_handler.go
View File

@ -1,7 +1,9 @@
package server
import "git.loafle.net/commons_go/server"
import (
oopcs "git.loafle.net/overflow/overflow_probe_container/server"
)
type ServerHandler interface {
server.ServerHandler
oopcs.ServerHandler
}

28
server/server_handlers.go
View File

@ -1,31 +1,23 @@
package server
import (
"fmt"
"git.loafle.net/commons_go/logging"
"git.loafle.net/commons_go/server"
oopcs "git.loafle.net/overflow/overflow_probe_container/server"
)
func newServerHandler(addr string, socketHandler SocketHandler) ServerHandler {
sh := &ServerHandlers{
addr: addr,
}
sh := &ServerHandlers{}
sh.ServerHandler = oopcs.NewServerHandler(addr, "Discovery Containter", socketHandler)
sh.Name = "Network Container"
sh.SocketHandler = socketHandler
return sh
}
type ServerHandlers struct {
server.ServerHandlers
addr string
oopcs.ServerHandler
}
func (sh *ServerHandlers) Init(serverCTX server.ServerContext) error {
if err := sh.ServerHandlers.Init(serverCTX); nil != err {
if err := sh.ServerHandler.Init(serverCTX); nil != err {
return err
}
@ -33,20 +25,16 @@ func (sh *ServerHandlers) Init(serverCTX server.ServerContext) error {
}
func (sh *ServerHandlers) OnStart(serverCTX server.ServerContext) {
sh.ServerHandlers.OnStart(serverCTX)
sh.ServerHandler.OnStart(serverCTX)
}
func (sh *ServerHandlers) OnStop(serverCTX server.ServerContext) {
sh.ServerHandlers.OnStop(serverCTX)
sh.ServerHandler.OnStop(serverCTX)
}
func (sh *ServerHandlers) Validate() {
sh.ServerHandlers.Validate()
if "" == sh.addr {
logging.Logger().Panic(fmt.Sprintf("Server: Address of server must be specified"))
}
sh.ServerHandler.Validate()
}

17
server/server_handlers_unix.go
View File

@ -1,17 +0,0 @@
package server
import (
"net"
"os"
"git.loafle.net/commons_go/server"
)
func (sh *ServerHandlers) Listen(serverCTX server.ServerContext) (net.Listener, error) {
os.Remove(sh.addr)
l, err := net.ListenUnix("unix", &net.UnixAddr{Name: sh.addr, Net: "unix"})
if nil == err {
os.Chmod(sh.addr, 0777)
}
return l, err
}

16
server/server_handlers_windows.go
View File

@ -1,16 +0,0 @@
package server
import (
"net"
"git.loafle.net/commons_go/server"
npipe "gopkg.in/natefinch/npipe.v2"
)
func (sh *ServerHandlers) Listen(serverCTX server.ServerContext) (net.Listener, error) {
ln, err := npipe.Listen(`\\.\pipe\` + sh.addr)
if err != nil {
// handle error
}
return ln, err
}

4
server/socket_handler.go
View File

@ -1,9 +1,9 @@
package server
import (
"git.loafle.net/commons_go/server"
oopcs "git.loafle.net/overflow/overflow_probe_container/server"
)
type SocketHandler interface {
server.SocketHandler
oopcs.SocketHandler
}

73
server/socket_handlers.go
View File

@ -2,35 +2,24 @@ package server
import (
"net"
"sync"
cRPC "git.loafle.net/commons_go/rpc"
crsrs "git.loafle.net/commons_go/rpc/server/rwc/socket"
"git.loafle.net/commons_go/rpc/protocol/json"
"git.loafle.net/commons_go/server"
oopcs "git.loafle.net/overflow/overflow_probe_container/server"
)
func newSocketHandler(rpcSH RPCServletHandler) SocketHandler {
rpcRWCSH := crsrs.New()
sh := &SocketHandlers{
rpcSH: rpcSH,
rpcRWCSH: rpcRWCSH,
}
func newSocketHandler(rpcSH oopcs.RPCServletHandler) SocketHandler {
sh := &SocketHandlers{}
sh.SocketHandler = oopcs.NewSocketHandler(rpcSH)
return sh
}
type SocketHandlers struct {
server.SocketHandlers
rpcRWCSH cRPC.ServletReadWriteCloseHandler
rpcSH RPCServletHandler
oopcs.SocketHandler
}
func (sh *SocketHandlers) Init(serverCTX server.ServerContext) error {
if err := sh.SocketHandlers.Init(serverCTX); nil != err {
if err := sh.SocketHandler.Init(serverCTX); nil != err {
return err
}
@ -42,60 +31,22 @@ func (sh *SocketHandlers) Handshake(socketCTX server.SocketContext, conn net.Con
}
func (sh *SocketHandlers) OnConnect(soc server.Socket) {
sh.SocketHandlers.OnConnect(soc)
sh.SocketHandler.OnConnect(soc)
soc.Context().SetAttribute(cRPC.ContentTypeKey, json.Name)
}
func (sh *SocketHandlers) Handle(soc server.Socket, stopChan <-chan struct{}, doneChan chan<- error) {
var err error
rpcServlet := retainRPCServlet(sh.rpcSH, sh.rpcRWCSH)
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()
<-rpcDoneChan
}
// discovery.RPCServlet = soc.Context().GetAttribute(oopc.RPCServletKey).(cRPC.Servlet)
}
func (sh *SocketHandlers) OnDisconnect(soc server.Socket) {
// discovery.RPCServlet = nil
sh.SocketHandlers.OnDisconnect(soc)
sh.SocketHandler.OnDisconnect(soc)
}
func (sh *SocketHandlers) Destroy() {
sh.SocketHandlers.Destroy()
sh.SocketHandler.Destroy()
}
func (sh *SocketHandlers) Validate() {
}
var rpcServletPool sync.Pool
func retainRPCServlet(sh RPCServletHandler, rpcRWCSH cRPC.ServletReadWriteCloseHandler) cRPC.Servlet {
v := rpcServletPool.Get()
if v == nil {
return cRPC.NewServlet(sh, rpcRWCSH)
}
return v.(cRPC.Servlet)
}
func releaseRPCServlet(s cRPC.Servlet) {
rpcServletPool.Put(s)
sh.SocketHandler.Validate()
}