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 18:55:24 +09:00
parent 8e6981767b
commit afd49dd0dc
17 changed files with 379 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
client/client.go
View File

@ -4,7 +4,6 @@ import (
"fmt"
"io"
"log"
"net"
"reflect"
"runtime"
"sync"
@ -15,9 +14,10 @@ import (
"git.loafle.net/commons_go/rpc/protocol"
)
func New(ch ClientHandler) Client {
func New(ch ClientHandler, rwcHandler ClientReadWriteCloseHandler) Client {
c := &client{
ch: ch,
ch: ch,
rwcHandler: rwcHandler,
}
return c
}
@ -32,9 +32,11 @@ type Client interface {
}
type client struct {
ch ClientHandler
ctx ClientContext
ch ClientHandler
rwcHandler ClientReadWriteCloseHandler
conn net.Conn
conn interface{}
pendingRequestsCount uint32
pendingRequests map[uint64]*RequestState
@ -48,13 +50,27 @@ type client struct {
func (c *client) Connect() error {
var err error
if nil == c.ch {
panic("RPC Client: Client handler must be specified.")
}
c.ch.Validate()
if nil == c.rwcHandler {
panic("RPC Client: Client RWC handler must be specified.")
}
c.rwcHandler.Validate()
if c.stopChan != nil {
panic("RPC Client: the given client is already started. Call Client.Stop() before calling Client.Start() again!")
}
c.ctx = c.ch.ClientContext(nil)
if c.conn, err = c.ch.Connect(); nil != err {
if err := c.ch.Init(c.ctx); nil != err {
logging.Logger().Panic(fmt.Sprintf("RPC Client: Initialization of client has been failed %v", err))
}
if c.conn, err = c.rwcHandler.Connect(); nil != err {
return err
}
c.stopChan = make(chan struct{})
@ -68,23 +84,28 @@ func (c *client) Connect() error {
func (c *client) Close() {
if c.stopChan == nil {
panic("Client: the client must be started before stopping it")
panic("RPC Client: the client must be started before stopping it")
}
c.ch.Destroy(c.ctx)
close(c.stopChan)
c.stopWg.Wait()
c.stopChan = nil
logging.Logger().Info(fmt.Sprintf("RPC Client: stopped"))
}
func (c *client) Send(method string, args ...interface{}) (err error) {
var cs *RequestState
if cs, err = c.send(true, false, nil, method, args...); nil != err {
var rs *RequestState
if rs, err = c.send(true, false, nil, method, args...); nil != err {
return
}
select {
case <-cs.DoneChan:
err = cs.Error
releaseCallState(cs)
case <-rs.DoneChan:
err = rs.Error
releaseCallState(rs)
}
return
@ -95,19 +116,19 @@ func (c *client) Call(result interface{}, method string, args ...interface{}) er
}
func (c *client) CallTimeout(timeout time.Duration, result interface{}, method string, args ...interface{}) (err error) {
var cs *RequestState
if cs, err = c.send(true, true, result, method, args...); nil != err {
var rs *RequestState
if rs, err = c.send(true, true, result, method, args...); nil != err {
return
}
t := retainTimer(timeout)
select {
case <-cs.DoneChan:
result, err = cs.Result, cs.Error
releaseCallState(cs)
case <-rs.DoneChan:
result, err = rs.Result, rs.Error
releaseCallState(rs)
case <-t.C:
cs.Cancel()
rs.Cancel()
err = getClientTimeoutError(c, timeout)
}
@ -116,30 +137,30 @@ func (c *client) CallTimeout(timeout time.Duration, result interface{}, method s
return
}
func (c *client) send(usePool bool, hasResponse bool, result interface{}, method string, args ...interface{}) (cs *RequestState, err error) {
func (c *client) send(usePool bool, hasResponse bool, result interface{}, method string, args ...interface{}) (rs *RequestState, err error) {
if !hasResponse {
usePool = true
}
if usePool {
cs = retainRequestState()
rs = retainRequestState()
} else {
cs = &RequestState{}
rs = &RequestState{}
}
cs.hasResponse = hasResponse
cs.Method = method
cs.Args = args
cs.DoneChan = make(chan *RequestState, 1)
rs.hasResponse = hasResponse
rs.Method = method
rs.Args = args
rs.DoneChan = make(chan *RequestState, 1)
if hasResponse {
cs.ID = c.ch.GetRequestID()
cs.Result = result
rs.ID = c.ch.GetRequestID()
rs.Result = result
}
select {
case c.requestQueueChan <- cs:
return cs, nil
case c.requestQueueChan <- rs:
return rs, nil
default:
// Try substituting the oldest async request by the new one
// on requests' queue overflow.
@ -149,7 +170,7 @@ func (c *client) send(usePool bool, hasResponse bool, result interface{}, method
// Immediately notify the caller not interested
// in the response on requests' queue overflow, since
// there are no other ways to notify it later.
releaseCallState(cs)
releaseCallState(rs)
return nil, getClientOverflowError(c)
}
@ -166,12 +187,12 @@ func (c *client) send(usePool bool, hasResponse bool, result interface{}, method
}
select {
case c.requestQueueChan <- cs:
return cs, nil
case c.requestQueueChan <- rs:
return rs, nil
default:
// Release m even if usePool = true, since m wasn't exposed
// to the caller yet.
releaseCallState(cs)
releaseCallState(rs)
return nil, getClientOverflowError(c)
}
}
@ -201,7 +222,7 @@ func (c *client) handleRPC() {
<-writerDone
}
c.conn.Close()
c.rwcHandler.Disconnect(c.conn)
if err != nil {
//c.LogError("%s", err)
@ -212,11 +233,11 @@ func (c *client) handleRPC() {
}
}
for _, cs := range c.pendingRequests {
for _, rs := range c.pendingRequests {
atomic.AddUint32(&c.pendingRequestsCount, ^uint32(0))
cs.Error = err
if cs.DoneChan != nil {
cs.Done()
rs.Error = err
if rs.DoneChan != nil {
rs.Done()
}
}
@ -229,10 +250,10 @@ func (c *client) rpcWriter(stopChan <-chan struct{}, writerDone chan<- error) {
}()
for {
var cs *RequestState
var rs *RequestState
select {
case cs = <-c.requestQueueChan:
case rs = <-c.requestQueueChan:
default:
// Give the last chance for ready goroutines filling c.requestsChan :)
runtime.Gosched()
@ -240,25 +261,25 @@ func (c *client) rpcWriter(stopChan <-chan struct{}, writerDone chan<- error) {
select {
case <-stopChan:
return
case cs = <-c.requestQueueChan:
case rs = <-c.requestQueueChan:
}
}
if cs.IsCanceled() {
if nil != cs.DoneChan {
// cs.Error = ErrCanceled
if rs.IsCanceled() {
if nil != rs.DoneChan {
// rs.Error = ErrCanceled
// close(m.done)
cs.Done()
rs.Done()
} else {
releaseCallState(cs)
releaseCallState(rs)
}
continue
}
if cs.hasResponse {
if rs.hasResponse {
c.pendingRequestsLock.Lock()
n := len(c.pendingRequests)
c.pendingRequests[cs.ID] = cs
c.pendingRequests[rs.ID] = rs
c.pendingRequestsLock.Unlock()
atomic.AddUint32(&c.pendingRequestsCount, 1)
@ -270,13 +291,14 @@ func (c *client) rpcWriter(stopChan <-chan struct{}, writerDone chan<- error) {
}
var requestID interface{}
if 0 < cs.ID {
requestID = cs.ID
if 0 < rs.ID {
requestID = rs.ID
}
err = c.ch.GetCodec().WriteRequest(c.conn, cs.Method, cs.Args, requestID)
if !cs.hasResponse {
cs.Error = err
cs.Done()
err = c.rwcHandler.WriteRequest(c.ctx, c.ch.GetCodec(), c.conn, rs.Method, rs.Args, requestID)
if !rs.hasResponse {
rs.Error = err
rs.Done()
}
if nil != err {
if err == io.ErrUnexpectedEOF || err == io.EOF {
@ -303,7 +325,7 @@ func (c *client) rpcReader(readerDone chan<- error) {
}()
for {
resCodec, err := c.ch.GetCodec().NewResponse(c.conn)
resCodec, err := c.rwcHandler.ReadResponse(c.ctx, c.ch.GetCodec(), c.conn)
if nil != err {
if err == io.ErrUnexpectedEOF || err == io.EOF {
logging.Logger().Info("Client: disconnected from server")
@ -332,7 +354,7 @@ func (c *client) handleResponse(resCodec protocol.ClientResponseCodec) error {
c.pendingRequestsLock.Lock()
id := reflect.ValueOf(resCodec.ID()).Convert(uint64Type).Uint()
cs, ok := c.pendingRequests[id]
rs, ok := c.pendingRequests[id]
if ok {
delete(c.pendingRequests, id)
}
@ -344,18 +366,18 @@ func (c *client) handleResponse(resCodec protocol.ClientResponseCodec) error {
atomic.AddUint32(&c.pendingRequestsCount, ^uint32(0))
if err := resCodec.Result(cs.Result); nil != err {
if err := resCodec.Result(rs.Result); nil != err {
log.Printf("responseHandle:%v", err)
}
if err := resCodec.Error(); nil != err {
log.Printf("responseHandle:%v", err)
// cs.Error = &ClientError{
// rs.Error = &ClientError{
// Server: true,
// err: fmt.Errorf("gorpc.Client: [%s]. Server error: [%s]", c.Addr, wr.Error),
// }
}
cs.Done()
rs.Done()
return nil
}

20
client/client_context.go Normal file
View File

@ -0,0 +1,20 @@
package client
import (
cuc "git.loafle.net/commons_go/util/context"
)
type ClientContext interface {
cuc.Context
}
type clientContext struct {
cuc.Context
}
func newClientContext(parent cuc.Context) ClientContext {
cCTX := &clientContext{}
cCTX.Context = cuc.NewContext(parent)
return cCTX
}

10
client/client_handler.go
View File

@ -1,21 +1,23 @@
package client
import (
"net"
"time"
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol"
cuc "git.loafle.net/commons_go/util/context"
)
type ClientHandler interface {
Connect() (net.Conn, error)
GetCodec() protocol.ClientCodec
ClientContext(parent cuc.Context) ClientContext
Init(clientCTX ClientContext) error
Destroy(clientCTX ClientContext)
GetCodec() protocol.ClientCodec
GetRPCRegistry() rpc.Registry
GetRequestTimeout() time.Duration
GetPendingRequests() int
GetRequestID() uint64
Validate()
}

29
client/client_handlers.go
View File

@ -1,17 +1,19 @@
package client
import (
"errors"
"net"
"sync"
"time"
"git.loafle.net/commons_go/logging"
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol"
)
type ClientHandlers struct {
Codec protocol.ClientCodec
Codec protocol.ClientCodec
RPCRegistry rpc.Registry
// Maximum request time.
// Default value is DefaultRequestTimeout.
RequestTimeout time.Duration
@ -24,14 +26,20 @@ type ClientHandlers struct {
// Default is DefaultPendingMessages.
PendingRequests int
RPCRegistry rpc.Registry
requestID uint64
requestIDMtx sync.Mutex
}
func (ch *ClientHandlers) Connect() (net.Conn, error) {
return nil, errors.New("RPC Client: ClientHandlers method[Connect] is not implement")
func (ch *ClientHandlers) ClientContext(parent cuc.Context) ClientContext {
return newClientContext(parent)
}
func (ch *ClientHandlers) Init(clientCTX ClientContext) error {
return nil
}
func (ch *ClientHandlers) Destroy(clientCTX ClientContext) {
// no op
}
func (ch *ClientHandlers) GetCodec() protocol.ClientCodec {
@ -61,6 +69,13 @@ func (ch *ClientHandlers) GetRequestID() uint64 {
}
func (ch *ClientHandlers) Validate() {
if nil == ch.Codec {
logging.Logger().Panic("RPC Client Handler: Codec must be specified")
}
if nil == ch.RPCRegistry {
logging.Logger().Panic("RPC Client Handler: RPCRegistry must be specified")
}
if ch.RequestTimeout <= 0 {
ch.RequestTimeout = DefaultRequestTimeout
}

12
client/client_rwc_handler.go Normal file
View File

@ -0,0 +1,12 @@
package client
import "git.loafle.net/commons_go/rpc/protocol"
type ClientReadWriteCloseHandler interface {
Connect() (interface{}, error)
ReadResponse(clientCTX ClientContext, codec protocol.ClientCodec, conn interface{}) (protocol.ClientResponseCodec, error)
WriteRequest(clientCTX ClientContext, codec protocol.ClientCodec, conn interface{}, method string, params interface{}, id interface{}) error
Disconnect(conn interface{})
Validate()
}

30
client/client_rwc_handlers.go Normal file
View File

@ -0,0 +1,30 @@
package client
import (
"fmt"
"git.loafle.net/commons_go/rpc/protocol"
)
type ClientReadWriteCloseHandlers struct {
}
func (crwch *ClientReadWriteCloseHandlers) Connect() (interface{}, error) {
return nil, fmt.Errorf("RPC Client RWC Handler: ClientHandlers method[Connect] is not implement")
}
func (crwch *ClientReadWriteCloseHandlers) ReadResponse(clientCTX ClientContext, codec protocol.ClientCodec, conn interface{}) (protocol.ClientResponseCodec, error) {
return nil, fmt.Errorf("RPC Client RWC Handler: ClientHandlers method[ReadResponse] is not implement")
}
func (crwch *ClientReadWriteCloseHandlers) WriteRequest(clientCTX ClientContext, codec protocol.ClientCodec, conn interface{}, method string, params interface{}, id interface{}) error {
return fmt.Errorf("RPC Client RWC Handler: ClientHandlers method[WriteRequest] is not implement")
}
func (crwch *ClientReadWriteCloseHandlers) Disconnect(conn interface{}) {
// no op
}
func (crwch *ClientReadWriteCloseHandlers) Validate() {
}

2
client/request_state.go
View File

@ -11,7 +11,7 @@ var zeroTime time.Time
type RequestState struct {
ID uint64
Method string
Args interface{}
Args []interface{}
Result interface{}
Error error
DoneChan chan *RequestState

48
client/rwc/socket/client_handlers.go Normal file
View File

@ -0,0 +1,48 @@
package socket
import (
"fmt"
"git.loafle.net/commons_go/logging"
"git.loafle.net/commons_go/rpc/client"
"git.loafle.net/commons_go/rpc/protocol"
"git.loafle.net/commons_go/server"
)
type ClientReadWriteCloseHandlers struct {
client.ClientReadWriteCloseHandlers
Address string
}
func (crwch *ClientReadWriteCloseHandlers) Connect() (interface{}, error) {
return nil, fmt.Errorf("RPC Client RWC Handler: ClientHandlers method[Connect] is not implement")
}
func (crwch *ClientReadWriteCloseHandlers) ReadResponse(clientCTX ClientContext, codec protocol.ClientCodec, conn interface{}) (protocol.ClientResponseCodec, error) {
soc := conn.(server.Socket)
resCodec, err := codec.NewResponse(soc)
return resCodec, err
}
func (crwch *ClientReadWriteCloseHandlers) WriteRequest(clientCTX ClientContext, codec protocol.ClientCodec, conn interface{}, method string, params interface{}, id interface{}) error {
soc := conn.(server.Socket)
if wErr := codec.WriteRequest(soc, method, params); nil != wErr {
return wErr
}
return nil
}
func (crwch *ClientReadWriteCloseHandlers) Disconnect(conn interface{}) {
soc := conn.(server.Socket)
soc.Close()
}
func (crwch *ClientReadWriteCloseHandlers) Validate() {
if "" == crwch.Address {
logging.Logger().Panic("RPC Client RWC Handler: Address must be specified")
}
}

52
client/rwc/websocket/fasthttp/client_handlers.go Normal file
View File

@ -0,0 +1,52 @@
package fasthttp
import (
"fmt"
"github.com/gorilla/websocket"
"git.loafle.net/commons_go/rpc/client"
"git.loafle.net/commons_go/rpc/protocol"
cwf "git.loafle.net/commons_go/websocket_fasthttp"
)
type ClientReadWriteCloseHandlers struct {
client.ClientReadWriteCloseHandlers
}
func (crwch *ClientReadWriteCloseHandlers) Connect() (interface{}, error) {
return nil, fmt.Errorf("RPC Client RWC Handler: ClientHandlers method[Connect] is not implement")
}
func (crwch *ClientReadWriteCloseHandlers) ReadResponse(clientCTX ClientContext, codec protocol.ClientCodec, conn interface{}) (protocol.ClientResponseCodec, error) {
soc := conn.(cwf.Socket)
_, r, err := soc.NextReader()
resCodec, err := codec.NewResponse(r)
return resCodec, err
}
func (crwch *ClientReadWriteCloseHandlers) WriteRequest(clientCTX ClientContext, codec protocol.ClientCodec, conn interface{}, method string, params interface{}, id interface{}) error {
soc := conn.(cwf.Socket)
wc, wErr := soc.NextWriter(websocket.TextMessage)
if nil != wErr {
return wErr
}
if wErr := codec.WriteRequest(wc, method, params); nil != wErr {
return wErr
}
return nil
}
func (crwch *ClientReadWriteCloseHandlers) Disconnect(conn interface{}) {
soc := conn.(cwf.Socket)
soc.Close()
}
func (crwch *ClientReadWriteCloseHandlers) Validate() {
}

44
connection/socket/servlet_handlers.go
View File

@ -1,44 +0,0 @@
package socket
import (
"net"
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol"
)
type ServletHandlers struct {
}
func (sh *ServletHandlers) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
nConn := conn.(net.Conn)
requestCodec, err := codec.NewRequest(nConn)
return requestCodec, err
}
func (sh *ServletHandlers) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error {
nConn := conn.(net.Conn)
if nil != err {
if wErr := requestCodec.WriteError(nConn, 500, err); nil != wErr {
return wErr
}
} else {
if wErr := requestCodec.WriteResponse(nConn, result); nil != wErr {
return wErr
}
}
return nil
}
func (sh *ServletHandlers) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error {
nConn := conn.(net.Conn)
if wErr := codec.WriteNotification(nConn, method, args); nil != wErr {
return wErr
}
return nil
}

44
server/rwc/socket/servlet_handlers.go Normal file
View File

@ -0,0 +1,44 @@
package socket
import (
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol"
"git.loafle.net/commons_go/server"
)
type ServletReadWriteCloseHandler struct {
rpc.ServletReadWriteCloseHandler
}
func (srwch *ServletReadWriteCloseHandler) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
soc := conn.(server.Socket)
reqCodec, err := codec.NewRequest(soc)
return reqCodec, err
}
func (srwch *ServletReadWriteCloseHandler) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, reqCodec protocol.ServerRequestCodec, result interface{}, err error) error {
soc := conn.(server.Socket)
if nil != err {
if wErr := reqCodec.WriteError(soc, 500, err); nil != wErr {
return wErr
}
} else {
if wErr := reqCodec.WriteResponse(soc, result); nil != wErr {
return wErr
}
}
return nil
}
func (srwch *ServletReadWriteCloseHandler) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, params interface{}) error {
soc := conn.(server.Socket)
if wErr := codec.WriteNotification(soc, method, params); nil != wErr {
return wErr
}
return nil
}

9
connection/websocket/fasthttp/servlet_handlers.go → server/rwc/websocket/fasthttp/servlet_handlers.go
View File

@ -7,10 +7,11 @@ import (
"git.loafle.net/commons_go/websocket_fasthttp/websocket"
)
type ServletHandlers struct {
type ServletReadWriteCloseHandler struct {
rpc.ServletReadWriteCloseHandler
}
func (sh *ServletHandlers) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
func (srwch *ServletReadWriteCloseHandler) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
soc := conn.(cwf.Socket)
_, r, err := soc.NextReader()
@ -19,7 +20,7 @@ func (sh *ServletHandlers) ReadRequest(servletCTX rpc.ServletContext, codec prot
return requestCodec, err
}
func (sh *ServletHandlers) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error {
func (srwch *ServletReadWriteCloseHandler) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error {
soc := conn.(cwf.Socket)
wc, wErr := soc.NextWriter(websocket.TextMessage)
@ -40,7 +41,7 @@ func (sh *ServletHandlers) WriteResponse(servletCTX rpc.ServletContext, conn int
return nil
}
func (sh *ServletHandlers) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error {
func (srwch *ServletReadWriteCloseHandler) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, args []interface{}) error {
soc := conn.(cwf.Socket)
wc, wErr := soc.NextWriter(websocket.TextMessage)

32
servlet.go
View File

@ -11,9 +11,10 @@ import (
cuc "git.loafle.net/commons_go/util/context"
)
func NewServlet(sh ServletHandler) Servlet {
func NewServlet(sh ServletHandler, rwcSH ServletReadWriteCloseHandler) Servlet {
return &rpcServlet{
sh: sh,
sh: sh,
rwcSH: rwcSH,
}
}
@ -27,8 +28,10 @@ type Servlet interface {
}
type rpcServlet struct {
ctx ServletContext
sh ServletHandler
ctx ServletContext
sh ServletHandler
rwcSH ServletReadWriteCloseHandler
responseQueueChan chan *responseState
doneChan chan<- error
@ -45,12 +48,17 @@ func (s *rpcServlet) Start(parentCTX cuc.Context, conn interface{}, doneChan cha
}
s.sh.Validate()
if nil == s.rwcSH {
panic("Servlet: servlet RWC handler must be specified.")
}
s.rwcSH.Validate()
if s.stopChan != nil {
return fmt.Errorf("Servlet: servlet is already running. Stop it before starting it again")
}
servletCTX := s.sh.ServletContext(parentCTX)
s.ctx = s.sh.ServletContext(parentCTX)
sc, err := s.sh.getCodec(servletCTX.GetAttribute(ContentTypeKey).(string))
sc, err := s.sh.getCodec(s.ctx.GetAttribute(ContentTypeKey).(string))
if nil != err {
return err
}
@ -76,10 +84,12 @@ func (s *rpcServlet) Stop() {
if s.stopChan == nil {
panic("Server: server must be started before stopping it")
}
s.sh.Destroy(s.ctx)
close(s.stopChan)
s.stopWg.Wait()
s.stopChan = nil
s.sh.Destroy(s.ctx)
s.responseQueueChan = nil
@ -153,7 +163,7 @@ func handleReader(s *rpcServlet, stopChan chan struct{}, doneChan chan error) {
}()
for {
requestCodec, err := s.sh.ReadRequest(s.ctx, s.serverCodec, s.conn)
requestCodec, err := s.rwcSH.ReadRequest(s.ctx, s.serverCodec, s.conn)
if nil != err {
if err == io.ErrUnexpectedEOF || err == io.EOF {
err = fmt.Errorf("RPC Server: disconnected from client")
@ -204,11 +214,11 @@ func handleWriter(s *rpcServlet, stopChan chan struct{}, doneChan chan error) {
}
if nil != rs.requestCodec {
if err := s.sh.WriteResponse(s.ctx, s.conn, rs.requestCodec, rs.result, rs.err); nil != err {
if err := s.rwcSH.WriteResponse(s.ctx, s.conn, rs.requestCodec, rs.result, rs.err); nil != err {
logging.Logger().Error(fmt.Sprintf("RPC Server: response error %v", err))
}
} else {
if err := s.sh.WriteNotification(s.ctx, s.conn, s.serverCodec, rs.noti.method, rs.noti.args); nil != err {
if err := s.rwcSH.WriteNotification(s.ctx, s.conn, s.serverCodec, rs.noti.method, rs.noti.args); nil != err {
logging.Logger().Error(fmt.Sprintf("RPC Server: notification error %v", err))
}
}
@ -242,5 +252,5 @@ type responseState struct {
type notification struct {
method string
args []interface{}
args interface{}
}

3
servlet_handler.go
View File

@ -10,10 +10,7 @@ type ServletHandler interface {
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)

12
servlet_handlers.go
View File

@ -29,22 +29,10 @@ func (sh *ServletHandlers) Init(servletCTX ServletContext) error {
return nil
}
func (sh *ServletHandlers) ReadRequest(servletCTX ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
return nil, fmt.Errorf("Servlet Handler: ReadRequest is not implemented")
}
func (sh *ServletHandlers) Invoke(servletCTX ServletContext, requestCodec protocol.RegistryCodec) (result interface{}, err error) {
return nil, fmt.Errorf("Servlet Handler: Invoke is not implemented")
}
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")
}
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")
}
func (sh *ServletHandlers) Destroy(servletCTX ServletContext) {
// no op
}

11
servlet_rwc_handler.go Normal file
View File

@ -0,0 +1,11 @@
package rpc
import "git.loafle.net/commons_go/rpc/protocol"
type ServletReadWriteCloseHandler interface {
ReadRequest(servletCTX ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error)
WriteResponse(servletCTX ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error
WriteNotification(servletCTX ServletContext, conn interface{}, codec protocol.ServerCodec, method string, params interface{}) error
Validate()
}

26
servlet_rwc_handlers.go Normal file
View File

@ -0,0 +1,26 @@
package rpc
import (
"fmt"
"git.loafle.net/commons_go/rpc/protocol"
)
type ServletReadWriteCloseHandlers struct {
}
func (srwch *ServletReadWriteCloseHandlers) ReadRequest(servletCTX ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
return nil, fmt.Errorf("Servlet RWC Handler: ReadRequest is not implemented")
}
func (srwch *ServletReadWriteCloseHandlers) WriteResponse(servletCTX ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error {
return fmt.Errorf("Servlet RWC Handler: WriteResponse is not implemented")
}
func (srwch *ServletReadWriteCloseHandlers) WriteNotification(servletCTX ServletContext, conn interface{}, codec protocol.ServerCodec, method string, params interface{}) error {
return fmt.Errorf("Servlet RWC Handler: WriteNotification is not implemented")
}
func (srwch *ServletReadWriteCloseHandlers) Validate() {
}