ing

server/rpc_servlet_handler.go

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

server/rpc_servlet_handlers.go

@@ -0,0 +1,22 @@
+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()
+
+}

server/server.go

@@ -0,0 +1,24 @@
+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"
+
+	"git.loafle.net/overflow/overflow_probe_container_network/service"
+)
+
+func New(addr string) server.Server {
+	rpcRegistry := crr.NewRPCRegistry()
+	service.RegisterRPC(rpcRegistry)
+
+	rpcSH := newRPCServletHandler(rpcRegistry)
+	rpcSH.RegisterCodec(crpj.Name, crpj.NewServerCodec())
+
+	socketHandler := newSocketHandler(rpcSH)
+
+	sh := newServerHandler(addr, socketHandler)
+	s := server.New(sh)
+
+	return s
+}

server/server_handler.go

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

server/server_handlers.go

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

server/server_handlers_unix.go

@@ -0,0 +1,17 @@
+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
+}

server/server_handlers_windows.go

@@ -0,0 +1,16 @@
+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
+}

server/socket_handler.go

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

server/socket_handlers.go

@@ -0,0 +1,101 @@
+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"
+)
+
+func newSocketHandler(rpcSH RPCServletHandler) SocketHandler {
+	rpcRWCSH := crsrs.New()
+
+	sh := &SocketHandlers{
+		rpcSH:    rpcSH,
+		rpcRWCSH: rpcRWCSH,
+	}
+
+	return sh
+}
+
+type SocketHandlers struct {
+	server.SocketHandlers
+
+	rpcRWCSH cRPC.ServletReadWriteCloseHandler
+	rpcSH    RPCServletHandler
+}
+
+func (sh *SocketHandlers) Init(serverCTX server.ServerContext) error {
+	if err := sh.SocketHandlers.Init(serverCTX); nil != err {
+		return err
+	}
+
+	return nil
+}
+
+func (sh *SocketHandlers) Handshake(socketCTX server.SocketContext, conn net.Conn) (id string) {
+	return "discovery"
+}
+
+func (sh *SocketHandlers) OnConnect(soc server.Socket) {
+	sh.SocketHandlers.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
+	}
+}
+
+func (sh *SocketHandlers) OnDisconnect(soc server.Socket) {
+
+	sh.SocketHandlers.OnDisconnect(soc)
+}
+
+func (sh *SocketHandlers) Destroy() {
+
+	sh.SocketHandlers.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)
+}