This commit is contained in:
crusader 2017-11-01 16:25:29 +09:00
parent 3e14774956
commit a8044481d9
5 changed files with 148 additions and 137 deletions

View File

@ -0,0 +1,24 @@
package server
import (
"io"
"git.loafle.net/commons_go/rpc/protocol"
)
type RPCServerHandler interface {
GetContentType(r io.Reader) string
RegisterCodec(codec protocol.ServerCodec, contentType string)
OnPreRead(r io.Reader)
OnPostRead(r io.Reader)
OnPreWriteResult(w io.Writer, result interface{})
OnPostWriteResult(w io.Writer, result interface{})
OnPreWriteError(w io.Writer, err error)
OnPostWriteError(w io.Writer, err error)
getCodec(contentType string) (protocol.ServerCodec, error)
invoke(codec protocol.RegistryCodec) (result interface{}, err error)
}

View File

@ -0,0 +1,80 @@
package server
import (
"fmt"
"io"
"strings"
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol"
)
type RPCServerHandlers struct {
RPCRegistry rpc.Registry
codecs map[string]protocol.ServerCodec
}
// RegisterCodec adds a new codec to the server.
//
// Codecs are defined to process a given serialization scheme, e.g., JSON or
// XML. A codec is chosen based on the "Content-Type" header from the request,
// excluding the charset definition.
func (rpcSH *RPCServerHandlers) RegisterCodec(codec protocol.ServerCodec, contentType string) {
if nil == rpcSH.codecs {
rpcSH.codecs = make(map[string]protocol.ServerCodec)
}
rpcSH.codecs[strings.ToLower(contentType)] = codec
}
func (rpcSH *RPCServerHandlers) GetContentType(r io.Reader) string {
return ""
}
func (rpcSH *RPCServerHandlers) OnPreRead(r io.Reader) {
// no op
}
func (rpcSH *RPCServerHandlers) OnPostRead(r io.Reader) {
// no op
}
func (rpcSH *RPCServerHandlers) OnPreWriteResult(w io.Writer, result interface{}) {
// no op
}
func (rpcSH *RPCServerHandlers) OnPostWriteResult(w io.Writer, result interface{}) {
// no op
}
func (rpcSH *RPCServerHandlers) OnPreWriteError(w io.Writer, err error) {
// no op
}
func (rpcSH *RPCServerHandlers) OnPostWriteError(w io.Writer, err error) {
// no op
}
func (rpcSH *RPCServerHandlers) Validate() {
if nil == rpcSH.RPCRegistry {
panic("RPCRegistry must be specified.")
}
}
func (rpcSH *RPCServerHandlers) getCodec(contentType string) (protocol.ServerCodec, error) {
var codec protocol.ServerCodec
if contentType == "" && len(rpcSH.codecs) == 1 {
// If Content-Type is not set and only one codec has been registered,
// then default to that codec.
for _, c := range rpcSH.codecs {
codec = c
}
} else if codec = rpcSH.codecs[strings.ToLower(contentType)]; codec == nil {
return nil, fmt.Errorf("Unrecognized Content-Type: %s", contentType)
}
return codec, nil
}
func (rpcSH *RPCServerHandlers) invoke(codec protocol.RegistryCodec) (result interface{}, err error) {
return rpcSH.RPCRegistry.Invoke(codec)
}

View File

@ -1,67 +1,11 @@
package server package server
import ( import (
"log"
"net"
"git.loafle.net/commons_go/server" "git.loafle.net/commons_go/server"
) )
func New(sh ServerHandler) server.Server { func New(sh ServerHandler) server.Server {
lsh := &serverHandlers{} s := server.New(sh)
lsh.ServerHandler = sh
lsh.lsh = sh
s := server.New(lsh)
return s return s
} }
type serverHandlers struct {
server.ServerHandler
lsh ServerHandler
}
func (sh *serverHandlers) Handle(conn net.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}) {
contentType := sh.lsh.GetContentType(conn)
codec, err := sh.lsh.getCodec(contentType)
if nil != err {
log.Printf("RPC Handle: %v", err)
doneChan <- struct{}{}
return
}
for {
sh.lsh.OnPreRead(conn)
// Create a new codec request.
codecReq, errNew := codec.NewRequest(conn)
if nil != errNew {
if sh.IsClientDisconnect(errNew) {
doneChan <- struct{}{}
return
}
log.Printf("RPC Handle: %v", errNew)
doneChan <- struct{}{}
return
}
sh.lsh.OnPostRead(conn)
result, err := sh.lsh.invoke(codecReq)
if nil != err {
sh.lsh.OnPreWriteError(conn, err)
codecReq.WriteError(conn, 400, err)
sh.lsh.OnPostWriteError(conn, err)
} else {
sh.lsh.OnPreWriteResult(conn, result)
codecReq.WriteResponse(conn, result)
sh.lsh.OnPostWriteResult(conn, result)
}
select {
case <-stopChan:
return
default:
}
}
}

