commons_go/rpc
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2017-11-29 01:24:16 +09:00
parent 6aa5c1536d
commit c0fdabe039
10 changed files with 72 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
connection/socket/servlet_handlers.go
View File

@ -10,14 +10,14 @@ import (
type ServletHandlers struct { type ServletHandlers struct {
} }
func (sh *ServletHandlers) ReadRequest(servletCTX rpc.RPCServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) { func (sh *ServletHandlers) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
nConn := conn.(net.Conn) nConn := conn.(net.Conn)
requestCodec, err := codec.NewRequest(nConn) requestCodec, err := codec.NewRequest(nConn)
return requestCodec, err return requestCodec, err
} }
func (sh *ServletHandlers) WriteResponse(servletCTX rpc.RPCServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error { func (sh *ServletHandlers) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error {
nConn := conn.(net.Conn) nConn := conn.(net.Conn)
if nil != err { if nil != err {
@ -33,7 +33,7 @@ func (sh *ServletHandlers) WriteResponse(servletCTX rpc.RPCServletContext, conn
return nil return nil
} }
func (sh *ServletHandlers) WriteNotification(servletCTX rpc.RPCServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error { func (sh *ServletHandlers) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error {
nConn := conn.(net.Conn) nConn := conn.(net.Conn)
if wErr := codec.WriteNotification(nConn, method, args); nil != wErr { if wErr := codec.WriteNotification(nConn, method, args); nil != wErr {

6
connection/websocket/fasthttp/servlet_handlers.go
View File

@ -10,7 +10,7 @@ import (
type ServletHandlers struct { type ServletHandlers struct {
} }
func (sh *ServletHandlers) ReadRequest(servletCTX rpc.RPCServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) { func (sh *ServletHandlers) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
soc := conn.(cwf.Socket) soc := conn.(cwf.Socket)
_, r, err := soc.NextReader() _, r, err := soc.NextReader()
@ -19,7 +19,7 @@ func (sh *ServletHandlers) ReadRequest(servletCTX rpc.RPCServletContext, codec p
return requestCodec, err return requestCodec, err
} }
func (sh *ServletHandlers) WriteResponse(servletCTX rpc.RPCServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error { func (sh *ServletHandlers) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error {
soc := conn.(cwf.Socket) soc := conn.(cwf.Socket)
wc, wErr := soc.NextWriter(websocket.TextMessage) wc, wErr := soc.NextWriter(websocket.TextMessage)
@ -40,7 +40,7 @@ func (sh *ServletHandlers) WriteResponse(servletCTX rpc.RPCServletContext, conn
return nil return nil
} }
func (sh *ServletHandlers) WriteNotification(servletCTX rpc.RPCServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error { func (sh *ServletHandlers) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error {
soc := conn.(cwf.Socket) soc := conn.(cwf.Socket)
wc, wErr := soc.NextWriter(websocket.TextMessage) wc, wErr := soc.NextWriter(websocket.TextMessage)

20
rpc_servlet_context.go
View File

@ -1,20 +0,0 @@
package rpc
import (
cuc "git.loafle.net/commons_go/util/context"
)
type RPCServletContext interface {
cuc.Context
}
type rpcServletContext struct {
cuc.Context
}
func newRPCServletContext(parent cuc.Context) RPCServletContext {
sCTX := &rpcServletContext{}
sCTX.Context = cuc.NewContext(parent)
return sCTX
}

25
rpc_servlet_handler.go
View File

@ -1,25 +0,0 @@
package rpc
import (
"git.loafle.net/commons_go/rpc/protocol"
cuc "git.loafle.net/commons_go/util/context"
)
type RPCServletHandler interface {
ServletContext(parent cuc.Context) RPCServletContext
Init(servletCTX RPCServletContext) error
ReadRequest(servletCTX RPCServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error)
Invoke(servletCTX RPCServletContext, requestCodec protocol.RegistryCodec) (result interface{}, err error)
WriteResponse(servletCTX RPCServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error
WriteNotification(servletCTX RPCServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error
Destroy(servletCTX RPCServletContext)
RegisterCodec(contentType string, codec protocol.ServerCodec)
getCodec(contentType string) (protocol.ServerCodec, error)
GetPendingResponses() int
Validate()
}

2
server/servlet_handler.go
View File

@ -3,5 +3,5 @@ package server
import "git.loafle.net/commons_go/rpc" import "git.loafle.net/commons_go/rpc"
type ServletHandler interface { type ServletHandler interface {
rpc.RPCServletHandler rpc.ServletHandler
} }

6
server/servlet_handlers.go
View File

@ -9,12 +9,12 @@ import (
) )
type ServletHandlers struct { type ServletHandlers struct {
rpc.RPCServletHandlers rpc.ServletHandlers
RPCRegistry rpc.Registry RPCRegistry rpc.Registry
} }
func (sh *ServletHandlers) Invoke(servletCTX rpc.RPCServletContext, requestCodec protocol.RegistryCodec) (result interface{}, err error) { func (sh *ServletHandlers) Invoke(servletCTX rpc.ServletContext, requestCodec protocol.RegistryCodec) (result interface{}, err error) {
if !sh.RPCRegistry.HasMethod(requestCodec.Method()) { if !sh.RPCRegistry.HasMethod(requestCodec.Method()) {
return nil, fmt.Errorf("RPC Servlet Handler: Method[%s] is not exist", requestCodec.Method()) return nil, fmt.Errorf("RPC Servlet Handler: Method[%s] is not exist", requestCodec.Method())
} }
@ -32,7 +32,7 @@ func (sh *ServletHandlers) Invoke(servletCTX rpc.RPCServletContext, requestCodec
} }
func (sh *ServletHandlers) Validate() { func (sh *ServletHandlers) Validate() {
sh.RPCServletHandlers.Validate() sh.ServletHandlers.Validate()
if nil == sh.RPCRegistry { if nil == sh.RPCRegistry {
logging.Logger().Panic(fmt.Sprintf("RPC Servlet Handler: RPC Registry must be specified")) logging.Logger().Panic(fmt.Sprintf("RPC Servlet Handler: RPC Registry must be specified"))

10
rpc_servlet.go → servlet.go
View File

@ -11,7 +11,7 @@ import (
cuc "git.loafle.net/commons_go/util/context" cuc "git.loafle.net/commons_go/util/context"
) )
func NewRPCServlet(sh RPCServletHandler) RPCServlet { func NewRPCServlet(sh ServletHandler) RPCServlet {
return &rpcServlet{ return &rpcServlet{
sh: sh, sh: sh,
} }
@ -23,12 +23,12 @@ type RPCServlet interface {
Send(method string, args ...interface{}) (err error) Send(method string, args ...interface{}) (err error)
Context() RPCServletContext Context() ServletContext
} }
type rpcServlet struct { type rpcServlet struct {
ctx RPCServletContext ctx ServletContext
sh RPCServletHandler sh ServletHandler
responseQueueChan chan *responseState responseQueueChan chan *responseState
doneChan chan<- error doneChan chan<- error
@ -103,7 +103,7 @@ func (s *rpcServlet) Send(method string, args ...interface{}) (err error) {
return nil return nil
} }
func (s *rpcServlet) Context() RPCServletContext { func (s *rpcServlet) Context() ServletContext {
return s.ctx return s.ctx
} }

20
servlet_context.go Normal file
View File

@ -0,0 +1,20 @@
package rpc
import (
cuc "git.loafle.net/commons_go/util/context"
)
type ServletContext interface {
cuc.Context
}
type servletContext struct {
cuc.Context
}
func newServletContext(parent cuc.Context) ServletContext {
sCTX := &servletContext{}
sCTX.Context = cuc.NewContext(parent)
return sCTX
}

25
servlet_handler.go Normal file
View File

@ -0,0 +1,25 @@
package rpc
import (
"git.loafle.net/commons_go/rpc/protocol"
cuc "git.loafle.net/commons_go/util/context"
)
type ServletHandler interface {
ServletContext(parent cuc.Context) ServletContext
Init(servletCTX ServletContext) error
ReadRequest(servletCTX ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error)
Invoke(servletCTX ServletContext, requestCodec protocol.RegistryCodec) (result interface{}, err error)
WriteResponse(servletCTX ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error
WriteNotification(servletCTX ServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error
Destroy(servletCTX ServletContext)
RegisterCodec(contentType string, codec protocol.ServerCodec)
getCodec(contentType string) (protocol.ServerCodec, error)
GetPendingResponses() int
Validate()
}

24
rpc_servlet_handlers.go → servlet_handlers.go
View File

@ -8,7 +8,7 @@ import (
cuc "git.loafle.net/commons_go/util/context" cuc "git.loafle.net/commons_go/util/context"
) )
type RPCServletHandlers struct { type ServletHandlers struct {
// The maximum number of pending messages in the queue. // The maximum number of pending messages in the queue.
// //
// The number of pending requsts should exceed the expected number // The number of pending requsts should exceed the expected number
@ -21,31 +21,31 @@ type RPCServletHandlers struct {
codecs map[string]protocol.ServerCodec codecs map[string]protocol.ServerCodec
} }
func (sh *RPCServletHandlers) ServletContext(parent cuc.Context) RPCServletContext { func (sh *ServletHandlers) ServletContext(parent cuc.Context) ServletContext {
return newRPCServletContext(parent) return newRPCServletContext(parent)
} }
func (sh *RPCServletHandlers) Init(servletCTX RPCServletContext) error { func (sh *ServletHandlers) Init(servletCTX ServletContext) error {
return nil return nil
} }
func (sh *RPCServletHandlers) ReadRequest(servletCTX RPCServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) { func (sh *ServletHandlers) ReadRequest(servletCTX ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
return nil, fmt.Errorf("Servlet Handler: ReadRequest is not implemented") return nil, fmt.Errorf("Servlet Handler: ReadRequest is not implemented")
} }
func (sh *RPCServletHandlers) Invoke(servletCTX RPCServletContext, requestCodec protocol.RegistryCodec) (result interface{}, err error) { func (sh *ServletHandlers) Invoke(servletCTX ServletContext, requestCodec protocol.RegistryCodec) (result interface{}, err error) {
return nil, fmt.Errorf("Servlet Handler: Invoke is not implemented") return nil, fmt.Errorf("Servlet Handler: Invoke is not implemented")
} }
func (sh *RPCServletHandlers) WriteResponse(servletCTX RPCServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error { func (sh *ServletHandlers) WriteResponse(servletCTX ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error {
return fmt.Errorf("Servlet Handler: WriteResponse is not implemented") return fmt.Errorf("Servlet Handler: WriteResponse is not implemented")
} }
func (sh *RPCServletHandlers) WriteNotification(servletCTX RPCServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error { func (sh *ServletHandlers) WriteNotification(servletCTX ServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error {
return fmt.Errorf("Servlet Handler: WriteNotification is not implemented") return fmt.Errorf("Servlet Handler: WriteNotification is not implemented")
} }
func (sh *RPCServletHandlers) Destroy(servletCTX RPCServletContext) { func (sh *ServletHandlers) Destroy(servletCTX ServletContext) {
// no op // no op
} }
@ -54,14 +54,14 @@ func (sh *RPCServletHandlers) Destroy(servletCTX RPCServletContext) {
// Codecs are defined to process a given serialization scheme, e.g., JSON or // 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, // XML. A codec is chosen based on the "Content-Type" header from the request,
// excluding the charset definition. // excluding the charset definition.
func (sh *RPCServletHandlers) RegisterCodec(contentType string, codec protocol.ServerCodec) { func (sh *ServletHandlers) RegisterCodec(contentType string, codec protocol.ServerCodec) {
if nil == sh.codecs { if nil == sh.codecs {
sh.codecs = make(map[string]protocol.ServerCodec) sh.codecs = make(map[string]protocol.ServerCodec)
} }
sh.codecs[strings.ToLower(contentType)] = codec sh.codecs[strings.ToLower(contentType)] = codec
} }
func (sh *RPCServletHandlers) getCodec(contentType string) (protocol.ServerCodec, error) { func (sh *ServletHandlers) getCodec(contentType string) (protocol.ServerCodec, error) {
var codec protocol.ServerCodec var codec protocol.ServerCodec
if contentType == "" && len(sh.codecs) == 1 { if contentType == "" && len(sh.codecs) == 1 {
// If Content-Type is not set and only one codec has been registered, // If Content-Type is not set and only one codec has been registered,
@ -76,11 +76,11 @@ func (sh *RPCServletHandlers) getCodec(contentType string) (protocol.ServerCodec
return codec, nil return codec, nil
} }
func (sh *RPCServletHandlers) GetPendingResponses() int { func (sh *ServletHandlers) GetPendingResponses() int {
return sh.PendingResponses return sh.PendingResponses
} }
func (sh *RPCServletHandlers) Validate() { func (sh *ServletHandlers) Validate() {
if 0 >= sh.PendingResponses { if 0 >= sh.PendingResponses {
sh.PendingResponses = DefaultPendingResponses sh.PendingResponses = DefaultPendingResponses
} }