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 2018-03-22 22:35:08 +09:00
parent e3a873a067
commit 2423b22685
7 changed files with 77 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
client/client_rwc_handlers.go
View File

@ -2,11 +2,14 @@ package client
import ( import (
"fmt" "fmt"
"sync"
"git.loafle.net/commons_go/rpc/protocol" "git.loafle.net/commons_go/rpc/protocol"
) )
type ClientReadWriteCloseHandlers struct { type ClientReadWriteCloseHandlers struct {
ReadMtx sync.RWMutex
WriteMtx sync.RWMutex
} }
func (crwch *ClientReadWriteCloseHandlers) Connect(clientCTX ClientContext) (interface{}, error) { func (crwch *ClientReadWriteCloseHandlers) Connect(clientCTX ClientContext) (interface{}, error) {

14
client/rwc/socket/client_rwc_handlers.go
View File

@ -26,19 +26,25 @@ func (crwch *ClientReadWriteCloseHandlers) Connect(clientCTX client.ClientContex
func (crwch *ClientReadWriteCloseHandlers) ReadResponse(clientCTX client.ClientContext, codec protocol.ClientCodec, conn interface{}) (protocol.ClientResponseCodec, error) { func (crwch *ClientReadWriteCloseHandlers) ReadResponse(clientCTX client.ClientContext, codec protocol.ClientCodec, conn interface{}) (protocol.ClientResponseCodec, error) {
soc := conn.(csc.Socket) soc := conn.(csc.Socket)
crwch.ReadMtx.RLock()
resCodec, err := codec.NewResponse(soc) resCodec, err := codec.NewResponse(soc)
crwch.ReadMtx.RUnlock()
return resCodec, err return resCodec, err
} }
func (crwch *ClientReadWriteCloseHandlers) WriteRequest(clientCTX client.ClientContext, codec protocol.ClientCodec, conn interface{}, method string, params []interface{}, id interface{}) error { func (crwch *ClientReadWriteCloseHandlers) WriteRequest(clientCTX client.ClientContext, codec protocol.ClientCodec, conn interface{}, method string, params []interface{}, id interface{}) error {
var (
wErr error
)
soc := conn.(csc.Socket) soc := conn.(csc.Socket)
if wErr := codec.WriteRequest(soc, method, params, id); nil != wErr { crwch.WriteMtx.RLock()
return wErr wErr = codec.WriteRequest(soc, method, params, id)
} crwch.WriteMtx.RUnlock()
return nil return wErr
} }
func (crwch *ClientReadWriteCloseHandlers) Disconnect(clientCTX client.ClientContext, conn interface{}) { func (crwch *ClientReadWriteCloseHandlers) Disconnect(clientCTX client.ClientContext, conn interface{}) {

22
client/rwc/websocket/fasthttp/client_rwc_handlers.go
View File

@ -1,6 +1,8 @@
package fasthttp package fasthttp
import ( import (
"io"
"git.loafle.net/commons_go/logging" "git.loafle.net/commons_go/logging"
"git.loafle.net/commons_go/rpc/client" "git.loafle.net/commons_go/rpc/client"
"git.loafle.net/commons_go/rpc/protocol" "git.loafle.net/commons_go/rpc/protocol"
@ -27,29 +29,37 @@ func (crwch *ClientReadWriteCloseHandlers) Connect(clientCTX client.ClientContex
func (crwch *ClientReadWriteCloseHandlers) ReadResponse(clientCTX client.ClientContext, codec protocol.ClientCodec, conn interface{}) (protocol.ClientResponseCodec, error) { func (crwch *ClientReadWriteCloseHandlers) ReadResponse(clientCTX client.ClientContext, codec protocol.ClientCodec, conn interface{}) (protocol.ClientResponseCodec, error) {
soc := conn.(cwfc.Socket) soc := conn.(cwfc.Socket)
_, r, err := soc.NextReader()
crwch.ReadMtx.RLock()
_, r, err := soc.NextReader()
resCodec, err := codec.NewResponse(r) resCodec, err := codec.NewResponse(r)
crwch.ReadMtx.RUnlock()
return resCodec, err return resCodec, err
} }
func (crwch *ClientReadWriteCloseHandlers) WriteRequest(clientCTX client.ClientContext, codec protocol.ClientCodec, conn interface{}, method string, params []interface{}, id interface{}) error { func (crwch *ClientReadWriteCloseHandlers) WriteRequest(clientCTX client.ClientContext, codec protocol.ClientCodec, conn interface{}, method string, params []interface{}, id interface{}) error {
var (
wc io.WriteCloser
wErr error
)
soc := conn.(cwfc.Socket) soc := conn.(cwfc.Socket)
wc, wErr := soc.NextWriter(websocket.TextMessage) crwch.WriteMtx.RLock()
wc, wErr = soc.NextWriter(websocket.TextMessage)
if nil != wErr { if nil != wErr {
crwch.WriteMtx.RUnlock()
return wErr return wErr
} }
defer func() { defer func() {
wc.Close() wc.Close()
}() }()
if wErr := codec.WriteRequest(wc, method, params, id); nil != wErr { wErr = codec.WriteRequest(wc, method, params, id)
return wErr
}
return nil crwch.WriteMtx.RUnlock()
return wErr
} }
func (crwch *ClientReadWriteCloseHandlers) Disconnect(clientCTX client.ClientContext, conn interface{}) { func (crwch *ClientReadWriteCloseHandlers) Disconnect(clientCTX client.ClientContext, conn interface{}) {

1
protocol/json/client_response.go
View File

@ -66,6 +66,7 @@ func (crc *ClientResponseCodec) Notification() (protocol.ClientNotificationCodec
// newClientMessageCodec returns a new ClientMessageCodec. // newClientMessageCodec returns a new ClientMessageCodec.
func newClientResponseCodec(r io.Reader, codec codec.Codec) (protocol.ClientResponseCodec, error) { func newClientResponseCodec(r io.Reader, codec codec.Codec) (protocol.ClientResponseCodec, error) {
decoder := json.NewDecoder(r) decoder := json.NewDecoder(r)
if nil == r { if nil == r {
return nil, io.EOF return nil, io.EOF

29
server/rwc/socket/servlet_rwc_handlers.go
View File

@ -16,35 +16,42 @@ type ServletReadWriteCloseHandlers struct {
func (srwch *ServletReadWriteCloseHandlers) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) { func (srwch *ServletReadWriteCloseHandlers) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
soc := conn.(server.Socket) soc := conn.(server.Socket)
srwch.ReadMtx.RLock()
reqCodec, err := codec.NewRequest(soc) reqCodec, err := codec.NewRequest(soc)
srwch.ReadMtx.RUnlock()
return reqCodec, err return reqCodec, err
} }
func (srwch *ServletReadWriteCloseHandlers) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, reqCodec protocol.ServerRequestCodec, result interface{}, err error) error { func (srwch *ServletReadWriteCloseHandlers) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, reqCodec protocol.ServerRequestCodec, result interface{}, err error) error {
var (
wErr error
)
soc := conn.(server.Socket) soc := conn.(server.Socket)
srwch.WriteMtx.RLock()
if nil != err { if nil != err {
if wErr := reqCodec.WriteError(soc, 500, err); nil != wErr { wErr = reqCodec.WriteError(soc, 500, err)
return wErr
}
} else { } else {
if wErr := reqCodec.WriteResponse(soc, result); nil != wErr { wErr = reqCodec.WriteResponse(soc, result)
return wErr
}
} }
srwch.WriteMtx.RUnlock()
return nil return wErr
} }
func (srwch *ServletReadWriteCloseHandlers) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, params []interface{}) error { func (srwch *ServletReadWriteCloseHandlers) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, params []interface{}) error {
var (
wErr error
)
soc := conn.(server.Socket) soc := conn.(server.Socket)
if wErr := codec.WriteNotification(soc, method, params); nil != wErr { srwch.WriteMtx.RLock()
return wErr wErr = codec.WriteNotification(soc, method, params)
} srwch.WriteMtx.RUnlock()
return nil return wErr
} }
func (srwch *ServletReadWriteCloseHandlers) Validate() { func (srwch *ServletReadWriteCloseHandlers) Validate() {

40
server/rwc/websocket/fasthttp/servlet_rwc_handlers.go
View File

@ -1,6 +1,8 @@
package fasthttp package fasthttp
import ( import (
"io"
"git.loafle.net/commons_go/rpc" "git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/rpc/protocol" "git.loafle.net/commons_go/rpc/protocol"
cwf "git.loafle.net/commons_go/websocket_fasthttp" cwf "git.loafle.net/commons_go/websocket_fasthttp"
@ -17,18 +19,26 @@ type ServletReadWriteCloseHandlers struct {
func (srwch *ServletReadWriteCloseHandlers) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) { func (srwch *ServletReadWriteCloseHandlers) ReadRequest(servletCTX rpc.ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {
soc := conn.(cwf.Socket) soc := conn.(cwf.Socket)
_, r, err := soc.NextReader()
srwch.ReadMtx.RLock()
_, r, err := soc.NextReader()
requestCodec, err := codec.NewRequest(r) requestCodec, err := codec.NewRequest(r)
srwch.ReadMtx.RUnlock()
return requestCodec, err return requestCodec, err
} }
func (srwch *ServletReadWriteCloseHandlers) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error { func (srwch *ServletReadWriteCloseHandlers) WriteResponse(servletCTX rpc.ServletContext, conn interface{}, requestCodec protocol.ServerRequestCodec, result interface{}, err error) error {
var (
wc io.WriteCloser
wErr error
)
soc := conn.(cwf.Socket) soc := conn.(cwf.Socket)
wc, wErr := soc.NextWriter(websocket.TextMessage) srwch.WriteMtx.RLock()
wc, wErr = soc.NextWriter(websocket.TextMessage)
if nil != wErr { if nil != wErr {
srwch.WriteMtx.RUnlock()
return wErr return wErr
} }
@ -37,23 +47,26 @@ func (srwch *ServletReadWriteCloseHandlers) WriteResponse(servletCTX rpc.Servlet
}() }()
if nil != err { if nil != err {
if wErr := requestCodec.WriteError(wc, 500, err); nil != wErr { wErr = requestCodec.WriteError(wc, 500, err)
return wErr
}
} else { } else {
if wErr := requestCodec.WriteResponse(wc, result); nil != wErr { wErr = requestCodec.WriteResponse(wc, result)
return wErr
}
} }
srwch.WriteMtx.RUnlock()
return nil return wErr
} }
func (srwch *ServletReadWriteCloseHandlers) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, params []interface{}) error { func (srwch *ServletReadWriteCloseHandlers) WriteNotification(servletCTX rpc.ServletContext, conn interface{}, codec protocol.ServerCodec, method string, params []interface{}) error {
var (
wc io.WriteCloser
wErr error
)
soc := conn.(cwf.Socket) soc := conn.(cwf.Socket)
wc, wErr := soc.NextWriter(websocket.TextMessage) srwch.WriteMtx.RLock()
wc, wErr = soc.NextWriter(websocket.TextMessage)
if nil != wErr { if nil != wErr {
srwch.WriteMtx.RUnlock()
return wErr return wErr
} }
@ -61,11 +74,10 @@ func (srwch *ServletReadWriteCloseHandlers) WriteNotification(servletCTX rpc.Ser
wc.Close() wc.Close()
}() }()
if wErr := codec.WriteNotification(wc, method, params); nil != wErr { wErr = codec.WriteNotification(wc, method, params)
return wErr srwch.WriteMtx.RUnlock()
}
return nil return wErr
} }
func (srwch *ServletReadWriteCloseHandlers) Validate() { func (srwch *ServletReadWriteCloseHandlers) Validate() {

3
servlet_rwc_handlers.go
View File

@ -2,11 +2,14 @@ package rpc
import ( import (
"fmt" "fmt"
"sync"
"git.loafle.net/commons_go/rpc/protocol" "git.loafle.net/commons_go/rpc/protocol"
) )
type ServletReadWriteCloseHandlers struct { type ServletReadWriteCloseHandlers struct {
ReadMtx sync.RWMutex
WriteMtx sync.RWMutex
} }
func (srwch *ServletReadWriteCloseHandlers) ReadRequest(servletCTX ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) { func (srwch *ServletReadWriteCloseHandlers) ReadRequest(servletCTX ServletContext, codec protocol.ServerCodec, conn interface{}) (protocol.ServerRequestCodec, error) {