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-12-05 14:58:32 +09:00
parent 84a1d91c52
commit b30f3d10ea
16 changed files with 30 additions and 277 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
client/client.go
View File

@ -1,15 +0,0 @@
package client
import (
crc "git.loafle.net/commons_go/rpc/client"
crcrs "git.loafle.net/commons_go/rpc/client/rwc/socket"
csc "git.loafle.net/commons_go/server/client"
)
func New(clientHandler ClientHandler, socketBuilder csc.SocketBuilder) crc.Client {
cRWCHandler := crcrs.New(socketBuilder)
c := crc.New(clientHandler, cRWCHandler)
return c
}

7
client/client_handler.go
View File

@ -1,7 +0,0 @@
package client
import "git.loafle.net/commons_go/rpc/client"
type ClientHandler interface {
client.ClientHandler
}

19
client/client_handlers.go
View File

@ -1,19 +0,0 @@
package client
import (
crc "git.loafle.net/commons_go/rpc/client"
"git.loafle.net/commons_go/rpc/protocol/json"
crr "git.loafle.net/commons_go/rpc/registry"
)
func NewClientHandler(rpcInvoker crr.RPCInvoker) ClientHandler {
ch := &ClientHandlers{}
ch.RPCInvoker = rpcInvoker
ch.Codec = json.NewClientCodec()
return ch
}
type ClientHandlers struct {
crc.ClientHandlers
}

17
client/socket_builders.go
View File

@ -1,17 +0,0 @@
package client
import (
csc "git.loafle.net/commons_go/server/client"
)
func NewSocketBuilder(address string) csc.SocketBuilder {
return newSocketBuilder(address)
}
type SocketBuilders struct {
csc.SocketBuilders
}
func (sb *SocketBuilders) SocketHandler() csc.SocketHandler {
return NewSocketHandler()
}

13
client/socket_builders_unix.go
View File

@ -1,13 +0,0 @@
package client
import (
csc "git.loafle.net/commons_go/server/client"
)
func newSocketBuilder(address string) csc.SocketBuilder {
sb := &SocketBuilders{}
sb.Network = "unix"
sb.Address = address
return sb
}

23
client/socket_builders_windows.go
View File