View File

@ -1,26 +1,7 @@
package server package server
import ( import "git.loafle.net/commons_go/server"
"io"
"git.loafle.net/commons_go/rpc/protocol"
"git.loafle.net/commons_go/server"
)
type ServerHandler interface { type ServerHandler interface {
server.ServerHandler server.ServerHandler
GetContentType(r io.Reader) string
RegisterCodec(codec protocol.ServerCodec, contentType string)
OnPreRead(r io.Reader)
OnPostRead(r io.Reader)
OnPreWriteResult(w io.Writer, result interface{})
OnPostWriteResult(w io.Writer, result interface{})
OnPreWriteError(w io.Writer, err error)
OnPostWriteError(w io.Writer, err error)
getCodec(contentType string) (protocol.ServerCodec, error)
invoke(codec protocol.RegistryCodec) (result interface{}, err error)
} }

View File

@ -1,85 +1,67 @@
package server package server
import ( import (
"fmt" "log"
"io" "net"
"strings"
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol"
"git.loafle.net/commons_go/server" "git.loafle.net/commons_go/server"
) )
type ServerHandlers struct { type ServerHandlers struct {
server.ServerHandlers server.ServerHandlers
RPCRegistry rpc.Registry RPCServerHandler RPCServerHandler
codecs map[string]protocol.ServerCodec
} }
// RegisterCodec adds a new codec to the server. func (sh *ServerHandlers) Handle(conn net.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}) {
// contentType := sh.RPCServerHandler.GetContentType(conn)
// Codecs are defined to process a given serialization scheme, e.g., JSON or codec, err := sh.RPCServerHandler.getCodec(contentType)
// XML. A codec is chosen based on the "Content-Type" header from the request, if nil != err {
// excluding the charset definition. log.Printf("RPC Handle: %v", err)
func (sh *ServerHandlers) RegisterCodec(codec protocol.ServerCodec, contentType string) { doneChan <- struct{}{}
if nil == sh.codecs { return
sh.codecs = make(map[string]protocol.ServerCodec)
}
sh.codecs[strings.ToLower(contentType)] = codec
} }
func (sh *ServerHandlers) GetContentType(r io.Reader) string { for {
return "" sh.RPCServerHandler.OnPreRead(conn)
// Create a new codec request.
codecReq, errNew := codec.NewRequest(conn)
if nil != errNew {
if sh.IsClientDisconnect(errNew) {
doneChan <- struct{}{}
return
}
log.Printf("RPC Handle: %v", errNew)
doneChan <- struct{}{}
return
}
sh.RPCServerHandler.OnPostRead(conn)
result, err := sh.RPCServerHandler.invoke(codecReq)
if nil != err {
sh.RPCServerHandler.OnPreWriteError(conn, err)
codecReq.WriteError(conn, 400, err)
sh.RPCServerHandler.OnPostWriteError(conn, err)
} else {
sh.RPCServerHandler.OnPreWriteResult(conn, result)
codecReq.WriteResponse(conn, result)
sh.RPCServerHandler.OnPostWriteResult(conn, result)
} }
func (sh *ServerHandlers) OnPreRead(r io.Reader) { select {
// no op case <-stopChan:
return
default:
}
} }
func (sh *ServerHandlers) OnPostRead(r io.Reader) {
// no op
}
func (sh *ServerHandlers) OnPreWriteResult(w io.Writer, result interface{}) {
// no op
}
func (sh *ServerHandlers) OnPostWriteResult(w io.Writer, result interface{}) {
// no op
}
func (sh *ServerHandlers) OnPreWriteError(w io.Writer, err error) {
// no op
}
func (sh *ServerHandlers) OnPostWriteError(w io.Writer, err error) {
// no op
} }
func (sh *ServerHandlers) Validate() { func (sh *ServerHandlers) Validate() {
sh.ServerHandlers.Validate() sh.ServerHandlers.Validate()
if nil == sh.RPCRegistry { if nil == sh.RPCServerHandler {
panic("RPCRegistry must be specified.") panic("RPCServerHandler must be specified.")
} }
} }
func (sh *ServerHandlers) getCodec(contentType string) (protocol.ServerCodec, error) {
var codec protocol.ServerCodec
if contentType == "" && len(sh.codecs) == 1 {
// If Content-Type is not set and only one codec has been registered,
// then default to that codec.
for _, c := range sh.codecs {
codec = c
}
} else if codec = sh.codecs[strings.ToLower(contentType)]; codec == nil {
return nil, fmt.Errorf("Unrecognized Content-Type: %s", contentType)
}
return codec, nil
}
func (sh *ServerHandlers) invoke(codec protocol.RegistryCodec) (result interface{}, err error) {
return sh.RPCRegistry.Invoke(codec)
}