@ -1,23 +0,0 @@
package client
import (
"net"
csc "git.loafle.net/commons_go/server/client"
"gopkg.in/natefinch/npipe.v2"
)
func newSocketBuilder(address string) csc.SocketBuilder {
sb := &SocketBuilders{}
sb.Network = "pipe"
sb.Address = address
return sb
}
func (sb *SocketBuilders) Dial(network, address string) (net.Conn, error) {
if 0 == sb.HandshakeTimeout {
return npipe.Dial(`\\.\pipe\` + address)
}
return npipe.DialTimeout(`\\.\pipe\`+address, sb.HandshakeTimeout)
}

28
client/socket_handlers.go
View File

@ -1,28 +0,0 @@
package client
import (
"log"
"net"
csc "git.loafle.net/commons_go/server/client"
)
type SocketHandlers struct {
csc.SocketHandlers
}
func (sh *SocketHandlers) OnConnect(socketContext csc.SocketContext, conn net.Conn) {
log.Printf("OnConnect res: %v \n", conn)
}
func (sh *SocketHandlers) OnDisconnect(soc csc.Socket) {
log.Printf("OnDisconnect \n")
}
func (sh *SocketHandlers) Validate() {
sh.SocketHandlers.Validate()
}
func NewSocketHandler() csc.SocketHandler {
return &SocketHandlers{}
}

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,24 +1,21 @@
package server package server
import ( import (
crpj "git.loafle.net/commons_go/rpc/protocol/json"
crr "git.loafle.net/commons_go/rpc/registry" crr "git.loafle.net/commons_go/rpc/registry"
"git.loafle.net/commons_go/server" "git.loafle.net/commons_go/server"
"git.loafle.net/overflow/overflow_discovery/service" "git.loafle.net/overflow/overflow_discovery/service"
oopcs "git.loafle.net/overflow/overflow_probe_container/server"
) )
func New(addr string) server.Server { func New(addr string) server.Server {
rpcRegistry := crr.NewRPCRegistry() rpcRegistry := crr.NewRPCRegistry()
service.RegisterRPC(rpcRegistry) service.RegisterRPC(rpcRegistry)
rpcSH := newRPCServletHandler(rpcRegistry) rpcSH := oopcs.NewRPCServletHandler(rpcRegistry)
rpcSH.RegisterCodec(crpj.Name, crpj.NewServerCodec())
socketHandler := newSocketHandler(rpcSH) socketHandler := newSocketHandler(rpcSH)
sh := newServerHandler(addr, socketHandler) sh := newServerHandler(addr, socketHandler)
s := server.New(sh) s := oopcs.New(sh)
return s return s
} }

6
server/server_handler.go
View File

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

30
server/server_handlers.go
View File

@ -1,32 +1,24 @@
package server package server
import ( import (
"fmt"
"git.loafle.net/commons_go/logging"
"git.loafle.net/overflow/overflow_discovery/discovery"
"git.loafle.net/commons_go/server" "git.loafle.net/commons_go/server"
"git.loafle.net/overflow/overflow_discovery/discovery"
oopcs "git.loafle.net/overflow/overflow_probe_container/server"
) )
func newServerHandler(addr string, socketHandler SocketHandler) ServerHandler { func newServerHandler(addr string, socketHandler SocketHandler) ServerHandler {
sh := &ServerHandlers{ sh := &ServerHandlers{}
addr: addr, sh.ServerHandler = oopcs.NewServerHandler(addr, "Discovery Containter", socketHandler)
}
sh.Name = "Discovery"
sh.SocketHandler = socketHandler
return sh return sh
} }
type ServerHandlers struct { type ServerHandlers struct {
server.ServerHandlers oopcs.ServerHandler
addr string
} }
func (sh *ServerHandlers) Init(serverCTX server.ServerContext) error { 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 return err
} }
@ -34,7 +26,7 @@ func (sh *ServerHandlers) Init(serverCTX server.ServerContext) error {
} }
func (sh *ServerHandlers) OnStart(serverCTX server.ServerContext) { func (sh *ServerHandlers) OnStart(serverCTX server.ServerContext) {
sh.ServerHandlers.OnStart(serverCTX) sh.ServerHandler.OnStart(serverCTX)
discovery.DiscoveryInit() discovery.DiscoveryInit()
} }
@ -43,14 +35,10 @@ func (sh *ServerHandlers) OnStop(serverCTX server.ServerContext) {
discovery.DiscoveryDestroy() discovery.DiscoveryDestroy()
sh.ServerHandlers.OnStop(serverCTX) sh.ServerHandler.OnStop(serverCTX)
} }
func (sh *ServerHandlers) Validate() { func (sh *ServerHandlers) Validate() {
sh.ServerHandlers.Validate() sh.ServerHandler.Validate()
if "" == sh.addr {
logging.Logger().Panic(fmt.Sprintf("Server: Address of server must be specified"))
}
} }

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"
"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 package server
import ( import (
"git.loafle.net/commons_go/server" oopcs "git.loafle.net/overflow/overflow_probe_container/server"
) )
type SocketHandler interface { type SocketHandler interface {
server.SocketHandler oopcs.SocketHandler
} }

72
server/socket_handlers.go
View File

@ -2,36 +2,27 @@ package server
import ( import (
"net" "net"
"sync"
cRPC "git.loafle.net/commons_go/rpc" 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" "git.loafle.net/commons_go/server"
"git.loafle.net/overflow/overflow_discovery/discovery" "git.loafle.net/overflow/overflow_discovery/discovery"
oopc "git.loafle.net/overflow/overflow_probe_container"
oopcs "git.loafle.net/overflow/overflow_probe_container/server"
) )
func newSocketHandler(rpcSH RPCServletHandler) SocketHandler { func newSocketHandler(rpcSH oopcs.RPCServletHandler) SocketHandler {
rpcRWCSH := crsrs.New() sh := &SocketHandlers{}
sh.SocketHandler = oopcs.NewSocketHandler(rpcSH)
sh := &SocketHandlers{
rpcSH: rpcSH,
rpcRWCSH: rpcRWCSH,
}
return sh return sh
} }
type SocketHandlers struct { type SocketHandlers struct {
server.SocketHandlers oopcs.SocketHandler
rpcRWCSH cRPC.ServletReadWriteCloseHandler
rpcSH RPCServletHandler
} }
func (sh *SocketHandlers) Init(serverCTX server.ServerContext) error { 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 return err
} }
@ -43,61 +34,22 @@ func (sh *SocketHandlers) Handshake(socketCTX server.SocketContext, conn net.Con
} }
func (sh *SocketHandlers) OnConnect(soc server.Socket) { func (sh *SocketHandlers) OnConnect(soc server.Socket) {
sh.SocketHandlers.OnConnect(soc) sh.SocketHandler.OnConnect(soc)
soc.Context().SetAttribute(cRPC.ContentTypeKey, json.Name) discovery.RPCServlet = soc.Context().GetAttribute(oopc.RPCServletKey).(cRPC.Servlet)
}
func (sh *SocketHandlers) Handle(soc server.Socket, stopChan <-chan struct{}, doneChan chan<- error) {
var err error
rpcServlet := retainRPCServlet(sh.rpcSH, sh.rpcRWCSH)
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()
<-rpcDoneChan
}
} }
func (sh *SocketHandlers) OnDisconnect(soc server.Socket) { func (sh *SocketHandlers) OnDisconnect(soc server.Socket) {
discovery.RPCServlet = nil
sh.SocketHandlers.OnDisconnect(soc) sh.SocketHandler.OnDisconnect(soc)
} }
func (sh *SocketHandlers) Destroy() { func (sh *SocketHandlers) Destroy() {
sh.SocketHandlers.Destroy() sh.SocketHandler.Destroy()
} }
func (sh *SocketHandlers) Validate() { 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)
}