18 Commits

Author SHA1 Message Date
498abc18b4 delete timestamp from logger 2023-05-18 13:44:39 +02:00
5ea1af7fb1 disable default log timestamp 2023-05-18 13:34:54 +02:00
bc578cf25c fixed mistakes 2023-05-18 12:56:37 +02:00
cb6d26a7cc added listener close() 2023-05-18 12:53:37 +02:00
f814bb7d24 go fmt 2023-05-18 11:12:35 +02:00
aed740c933 added TLS transport 2023-05-18 11:12:16 +02:00
4716c25cf6 rename Handler to Handle 2023-05-17 22:35:35 +02:00
9bc3aef167 updated samples 2023-05-17 13:06:13 +02:00
0e5321d51a encoder changed to json 2023-05-17 13:03:52 +02:00
6f9835f399 context added 2023-05-17 12:59:01 +02:00
ef8db1d198 context added 2023-05-17 12:52:41 +02:00
c20faaecb3 Little refactoring 2023-04-01 01:03:38 +02:00
b5b293329e Little refactoring 2023-04-01 00:33:59 +02:00
8b3e722ea5 Little refactoring 2023-04-01 00:18:44 +02:00
8dc753cd95 update to go 1.19 2023-02-11 07:55:46 +02:00
59d850da3c go fmt 2023-02-11 07:52:40 +02:00
5366bd7ff7 Update README.md 2022-12-07 10:35:06 +02:00
f76cbc32aa changed: server stopping, dial/listen to tcp dial/listen, json rpc encoding to msgpack, info getters for logging ; added keepalive setter to server, etc 2022-07-26 14:26:40 +02:00
28 changed files with 1588 additions and 1415 deletions

View File

@@ -1,5 +1,6 @@
# dsrpc, Data RPC # dsrpc, Data RPC
DSRPC is easy and simple RPC framework over TCP socket. DSRPC is easy and simple RPC framework over TCP socket.
### Purpose ### Purpose
@@ -43,7 +44,7 @@ func server() error {
serv := dsrpc.NewService() serv := dsrpc.NewService()
cont := NewController() cont := NewController()
serv.Handler(api.HelloMethod, cont.HelloHandler) serv.Handle(api.HelloMethod, cont.HelloHandler)
serv.PreMiddleware(dsrpc.LogRequest) serv.PreMiddleware(dsrpc.LogRequest)
serv.PostMiddleware(dsrpc.LogResponse) serv.PostMiddleware(dsrpc.LogResponse)
@@ -219,9 +220,9 @@ func sampleServ(quiet bool) error {
serv.PreMiddleware(authMiddleware) serv.PreMiddleware(authMiddleware)
serv.PreMiddleware(dsrpc.LogRequest) serv.PreMiddleware(dsrpc.LogRequest)
serv.Handler(HelloMethod, helloHandler) serv.Handle(HelloMethod, helloHandler)
serv.Handler(SaveMethod, saveHandler) serv.Handle(SaveMethod, saveHandler)
serv.Handler(LoadMethod, loadHandler) serv.Handle(LoadMethod, loadHandler)
serv.PostMiddleware(dsrpc.LogResponse) serv.PostMiddleware(dsrpc.LogResponse)
serv.PostMiddleware(dsrpc.LogAccess) serv.PostMiddleware(dsrpc.LogAccess)

531
client.go
View File

@@ -1,285 +1,348 @@
/* /*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org> * Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/ */
package dsrpc package dsrpc
import ( import (
"encoding/json" "context"
"errors" "crypto/tls"
"io" "errors"
"net" "fmt"
"sync" "io"
"net"
"sync"
encoder "encoding/json"
) )
func Put(ctx context.Context, address string, method string, reader io.Reader, binSize int64, param, result any, auth *Auth) error {
var err error
func Put(address string, method string, reader io.Reader, size int64, param, result any, auth *Auth) error { addr, err := net.ResolveTCPAddr("tcp", address)
var err error if err != nil {
err = fmt.Errorf("unable to resolve adddress: %s", err)
return err
}
conn, err := net.DialTCP("tcp", nil, addr)
if err != nil {
return err
}
defer conn.Close()
conn, err := net.Dial("tcp", address) return ConnPut(ctx, conn, method, reader, binSize, param, result, auth)
if err != nil {
return Err(err)
}
defer conn.Close()
return ConnPut(conn, method, reader, size, param, result, auth)
} }
func PutTLS(ctx context.Context, tlsConfig *tls.Config, address string, method string, reader io.Reader, binSize int64, param, result any, auth *Auth) error {
var err error
func ConnPut(conn net.Conn, method string, reader io.Reader, size int64, param, result any, auth *Auth) error { conn, err := tls.Dial("tcp", address, tlsConfig)
var err error if err != nil {
context := CreateContext(conn) return err
context.reqRPC.Method = method }
context.reqRPC.Params = param defer conn.Close()
context.reqRPC.Auth = auth
context.resRPC.Result = result
context.binReader = reader return ConnPut(ctx, conn, method, reader, binSize, param, result, auth)
context.binWriter = conn
context.reqHeader.binSize = size
if context.reqRPC.Params == nil {
context.reqRPC.Params = NewEmpty()
}
err = context.CreateRequest()
if err != nil {
return Err(err)
}
err = context.WriteRequest()
if err != nil {
return Err(err)
}
var wg sync.WaitGroup
errChan := make(chan error, 1)
wg.Add(1)
go context.ReadResponseAsync(&wg, errChan)
wg.Add(1)
go context.UploadBinAsync(&wg)
wg.Wait()
err = <- errChan
if err != nil {
return Err(err)
}
err = context.BindResponse()
if err != nil {
return Err(err)
}
return Err(err)
} }
func Get(address string, method string, writer io.Writer, param, result any, auth *Auth) error { func ConnPut(ctx context.Context, conn net.Conn, method string, reader io.Reader, binSize int64, param, result any, auth *Auth) error {
var err error var err error
content := CreateContent(conn)
conn, err := net.Dial("tcp", address) content.reqBlock.Method = method
if err != nil { if param != nil {
return Err(err) content.reqBlock.Params = param
} }
defer conn.Close() if auth != nil {
content.reqBlock.Auth = auth
}
if result != nil {
content.resBlock.Result = result
}
return ConnGet(conn, method, writer, param, result, auth) content.binReader = reader
content.binWriter = conn
content.reqHeader.binSize = binSize
err = content.createRequest()
if err != nil {
return err
}
err = content.writeRequest()
if err != nil {
return err
}
var wg sync.WaitGroup
errChan := make(chan error, 1)
wg.Add(1)
go content.readResponseAsync(&wg, errChan)
wg.Add(1)
go content.uploadBinAsync(ctx, &wg)
wg.Wait()
err = <-errChan
if err != nil {
return err
}
err = content.bindResponse()
if err != nil {
return err
}
return err
} }
func ConnGet(conn net.Conn, method string, writer io.Writer, param, result any, auth *Auth) error { func Get(ctx context.Context, address string, method string, writer io.Writer, param, result any, auth *Auth) error {
var err error var err error
context := CreateContext(conn) addr, err := net.ResolveTCPAddr("tcp", address)
context.reqRPC.Method = method if err != nil {
context.reqRPC.Params = param err = fmt.Errorf("unable to resolve adddress: %s", err)
context.reqRPC.Auth = auth return err
context.resRPC.Result = result }
conn, err := net.DialTCP("tcp", nil, addr)
if err != nil {
return err
}
defer conn.Close()
context.binReader = conn return ConnGet(ctx, conn, method, writer, param, result, auth)
context.binWriter = writer
if context.reqRPC.Params == nil {
context.reqRPC.Params = NewEmpty()
}
err = context.CreateRequest()
if err != nil {
return Err(err)
}
err = context.WriteRequest()
if err != nil {
return Err(err)
}
err = context.ReadResponse()
if err != nil {
return Err(err)
}
err = context.DownloadBin()
if err != nil {
return Err(err)
}
err = context.BindResponse()
if err != nil {
return Err(err)
}
return Err(err)
} }
func Exec(address, method string, param any, result any, auth *Auth) error { func GetTLS(ctx context.Context, tlsConfig *tls.Config, address string, method string, writer io.Writer, param, result any, auth *Auth) error {
var err error var err error
conn, err := net.Dial("tcp", address)
if err != nil {
return Err(err)
}
defer conn.Close()
err = ConnExec(conn, method, param, result, auth) conn, err := tls.Dial("tcp", address, tlsConfig)
if err != nil { if err != nil {
return Err(err) return err
} }
return Err(err) defer conn.Close()
return ConnGet(ctx, conn, method, writer, param, result, auth)
} }
func ConnGet(ctx context.Context, conn net.Conn, method string, writer io.Writer, param, result any, auth *Auth) error {
var err error
func ConnExec(conn net.Conn, method string, param any, result any, auth *Auth) error { content := CreateContent(conn)
var err error content.reqBlock.Method = method
if param != nil {
content.reqBlock.Params = param
}
if auth != nil {
content.reqBlock.Auth = auth
}
if result != nil {
content.resBlock.Result = result
}
context := CreateContext(conn) content.binReader = conn
context.reqRPC.Method = method content.binWriter = writer
context.reqRPC.Params = param
context.reqRPC.Auth = auth
context.resRPC.Result = result
if context.reqRPC.Params == nil { err = content.createRequest()
context.reqRPC.Params = NewEmpty() if err != nil {
} return err
}
err = context.CreateRequest() err = content.writeRequest()
if err != nil { if err != nil {
return Err(err) return err
} }
err = context.WriteRequest() err = content.readResponse()
if err != nil { if err != nil {
return Err(err) return err
} }
err = context.ReadResponse() err = content.downloadBin(ctx)
if err != nil { if err != nil {
return Err(err) return err
} }
err = context.BindResponse() err = content.bindResponse()
if err != nil { if err != nil {
return Err(err) return err
} }
return Err(err) return err
} }
func Exec(ctx context.Context, address, method string, param any, result any, auth *Auth) error {
var err error
func (context *Context) CreateRequest() error { addr, err := net.ResolveTCPAddr("tcp", address)
var err error if err != nil {
err = fmt.Errorf("unable to resolve adddress: %s", err)
return err
}
conn, err := net.DialTCP("tcp", nil, addr)
if err != nil {
return err
}
defer conn.Close()
context.reqPacket.rcpPayload, err = context.reqRPC.Pack() err = ConnExec(ctx, conn, method, param, result, auth)
if err != nil { if err != nil {
return Err(err) return err
} }
rpcSize := int64(len(context.reqPacket.rcpPayload)) return err
context.reqHeader.rpcSize = rpcSize
context.reqPacket.header, err = context.reqHeader.Pack()
if err != nil {
return Err(err)
}
return Err(err)
} }
func (context *Context) WriteRequest() error { func ExecTLS(ctx context.Context, tlsConfig *tls.Config, address, method string, param any, result any, auth *Auth) error {
var err error var err error
_, err = context.sockWriter.Write(context.reqPacket.header)
if err != nil { conn, err := tls.Dial("tcp", address, tlsConfig)
return Err(err) if err != nil {
} return err
_, err = context.sockWriter.Write(context.reqPacket.rcpPayload) }
if err != nil { defer conn.Close()
return Err(err)
} err = ConnExec(ctx, conn, method, param, result, auth)
return Err(err) if err != nil {
return err
}
return err
} }
func (context *Context) UploadBin() error { func ConnExec(ctx context.Context, conn net.Conn, method string, param any, result any, auth *Auth) error {
var err error var err error
_, err = CopyBytes(context.binReader, context.binWriter, context.reqHeader.binSize)
return Err(err) content := CreateContent(conn)
content.reqBlock.Method = method
if param != nil {
content.reqBlock.Params = param
}
if result != nil {
content.resBlock.Result = result
}
if auth != nil {
content.reqBlock.Auth = auth
}
err = content.createRequest()
if err != nil {
return err
}
err = content.writeRequest()
if err != nil {
return err
}
err = content.readResponse()
if err != nil {
return err
}
err = content.bindResponse()
if err != nil {
return err
}
return err
} }
func (context *Context) ReadResponse() error { func (content *Content) createRequest() error {
var err error var err error
context.resPacket.header, err = ReadBytes(context.sockReader, headerSize) content.reqPacket.rcpPayload, err = content.reqBlock.Pack()
if err != nil { if err != nil {
return Err(err) return err
} }
context.resHeader, err = UnpackHeader(context.resPacket.header) rpcSize := int64(len(content.reqPacket.rcpPayload))
if err != nil { content.reqHeader.rpcSize = rpcSize
return Err(err)
} content.reqPacket.header, err = content.reqHeader.Pack()
rpcSize := context.resHeader.rpcSize if err != nil {
context.resPacket.rcpPayload, err = ReadBytes(context.sockReader, rpcSize) return err
if err != nil { }
return Err(err) return err
}
return Err(err)
} }
func (context *Context) UploadBinAsync(wg *sync.WaitGroup) { func (content *Content) writeRequest() error {
exitFunc := func() { var err error
wg.Done() _, err = content.sockWriter.Write(content.reqPacket.header)
} if err != nil {
defer exitFunc() return err
_, _ = CopyBytes(context.binReader, context.binWriter, context.reqHeader.binSize) }
return _, err = content.sockWriter.Write(content.reqPacket.rcpPayload)
if err != nil {
return err
}
return err
} }
func (context *Context) ReadResponseAsync(wg *sync.WaitGroup, errChan chan error) { func (content *Content) uploadBin(ctx context.Context) error {
var err error var err error
exitFunc := func() { _, err = CopyBytes(ctx, content.binReader, content.binWriter, content.reqHeader.binSize)
errChan <- err return err
wg.Done()
}
defer exitFunc()
context.resPacket.header, err = ReadBytes(context.sockReader, headerSize)
if err != nil {
err = Err(err)
return
}
context.resHeader, err = UnpackHeader(context.resPacket.header)
if err != nil {
err = Err(err)
return
}
rpcSize := context.resHeader.rpcSize
context.resPacket.rcpPayload, err = ReadBytes(context.sockReader, rpcSize)
if err != nil {
err = Err(err)
return
}
return
} }
func (context *Context) DownloadBin() error { func (content *Content) readResponse() error {
var err error var err error
_, err = CopyBytes(context.binReader, context.binWriter, context.resHeader.binSize)
return Err(err) content.resPacket.header, err = ReadBytes(content.sockReader, headerSize)
if err != nil {
return err
}
content.resHeader, err = UnpackHeader(content.resPacket.header)
if err != nil {
return err
}
rpcSize := content.resHeader.rpcSize
content.resPacket.rcpPayload, err = ReadBytes(content.sockReader, rpcSize)
if err != nil {
return err
}
return err
} }
func (context *Context) BindResponse() error { func (content *Content) uploadBinAsync(ctx context.Context, wg *sync.WaitGroup) {
var err error exitFunc := func() {
wg.Done()
err = json.Unmarshal(context.resPacket.rcpPayload, context.resRPC) }
if err != nil { defer exitFunc()
return Err(err) _, _ = CopyBytes(ctx, content.binReader, content.binWriter, content.reqHeader.binSize)
} return
if len(context.resRPC.Error) > 0 { }
err = errors.New(context.resRPC.Error)
return Err(err) func (content *Content) readResponseAsync(wg *sync.WaitGroup, errChan chan error) {
} var err error
return Err(err) exitFunc := func() {
errChan <- err
wg.Done()
}
defer exitFunc()
content.resPacket.header, err = ReadBytes(content.sockReader, headerSize)
if err != nil {
err = err
return
}
content.resHeader, err = UnpackHeader(content.resPacket.header)
if err != nil {
err = err
return
}
rpcSize := content.resHeader.rpcSize
content.resPacket.rcpPayload, err = ReadBytes(content.sockReader, rpcSize)
if err != nil {
err = err
return
}
return
}
func (content *Content) downloadBin(ctx context.Context) error {
var err error
_, err = CopyBytes(ctx, content.binReader, content.binWriter, content.resHeader.binSize)
return err
}
func (content *Content) bindResponse() error {
var err error
err = encoder.Unmarshal(content.resPacket.rcpPayload, content.resBlock)
if err != nil {
return err
}
if len(content.resBlock.Error) > 0 {
err = errors.New(content.resBlock.Error)
return err
}
return err
} }

View File

@@ -1,3 +0,0 @@
package dsrpc
type any = interface{}

145
content.go Normal file
View File

@@ -0,0 +1,145 @@
/*
* Copyright 2022 Oleg Borodin <borodin@unix7.org>
*/
package dsrpc
import (
"io"
"net"
"time"
)
type Content struct {
start time.Time
remoteHost string
sockReader io.Reader
sockWriter io.Writer
reqPacket *Packet
reqHeader *Header
reqBlock *Request
resPacket *Packet
resHeader *Header
resBlock *Response
binReader io.Reader
binWriter io.Writer
}
func CreateContent(conn net.Conn) *Content {
context := &Content{
start: time.Now(),
sockReader: conn,
sockWriter: conn,
reqPacket: NewEmptyPacket(),
reqHeader: NewEmptyHeader(),
reqBlock: NewEmptyRequest(),
resPacket: NewEmptyPacket(),
resHeader: NewEmptyHeader(),
resBlock: NewEmptyResponse(),
}
return context
}
func (context *Content) Request() *Request {
return context.reqBlock
}
func (context *Content) RemoteHost() string {
return context.remoteHost
}
func (context *Content) Start() time.Time {
return context.start
}
func (context *Content) Method() string {
var method string
if context.reqBlock != nil {
method = context.reqBlock.Method
}
return method
}
func (context *Content) ReqRpcSize() int64 {
var size int64
if context.reqHeader != nil {
size = context.reqHeader.rpcSize
}
return size
}
func (context *Content) ReqBinSize() int64 {
var size int64
if context.reqHeader != nil {
size = context.reqHeader.binSize
}
return size
}
func (context *Content) ResBinSize() int64 {
var size int64
if context.resHeader != nil {
size = context.resHeader.binSize
}
return size
}
func (context *Content) ResRpcSize() int64 {
var size int64
if context.resHeader != nil {
size = context.resHeader.rpcSize
}
return size
}
func (context *Content) ReqSize() int64 {
var size int64
if context.reqHeader != nil {
size += context.reqHeader.binSize
size += context.reqHeader.rpcSize
}
return size
}
func (context *Content) ResSize() int64 {
var size int64
if context.resHeader != nil {
size += context.resHeader.binSize
size += context.resHeader.rpcSize
}
return size
}
func (context *Content) SetAuthIdent(ident []byte) {
context.reqBlock.Auth.Ident = ident
}
func (context *Content) SetAuthSalt(salt []byte) {
context.reqBlock.Auth.Salt = salt
}
func (context *Content) SetAuthHash(hash []byte) {
context.reqBlock.Auth.Hash = hash
}
func (context *Content) AuthIdent() []byte {
return context.reqBlock.Auth.Ident
}
func (context *Content) AuthSalt() []byte {
return context.reqBlock.Auth.Salt
}
func (context *Content) AuthHash() []byte {
return context.reqBlock.Auth.Hash
}
func (context *Content) Auth() *Auth {
return context.reqBlock.Auth
}

View File

@@ -1,90 +0,0 @@
/*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/
package dsrpc
import (
"io"
"net"
"time"
)
type Context struct {
start time.Time
remoteHost string
sockReader io.Reader
sockWriter io.Writer
reqHeader *Header
reqRPC *Request
reqPacket *Packet
resPacket *Packet
resHeader *Header
resRPC *Response
binReader io.Reader
binWriter io.Writer
}
func NewContext() *Context {
context := &Context{}
context.start = time.Now()
return context
}
func CreateContext(conn net.Conn) *Context {
context := &Context{}
context.start = time.Now()
context.sockReader = conn
context.sockWriter = conn
context.reqPacket = NewPacket()
context.resPacket = NewPacket()
context.reqHeader = NewHeader()
context.reqRPC = NewRequest()
context.resHeader = NewHeader()
context.resRPC = NewResponse()
context.resRPC = NewResponse()
return context
}
func (context *Context) Request() *Request {
return context.reqRPC
}
func (context *Context) SetAuthIdent(ident []byte) {
context.reqRPC.Auth.Ident = ident
}
func (context *Context) SetAuthSalt(salt []byte) {
context.reqRPC.Auth.Salt = salt
}
func (context *Context) SetAuthHash(hash []byte) {
context.reqRPC.Auth.Hash = hash
}
func (context *Context) AuthIdent() []byte {
return context.reqRPC.Auth.Ident
}
func (context *Context) AuthSalt() []byte {
return context.reqRPC.Auth.Salt
}
func (context *Context) AuthHash() []byte {
return context.reqRPC.Auth.Hash
}
func (context *Context) Auth() *Auth {
return context.reqRPC.Auth
}

View File

@@ -1,13 +0,0 @@
/*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/
package dsrpc
type Empty struct {}
func NewEmpty() *Empty {
return &Empty{}
}

View File

@@ -1,39 +0,0 @@
package dsrpc
import (
"fmt"
"runtime"
"io"
)
var develMode bool = false
var debugMode bool = false
func SetDevelMode(mode bool) {
develMode = mode
}
func SetDebugMode(mode bool) {
debugMode = mode
}
func Err(err error) error {
switch err {
case io.EOF:
return err
}
if err != nil {
switch {
case develMode == true:
pc, filename, line, _ := runtime.Caller(1)
funcName := runtime.FuncForPC(pc).Name()
err = fmt.Errorf(" %s:%d:%s:%s", filename, line, funcName, err.Error())
case debugMode == true:
pc, _, line, _ := runtime.Caller(1)
funcName := runtime.FuncForPC(pc).Name()
err = fmt.Errorf(" %s:%d:%s ", funcName, line, err.Error())
default:
}
}
return err
}

View File

@@ -12,14 +12,7 @@ type HelloParams struct {
Message string `msgpack:"message" json:"message"` Message string `msgpack:"message" json:"message"`
} }
func NewHelloParams() *HelloParams {
return &HelloParams{}
}
type HelloResult struct { type HelloResult struct {
Message string `msgpack:"message" json:"message"` Message string `msgpack:"message" json:"message"`
} }
func NewHelloResult() *HelloResult {
return &HelloResult{}
}

View File

@@ -7,32 +7,40 @@
package main package main
import ( import (
"fmt" "context"
"github.com/kindsoldier/dsrpc" "fmt"
"netsrv/api" "time"
"github.com/kindsoldier/dsrpc"
"netsrv/api"
) )
func main() { func main() {
err := exec() err := exec()
if err != nil { if err != nil {
fmt.Println("exec err:", err) fmt.Println("exec err:", err)
} }
} }
func exec() error { func exec() error {
var err error var err error
params := api.NewHelloParams() params := api.HelloParams{
params.Message = "hello, server!" Message: "hello, server!",
}
result := api.NewHelloResult() result := api.HelloResult{}
err = dsrpc.Exec("127.0.0.1:8081", api.HelloMethod, params, result, nil) ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
if err != nil { defer cancel()
return err
}
fmt.Println("result:", result.Message) err = dsrpc.Exec(ctx, "127.0.0.1:8081", api.HelloMethod, &params, &result, nil)
return err if err != nil {
return err
}
fmt.Println("result:", result.Message)
return err
} }

View File

@@ -1,7 +1,5 @@
module netsrv module netsrv
go 1.17 go 1.19
require github.com/kindsoldier/dsrpc v0.0.1 require github.com/kindsoldier/dsrpc v1.2.1
replace github.com/kindsoldier/dsrpc => ../

View File

@@ -1,10 +1,6 @@
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/kindsoldier/dsrpc v1.2.1 h1:sw1a3MAD83Do1Fu+Dh+AHArrwVMgZ/KTLUWWTkQ6vj8=
github.com/kindsoldier/dsrpc v1.2.1/go.mod h1:zYb5yYfE/18BYK+iCUNcpkZ4uArwUNNhwYkUK8xDHQk=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

View File

@@ -8,8 +8,9 @@ package main
import ( import (
"log" "log"
"github.com/kindsoldier/dsrpc"
"netsrv/api" "netsrv/api"
"github.com/kindsoldier/dsrpc"
) )
func main() { func main() {
@@ -25,7 +26,7 @@ func server() error {
serv := dsrpc.NewService() serv := dsrpc.NewService()
cont := NewController() cont := NewController()
serv.Handler(api.HelloMethod, cont.HelloHandler) serv.Handle(api.HelloMethod, cont.HelloHandler)
serv.PreMiddleware(dsrpc.LogRequest) serv.PreMiddleware(dsrpc.LogRequest)
serv.PostMiddleware(dsrpc.LogResponse) serv.PostMiddleware(dsrpc.LogResponse)
@@ -46,20 +47,19 @@ func NewController() *Controller {
return &Controller{} return &Controller{}
} }
func (cont *Controller) HelloHandler(context *dsrpc.Context) error { func (cont *Controller) HelloHandler(content *dsrpc.Content) error {
var err error var err error
params := api.NewHelloParams() params := api.HelloParams{}
err = context.BindParams(params) err = content.BindParams(&params)
if err != nil { if err != nil {
return err return err
} }
log.Println("hello message:", params.Message) log.Println("hello message:", params.Message)
result := api.NewHelloResult() result := api.HelloResult{
result.Message = "hello!" Message: "hello, client!",
}
err = context.SendResult(result, 0) err = content.SendResult(&result, 0)
if err != nil { if err != nil {
return err return err
} }

View File

@@ -1,376 +1,379 @@
/* /*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org> * Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/ */
package dsrpc package dsrpc
import ( import (
"bytes" "bytes"
"encoding/json" "context"
"errors" "encoding/json"
"io" "errors"
"math/rand" "io"
"testing" "math/rand"
"time" "testing"
"time"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
func TestLocalExec(t *testing.T) {
var err error
params := NewHelloParams()
params.Message = "hello server!"
result := NewHelloResult()
auth := CreateAuth([]byte("qwert"), []byte("12345"))
err = LocalExec(HelloMethod, params, result, auth, helloHandler)
require.NoError(t, err)
resultJSON, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON))
}
func TestLocalSave(t *testing.T) {
var err error
params := NewSaveParams()
params.Message = "save data!"
result := NewHelloResult()
auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16
rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize)
rand.Read(binBytes)
reader := bytes.NewReader(binBytes)
err = LocalPut(SaveMethod, reader, binSize, params, result, auth, saveHandler)
require.NoError(t, err)
resultJSON, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON))
}
func TestLocalLoad(t *testing.T) {
var err error
params := NewLoadParams()
params.Message = "load data!"
result := NewHelloResult()
auth := CreateAuth([]byte("qwert"), []byte("12345"))
binBytes := make([]byte, 0)
writer := bytes.NewBuffer(binBytes)
err = LocalGet(LoadMethod, writer, params, result, auth, loadHandler)
require.NoError(t, err)
resultJSON, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON))
logDebug("bin size:", len(writer.Bytes()))
}
func TestNetExec(t *testing.T) {
go testServ(false)
time.Sleep(10 * time.Millisecond)
err := clientHello()
require.NoError(t, err)
}
func TestNetSave(t *testing.T) {
go testServ(false)
time.Sleep(10 * time.Millisecond)
err := clientSave()
require.NoError(t, err)
}
func TestNetLoad(t *testing.T) {
go testServ(false)
time.Sleep(10 * time.Millisecond)
err := clientLoad()
require.NoError(t, err)
}
func BenchmarkNetPut(b *testing.B) {
go testServ(true)
time.Sleep(10 * time.Millisecond)
clientSave()
pBench := func(pb *testing.PB) {
for pb.Next() {
clientSave()
}
}
b.SetParallelism(10)
b.RunParallel(pBench)
}
func clientHello() error {
var err error
params := NewHelloParams()
params.Message = "hello server!"
result := NewHelloResult()
auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16
rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize)
rand.Read(binBytes)
err = Exec("127.0.0.1:8081", HelloMethod, params, result, auth)
if err != nil {
logError("method err:", err)
return err
}
resultJSON, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON))
return err
}
func clientSave() error {
var err error
params := NewSaveParams()
params.Message = "save data!"
result := NewHelloResult()
auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16
rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize)
rand.Read(binBytes)
reader := bytes.NewReader(binBytes)
err = Put("127.0.0.1:8081", SaveMethod, reader, binSize, params, result, auth)
if err != nil {
logError("method err:", err)
return err
}
resultJSON, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON))
return err
}
func clientLoad() error {
var err error
params := NewLoadParams()
params.Message = "load data!"
result := NewHelloResult()
auth := CreateAuth([]byte("qwert"), []byte("12345"))
binBytes := make([]byte, 0)
writer := bytes.NewBuffer(binBytes)
err = Get("127.0.0.1:8081", LoadMethod, writer, params, result, auth)
if err != nil {
logError("method err:", err)
return err
}
resultJSON, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON))
logDebug("bin size:", len(writer.Bytes()))
return err
}
var testServRun bool = false
func testServ(quiet bool) error {
var err error
if testServRun {
return err
}
testServRun = true
if quiet {
SetAccessWriter(io.Discard)
SetMessageWriter(io.Discard)
}
serv := NewService()
serv.Handler(HelloMethod, helloHandler)
serv.Handler(SaveMethod, saveHandler)
serv.Handler(LoadMethod, loadHandler)
serv.PreMiddleware(LogRequest)
serv.PreMiddleware(auth)
serv.PostMiddleware(LogResponse)
serv.PostMiddleware(LogAccess)
err = serv.Listen(":8081")
if err != nil {
return err
}
return err
}
func auth(context *Context) error {
var err error
reqIdent := context.AuthIdent()
reqSalt := context.AuthSalt()
reqHash := context.AuthHash()
ident := reqIdent
pass := []byte("12345")
auth := context.Auth()
logDebug("auth ", string(auth.JSON()))
ok := CheckHash(ident, pass, reqSalt, reqHash)
logDebug("auth ok:", ok)
if !ok {
err = errors.New("auth ident or pass missmatch")
context.SendError(err)
return err
}
return err
}
func helloHandler(context *Context) error {
var err error
params := NewHelloParams()
err = context.BindParams(params)
if err != nil {
return err
}
err = context.ReadBin(io.Discard)
if err != nil {
context.SendError(err)
return err
}
result := NewHelloResult()
result.Message = "hello, client!"
err = context.SendResult(result, 0)
if err != nil {
return err
}
return err
}
func saveHandler(context *Context) error {
var err error
params := NewSaveParams()
err = context.BindParams(params)
if err != nil {
return err
}
bufferBytes := make([]byte, 0, 1024)
binWriter := bytes.NewBuffer(bufferBytes)
err = context.ReadBin(binWriter)
if err != nil {
context.SendError(err)
return err
}
result := NewSaveResult()
result.Message = "saved successfully!"
err = context.SendResult(result, 0)
if err != nil {
return err
}
return err
}
func loadHandler(context *Context) error {
var err error
params := NewSaveParams()
err = context.BindParams(params)
if err != nil {
return err
}
err = context.ReadBin(io.Discard)
if err != nil {
context.SendError(err)
return err
}
var binSize int64 = 1024
rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize)
rand.Read(binBytes)
binReader := bytes.NewReader(binBytes)
result := NewSaveResult()
result.Message = "load successfully!"
err = context.SendResult(result, binSize)
if err != nil {
return err
}
binWriter := context.BinWriter()
_, err = CopyBytes(binReader, binWriter, binSize)
if err != nil {
return err
}
return err
}
const HelloMethod string = "hello" const HelloMethod string = "hello"
type HelloParams struct { type HelloParams struct {
Message string `json:"message" json:"message"` Message string `json:"message" msgpack:"message"`
}
func NewHelloParams() *HelloParams {
return &HelloParams{}
} }
type HelloResult struct { type HelloResult struct {
Message string `json:"message" json:"message"` Message string `json:"message" msgpack:"message"`
} }
func NewHelloResult() *HelloResult {
return &HelloResult{}
}
const SaveMethod string = "save" const SaveMethod string = "save"
type SaveParams HelloParams type SaveParams HelloParams
type SaveResult HelloResult type SaveResult HelloResult
func NewSaveParams() *SaveParams {
return &SaveParams{}
}
func NewSaveResult() *SaveResult {
return &SaveResult{}
}
const LoadMethod string = "load" const LoadMethod string = "load"
type LoadParams HelloParams type LoadParams HelloParams
type LoadResult HelloResult type LoadResult HelloResult
func NewLoadParams() *LoadParams { func TestLocalExec(t *testing.T) {
return &LoadParams{} var err error
params := HelloParams{}
params.Message = "hello server!"
result := HelloResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345"))
err = LocalExec(HelloMethod, &params, &result, auth, helloHandler)
require.NoError(t, err)
resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJson))
} }
func NewLoadResult() *LoadResult {
return &LoadResult{} func TestLocalSave(t *testing.T) {
var err error
params := SaveParams{}
params.Message = "save data!"
result := SaveResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16
rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize)
rand.Read(binBytes)
reader := bytes.NewReader(binBytes)
timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = LocalPut(ctx, SaveMethod, reader, binSize, &params, &result, auth, saveHandler)
require.NoError(t, err)
resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJson))
}
func TestLocalLoad(t *testing.T) {
var err error
params := LoadParams{}
params.Message = "load data!"
result := LoadResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345"))
binBytes := make([]byte, 0)
writer := bytes.NewBuffer(binBytes)
timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = LocalGet(ctx, LoadMethod, writer, &params, &result, auth, loadHandler)
require.NoError(t, err)
resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJson))
logDebug("bin size:", len(writer.Bytes()))
}
func TestNetExec(t *testing.T) {
go testServ(false)
time.Sleep(100 * time.Millisecond)
err := clientHello()
require.NoError(t, err)
}
func TestNetSave(t *testing.T) {
go testServ(false)
time.Sleep(100 * time.Millisecond)
err := clientSave()
require.NoError(t, err)
}
func TestNetLoad(t *testing.T) {
go testServ(false)
time.Sleep(100 * time.Millisecond)
err := clientLoad()
require.NoError(t, err)
}
func BenchmarkNetPut(b *testing.B) {
go testServ(true)
time.Sleep(1000 * time.Millisecond)
clientSave()
pBench := func(pb *testing.PB) {
for pb.Next() {
clientSave()
}
}
b.SetParallelism(2000)
b.RunParallel(pBench)
}
func clientHello() error {
var err error
params := HelloParams{}
params.Message = "hello server!"
result := HelloResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16
rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize)
rand.Read(binBytes)
timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = Exec(ctx, "127.0.0.1:18081", HelloMethod, &params, &result, auth)
if err != nil {
logError("method err:", err)
return err
}
resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJson))
return err
}
func clientSave() error {
var err error
params := SaveParams{}
params.Message = "save data!"
result := SaveResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16
rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize)
rand.Read(binBytes)
reader := bytes.NewReader(binBytes)
timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = Put(ctx, "127.0.0.1:18081", SaveMethod, reader, binSize, &params, &result, auth)
if err != nil {
logError("method err:", err)
return err
}
resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJson))
return err
}
func clientLoad() error {
var err error
params := LoadParams{}
params.Message = "load data!"
result := LoadResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345"))
binBytes := make([]byte, 0)
writer := bytes.NewBuffer(binBytes)
timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = Get(ctx, "127.0.0.1:18081", LoadMethod, writer, &params, &result, auth)
if err != nil {
logError("method err:", err)
return err
}
resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJson))
logDebug("bin size:", len(writer.Bytes()))
return err
}
var testServRun bool = false
func testServ(quiet bool) error {
var err error
if testServRun {
return err
}
testServRun = true
if quiet {
SetAccessWriter(io.Discard)
SetMessageWriter(io.Discard)
}
serv := NewService()
serv.Handle(HelloMethod, helloHandler)
serv.Handle(SaveMethod, saveHandler)
serv.Handle(LoadMethod, loadHandler)
serv.PreMiddleware(LogRequest)
serv.PreMiddleware(auth)
serv.PostMiddleware(LogResponse)
serv.PostMiddleware(LogAccess)
err = serv.Listen(":18081")
if err != nil {
return err
}
return err
}
func auth(content *Content) error {
var err error
reqIdent := content.AuthIdent()
reqSalt := content.AuthSalt()
reqHash := content.AuthHash()
ident := reqIdent
pass := []byte("12345")
auth := content.Auth()
logDebug("auth ", string(auth.Json()))
ok := CheckHash(ident, pass, reqSalt, reqHash)
logDebug("auth ok:", ok)
if !ok {
err = errors.New("auth ident or pass missmatch")
content.SendError(err)
return err
}
return err
}
func helloHandler(content *Content) error {
var err error
params := HelloParams{}
err = content.BindParams(&params)
if err != nil {
return err
}
timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, io.Discard)
if err != nil {
content.SendError(err)
return err
}
result := HelloResult{}
result.Message = "hello, client!"
err = content.SendResult(result, 0)
if err != nil {
return err
}
return err
}
func saveHandler(content *Content) error {
var err error
params := SaveParams{}
err = content.BindParams(&params)
if err != nil {
return err
}
bufferBytes := make([]byte, 0, 1024)
binWriter := bytes.NewBuffer(bufferBytes)
timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, binWriter)
if err != nil {
content.SendError(err)
return err
}
result := SaveResult{}
result.Message = "saved successfully!"
err = content.SendResult(result, 0)
if err != nil {
return err
}
return err
}
func loadHandler(content *Content) error {
var err error
params := SaveParams{}
err = content.BindParams(&params)
if err != nil {
return err
}
timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, io.Discard)
if err != nil {
content.SendError(err)
return err
}
var binSize int64 = 1024
rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize)
rand.Read(binBytes)
binReader := bytes.NewReader(binBytes)
result := SaveResult{}
result.Message = "load successfully!"
err = content.SendResult(result, binSize)
if err != nil {
return err
}
binWriter := content.BinWriter()
_, err = CopyBytes(ctx, binReader, binWriter, binSize)
if err != nil {
return err
}
return err
} }

View File

@@ -1,27 +1,26 @@
/* /*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org> * Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/ */
package dsrpc package dsrpc
type FAddr struct { type FAddr struct {
network string network string
address string address string
} }
func NewFAddr() *FAddr { func NewFAddr() *FAddr {
var addr FAddr addr := FAddr{
addr.network = "tcp" network: "tcp",
addr.address = "127.0.0.1:5000" address: "127.0.0.1:5000",
return &addr }
return &addr
} }
func (addr *FAddr) Network() string { func (addr *FAddr) Network() string {
return addr.network return addr.network
} }
func (addr *FAddr) String() string { func (addr *FAddr) String() string {
return addr.address return addr.address
} }

View File

@@ -7,51 +7,52 @@
package dsrpc package dsrpc
import ( import (
"net" "net"
"testing" "testing"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/require"
) )
func TestFConn0(t *testing.T) { func TestFConn0(t *testing.T) {
var cConn, sConn net.Conn var cConn, sConn net.Conn
sConn, cConn = NewFConn() sConn, cConn = NewFConn()
cData := []byte("qwerty") cData := []byte("qwerty")
count := 10 count := 10
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
wc, err := cConn.Write(cData) wc, err := cConn.Write(cData)
if err != nil { if err != nil {
t.Error(err) t.Error(err)
} }
require.Equal(t, wc, len(cData)) require.Equal(t, wc, len(cData))
sData := make([]byte, len(cData)) sData := make([]byte, len(cData))
rc, err := sConn.Read(sData) rc, err := sConn.Read(sData)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, rc, len(cData)) require.Equal(t, rc, len(cData))
require.Equal(t, cData, sData) require.Equal(t, cData, sData)
} }
} }
func TestFConn1(t *testing.T) { func TestFConn1(t *testing.T) {
var cConn, sConn net.Conn var cConn, sConn net.Conn
cConn, sConn = NewFConn() cConn, sConn = NewFConn()
cData := []byte("qwerty") cData := []byte("qwerty")
count := 10 count := 10
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
wc, err := cConn.Write(cData) wc, err := cConn.Write(cData)
if err != nil { if err != nil {
t.Error(err) t.Error(err)
} }
require.Equal(t, wc, len(cData)) require.Equal(t, wc, len(cData))
sData := make([]byte, len(cData)) sData := make([]byte, len(cData))
rc, err := sConn.Read(sData) rc, err := sConn.Read(sData)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, rc, len(cData)) require.Equal(t, rc, len(cData))
require.Equal(t, cData, sData) require.Equal(t, cData, sData)
} }
} }

View File

@@ -7,62 +7,62 @@
package dsrpc package dsrpc
import ( import (
"bytes" "bytes"
"io" "io"
"net" "net"
"time" "time"
) )
type FConn struct { type FConn struct {
reader io.Reader reader io.Reader
writer io.Writer writer io.Writer
} }
func NewFConn() (*FConn, *FConn){ func NewFConn() (*FConn, *FConn) {
c2sBuffer := bytes.NewBuffer(make([]byte, 0)) c2sBuffer := bytes.NewBuffer(make([]byte, 0))
s2cBuffer := bytes.NewBuffer(make([]byte, 0)) s2cBuffer := bytes.NewBuffer(make([]byte, 0))
var client FConn var client FConn
client.writer = c2sBuffer client.writer = c2sBuffer
client.reader = s2cBuffer client.reader = s2cBuffer
var server FConn var server FConn
server.writer = s2cBuffer server.writer = s2cBuffer
server.reader = c2sBuffer server.reader = c2sBuffer
return &client, &server return &client, &server
} }
func (conn FConn) SetDeadline(t time.Time) error { func (conn FConn) SetDeadline(t time.Time) error {
var err error var err error
return err return err
} }
func (conn FConn) SetReadDeadline(t time.Time) error { func (conn FConn) SetReadDeadline(t time.Time) error {
var err error var err error
return err return err
} }
func (conn FConn) SetWriteDeadline(t time.Time) error { func (conn FConn) SetWriteDeadline(t time.Time) error {
var err error var err error
return err return err
} }
func (conn FConn) LocalAddr() net.Addr { func (conn FConn) LocalAddr() net.Addr {
return NewFAddr() return NewFAddr()
} }
func (conn FConn) RemoteAddr() net.Addr { func (conn FConn) RemoteAddr() net.Addr {
return NewFAddr() return NewFAddr()
} }
func (conn FConn) Write(data []byte) (int, error) { func (conn FConn) Write(data []byte) (int, error) {
return conn.writer.Write(data) return conn.writer.Write(data)
} }
func (conn FConn) Read(data []byte) (int, error) { func (conn FConn) Read(data []byte) (int, error) {
return conn.reader.Read(data) return conn.reader.Read(data)
} }
func (conn FConn) Close() error { func (conn FConn) Close() error {
var err error var err error
return err return err
} }

8
go.mod
View File

@@ -1,11 +1,11 @@
module github.com/kindsoldier/dsrpc module github.com/kindsoldier/dsrpc
go 1.17 go 1.19
require github.com/stretchr/testify v1.7.1 require github.com/stretchr/testify v1.8.2
require ( require (
github.com/davecgh/go-spew v1.1.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect gopkg.in/yaml.v3 v3.0.1 // indirect
) )

12
go.sum
View File

@@ -1,11 +1,17 @@
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

125
header.go
View File

@@ -7,92 +7,95 @@
package dsrpc package dsrpc
import ( import (
"errors" "bytes"
"encoding/binary" "encoding/binary"
"encoding/json" "encoding/json"
"bytes" "errors"
) )
const headerSize int64 = 16 * 2 const (
const sizeOfInt64 int = 8 headerSize int64 = 16 * 2
const magicCodeA int64 = 0xEE00ABBA sizeOfInt64 int = 8
const magicCodeB int64 = 0xEE44ABBA magicCodeA int64 = 0xEE00ABBA
magicCodeB int64 = 0xEE44ABBA
)
type Header struct { type Header struct {
magicCodeA int64 `json:"magicCodeA"` magicCodeA int64 `json:"magicCodeA"`
rpcSize int64 `json:"rpcSize"` rpcSize int64 `json:"rpcSize"`
binSize int64 `json:"binSize"` binSize int64 `json:"binSize"`
magicCodeB int64 `json:"magicCodeB"` magicCodeB int64 `json:"magicCodeB"`
} }
func NewEmptyHeader() *Header {
func NewHeader() *Header { return &Header{
return &Header{ magicCodeA: magicCodeA,
magicCodeA: magicCodeA, magicCodeB: magicCodeB,
magicCodeB: magicCodeB, }
}
} }
func (this *Header) JSON() []byte { func (hdr *Header) ToJson() []byte {
jBytes, _ := json.Marshal(this) jBytes, _ := json.Marshal(hdr)
return jBytes return jBytes
} }
func (hdr *Header) Pack() ([]byte, error) {
var err error
headerBytes := make([]byte, 0, headerSize)
headerBuffer := bytes.NewBuffer(headerBytes)
func (this *Header) Pack() ([]byte, error) { magicCodeABytes := EncoderI64(hdr.magicCodeA)
var err error headerBuffer.Write(magicCodeABytes)
headerBytes := make([]byte, 0, headerSize)
headerBuffer := bytes.NewBuffer(headerBytes)
magicCodeABytes := encoderI64(this.magicCodeA) rpcSizeBytes := EncoderI64(hdr.rpcSize)
headerBuffer.Write(magicCodeABytes) headerBuffer.Write(rpcSizeBytes)
rpcSizeBytes := encoderI64(this.rpcSize) binSizeBytes := EncoderI64(hdr.binSize)
headerBuffer.Write(rpcSizeBytes) headerBuffer.Write(binSizeBytes)
binSizeBytes := encoderI64(this.binSize) magicCodeBBytes := EncoderI64(hdr.magicCodeB)
headerBuffer.Write(binSizeBytes) headerBuffer.Write(magicCodeBBytes)
magicCodeBBytes := encoderI64(this.magicCodeB) return headerBuffer.Bytes(), err
headerBuffer.Write(magicCodeBBytes)
return headerBuffer.Bytes(), Err(err)
} }
func UnpackHeader(headerBytes []byte) (*Header, error) { func UnpackHeader(headerBytes []byte) (*Header, error) {
var err error var err error
header := NewHeader()
headerReader := bytes.NewReader(headerBytes)
magicCodeABytes := make([]byte, sizeOfInt64) headerReader := bytes.NewReader(headerBytes)
headerReader.Read(magicCodeABytes)
header.magicCodeA = decoderI64(magicCodeABytes)
rpcSizeBytes := make([]byte, sizeOfInt64) magicCodeABytes := make([]byte, sizeOfInt64)
headerReader.Read(rpcSizeBytes) headerReader.Read(magicCodeABytes)
header.rpcSize = decoderI64(rpcSizeBytes)
binSizeBytes := make([]byte, sizeOfInt64) rpcSizeBytes := make([]byte, sizeOfInt64)
headerReader.Read(binSizeBytes) headerReader.Read(rpcSizeBytes)
header.binSize = decoderI64(binSizeBytes)
magicCodeBBytes := make([]byte, sizeOfInt64) binSizeBytes := make([]byte, sizeOfInt64)
headerReader.Read(magicCodeBBytes) headerReader.Read(binSizeBytes)
header.magicCodeB = decoderI64(magicCodeBBytes)
if header.magicCodeA != magicCodeA || header.magicCodeB != magicCodeB { magicCodeBBytes := make([]byte, sizeOfInt64)
err = errors.New("wrong protocol magic code") headerReader.Read(magicCodeBBytes)
return header, Err(err)
} header := &Header{
return header, Err(err) magicCodeA: DecoderI64(magicCodeABytes),
rpcSize: DecoderI64(rpcSizeBytes),
binSize: DecoderI64(binSizeBytes),
magicCodeB: DecoderI64(magicCodeBBytes),
}
if header.magicCodeA != magicCodeA || header.magicCodeB != magicCodeB {
err = errors.New("Wrong protocol magic code")
return header, err
}
return header, err
} }
func encoderI64(i int64) []byte { func EncoderI64(i int64) []byte {
buffer := make([]byte, sizeOfInt64) buffer := make([]byte, sizeOfInt64)
binary.BigEndian.PutUint64(buffer, uint64(i)) binary.BigEndian.PutUint64(buffer, uint64(i))
return buffer return buffer
} }
func decoderI64(b []byte) int64 { func DecoderI64(b []byte) int64 {
return int64(binary.BigEndian.Uint64(b)) return int64(binary.BigEndian.Uint64(b))
} }

View File

@@ -7,39 +7,36 @@
package dsrpc package dsrpc
import ( import (
"fmt" "fmt"
"io" "io"
"os" "os"
"time"
) )
var messageWriter io.Writer = os.Stdout var (
var accessWriter io.Writer = os.Stdout messageWriter io.Writer = os.Stdout
accessWriter io.Writer = os.Stdout
)
func logDebug(messages ...any) { func logDebug(messages ...any) {
stamp := time.Now().Format(time.RFC3339) fmt.Fprintln(messageWriter, "debug:", messages)
fmt.Fprintln(messageWriter, stamp, "dsrpc debug", messages)
} }
func logInfo(messages ...any) { func logInfo(messages ...any) {
stamp := time.Now().Format(time.RFC3339) fmt.Fprintln(messageWriter, "info:", messages)
fmt.Fprintln(messageWriter, stamp, "dsrpc info", messages)
} }
func logError(messages ...any) { func logError(messages ...any) {
stamp := time.Now().Format(time.RFC3339) fmt.Fprintln(messageWriter, "error:", messages)
fmt.Fprintln(messageWriter, stamp, "dsrpc error", messages)
} }
func logAccess(messages ...any) { func logAccess(messages ...any) {
stamp := time.Now().Format(time.RFC3339) fmt.Fprintln(accessWriter, "access:", messages)
fmt.Fprintln(accessWriter, stamp, "dsrpc access", messages)
} }
func SetAccessWriter(writer io.Writer) { func SetAccessWriter(writer io.Writer) {
accessWriter = writer accessWriter = writer
} }
func SetMessageWriter(writer io.Writer) { func SetMessageWriter(writer io.Writer) {
messageWriter = writer messageWriter = writer
} }

View File

@@ -1,31 +1,29 @@
/* /*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org> * Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/ */
package dsrpc package dsrpc
import ( import (
"time" "time"
) )
func LogRequest(context *Context) error { func LogRequest(content *Content) error {
var err error var err error
logDebug("request:", string(context.reqRPC.JSON())) logDebug("request:", string(content.reqBlock.ToJson()))
return Err(err) return err
} }
func LogResponse(context *Context) error { func LogResponse(content *Content) error {
var err error var err error
logDebug("response:", string(context.resRPC.JSON())) logDebug("response:", string(content.resBlock.ToJson()))
return Err(err) return err
} }
func LogAccess(context *Context) error { func LogAccess(content *Content) error {
var err error var err error
execTime := time.Now().Sub(context.start) execTime := time.Now().Sub(content.start)
logAccess(context.remoteHost, context.reqRPC.Method, execTime) login := string(content.AuthIdent())
return Err(err) logAccess(content.remoteHost, login, content.reqBlock.Method, execTime)
return err
} }

View File

@@ -1,25 +1,18 @@
/* /*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org> * Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/ */
package dsrpc package dsrpc
import (
"encoding/json"
)
type Packet struct { type Packet struct {
header []byte header []byte
rcpPayload []byte rcpPayload []byte
} }
func NewPacket() *Packet { func NewEmptyPacket() *Packet {
return &Packet{} packet := &Packet{
} header: make([]byte, 0),
rcpPayload: make([]byte, 0),
func (this *Packet) JSON() []byte { }
jBytes, _ := json.Marshal(this) return packet
return jBytes
} }

View File

@@ -7,27 +7,36 @@
package dsrpc package dsrpc
import ( import (
"encoding/json" "encoding/json"
encoder "encoding/json"
) )
type EmptyParams struct{}
func NewEmptyParams() *EmptyParams {
return &EmptyParams{}
}
type Request struct { type Request struct {
Method string `json:"method" msgpack:"method"` Method string `json:"method" msgpack:"method"`
Params any `json:"params,omitempty" msgpack:"params,omitempty"` Params any `json:"params,omitempty" msgpack:"params"`
Auth *Auth `json:"auth,omitempty" msgpack:"auth,omitempty"` Auth *Auth `json:"auth,omitempty" msgpack:"auth"`
} }
func NewRequest() *Request { func NewEmptyRequest() *Request {
req := &Request{} req := &Request{}
req.Auth = &Auth{} req.Auth = &Auth{}
return req req.Params = NewEmptyParams()
return req
} }
func (this *Request) Pack() ([]byte, error) { func (req *Request) Pack() ([]byte, error) {
rBytes, err := json.Marshal(this) rBytes, err := encoder.Marshal(req)
return rBytes, Err(err) return rBytes, err
} }
func (this *Request) JSON() []byte { func (req *Request) ToJson() []byte {
jBytes, _ := json.Marshal(this) jBytes, _ := json.Marshal(req)
return jBytes return jBytes
} }

View File

@@ -1,31 +1,38 @@
/* /*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org> * Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/ */
package dsrpc package dsrpc
import ( import (
"encoding/json" "encoding/json"
encoder "encoding/json"
) )
type EmptyResult struct{}
func NewEmptyResult() *EmptyResult {
return &EmptyResult{}
}
type Response struct { type Response struct {
Error string `json:"error,omitempty" msgpack:"error,omitempty"` Error string `json:"error" msgpack:"error"`
Result any `json:"result,omitemty" msgpack:"result,omitemty"` Result any `json:"result" msgpack:"result"`
} }
func NewResponse() *Response { func NewEmptyResponse() *Response {
return &Response{} return &Response{
Result: NewEmptyResult(),
}
} }
func (this *Response) JSON() []byte { func (resp *Response) ToJson() []byte {
jBytes, _ := json.Marshal(this) jBytes, _ := json.Marshal(resp)
return jBytes return jBytes
} }
func (this *Response) Pack() ([]byte, error) { func (resp *Response) Pack() ([]byte, error) {
rBytes, err := json.Marshal(this) rBytes, err := encoder.Marshal(resp)
return rBytes, Err(err) return rBytes, err
} }

512
server.go
View File

@@ -1,268 +1,348 @@
/* /*
*
* Copyright 2022 Oleg Borodin <borodin@unix7.org> * Copyright 2022 Oleg Borodin <borodin@unix7.org>
*
*/ */
package dsrpc package dsrpc
import ( import (
"context" "context"
"encoding/json" "crypto/tls"
"errors" "errors"
"io" "fmt"
"net" "io"
"sync" "net"
"sync"
"time"
encoder "encoding/json"
) )
type HandlerFunc = func(*Context) error type HandlerFunc = func(*Content) error
type Service struct { type Service struct {
handlers map[string]HandlerFunc handlers map[string]HandlerFunc
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
wg *sync.WaitGroup wg *sync.WaitGroup
preMw []HandlerFunc preMw []HandlerFunc
postMw []HandlerFunc postMw []HandlerFunc
keepalive bool
kaTime time.Duration
kaMtx sync.Mutex
listener net.Listener
tcpListener *net.TCPListener
} }
func NewService() *Service { func NewService() *Service {
rdrpc := &Service{} rdrpc := &Service{}
rdrpc.handlers = make(map[string]HandlerFunc) rdrpc.handlers = make(map[string]HandlerFunc)
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
rdrpc.ctx = ctx rdrpc.ctx = ctx
rdrpc.cancel = cancel rdrpc.cancel = cancel
var wg sync.WaitGroup var wg sync.WaitGroup
rdrpc.wg = &wg rdrpc.wg = &wg
rdrpc.preMw = make([]HandlerFunc, 0) rdrpc.preMw = make([]HandlerFunc, 0)
rdrpc.postMw = make([]HandlerFunc, 0) rdrpc.postMw = make([]HandlerFunc, 0)
return rdrpc return rdrpc
} }
func (this *Service) PreMiddleware(mw HandlerFunc) { func (svc *Service) PreMiddleware(mw HandlerFunc) {
this.preMw = append(this.preMw, mw) svc.preMw = append(svc.preMw, mw)
} }
func (this *Service) PostMiddleware(mw HandlerFunc) { func (svc *Service) PostMiddleware(mw HandlerFunc) {
this.postMw = append(this.postMw, mw) svc.postMw = append(svc.postMw, mw)
} }
func (svc *Service) Handle(method string, handler HandlerFunc) {
func (this *Service) Handler(method string, handler HandlerFunc) { svc.handlers[method] = handler
this.handlers[method] = handler
} }
func (this *Service) Listen(address string) error { func (svc *Service) SetKeepAlive(flag bool) {
var err error svc.kaMtx.Lock()
logInfo("server listen:", address) defer svc.kaMtx.Unlock()
svc.keepalive = true
listener, err := net.Listen("tcp", address)
if err != nil {
return err
}
this.wg.Add(1)
for {
select {
case <- this.ctx.Done():
this.wg.Done()
return err
default:
}
conn, err := listener.Accept()
if err != nil {
logError("conn accept err:", err)
}
go this.handleConn(conn)
}
} }
func notFound(context *Context) error { func (svc *Service) SetKeepAlivePeriod(interval time.Duration) {
execErr := errors.New("method not found") svc.kaMtx.Lock()
err := context.SendError(execErr) defer svc.kaMtx.Unlock()
return err svc.kaTime = interval
} }
func (this *Service) Stop() error { func (svc *Service) Listen(address string) error {
var err error var err error
this.cancel() logInfo("server listen:", address)
this.wg.Wait()
return err addr, err := net.ResolveTCPAddr("tcp", address)
if err != nil {
err = fmt.Errorf("unable to resolve adddress: %s", err)
return err
}
svc.tcpListener, err = net.ListenTCP("tcp", addr)
if err != nil {
err = fmt.Errorf("unable to start listener: %s", err)
return err
}
for {
conn, err := svc.tcpListener.AcceptTCP()
if err != nil {
logError("conn accept err:", err)
}
select {
case <-svc.ctx.Done():
return err
default:
}
svc.wg.Add(1)
go svc.handleTCPConn(conn, svc.wg)
}
return err
} }
func (this *Service) handleConn(conn net.Conn) { func (svc *Service) ListenTLS(address string, tlsConfig *tls.Config) error {
var err error var err error
logInfo("server listen:", address)
context := CreateContext(conn) svc.listener, err = tls.Listen("tcp", address, tlsConfig)
if err != nil {
err = fmt.Errorf("unable to start listener: %s", err)
return err
}
remoteAddr := conn.RemoteAddr().String() for {
remoteHost, _, _ := net.SplitHostPort(remoteAddr) conn, err := svc.listener.Accept()
context.remoteHost = remoteHost if err != nil {
logError("conn accept err:", err)
context.binReader = conn }
context.binWriter = io.Discard select {
case <-svc.ctx.Done():
exitFunc := func() { logInfo("accept loop done")
conn.Close() return err
if err != nil { default:
logError("conn handler err:", err) }
} svc.wg.Add(1)
} go svc.handleConn(conn, svc.wg)
defer exitFunc() }
return err
recovFunc := func () {
panicMsg := recover()
if panicMsg != nil {
logError("handler panic message:", panicMsg)
}
}
defer recovFunc()
err = context.ReadRequest()
if err != nil {
err = Err(err)
return
}
err = context.BindMethod()
if err != nil {
err = Err(err)
return
}
for _, mw := range this.preMw {
err = mw(context)
if err != nil {
err = Err(err)
return
}
}
err = this.Route(context)
if err != nil {
err = Err(err)
return
}
for _, mw := range this.postMw {
err = mw(context)
if err != nil {
err = Err(err)
return
}
}
return
} }
func (this *Service) Route(context *Context) error { func notFound(content *Content) error {
handler, ok := this.handlers[context.reqRPC.Method] execErr := errors.New("method not found")
if ok { err := content.SendError(execErr)
return Err(handler(context)) return err
}
return Err(notFound(context))
} }
func (context *Context) ReadRequest() error { func (svc *Service) Stop() error {
var err error var err error
// Disable new connection
context.reqPacket.header, err = ReadBytes(context.sockReader, headerSize) logInfo("cancel rpc accept loop")
if err != nil { if svc.listener != nil {
return Err(err) svc.listener.Close()
} }
context.reqHeader, err = UnpackHeader(context.reqPacket.header) if svc.tcpListener != nil {
if err != nil { svc.tcpListener.Close()
return Err(err) }
} svc.cancel()
// Wait handlers
rpcSize := context.reqHeader.rpcSize logInfo("wait rpc handlers")
context.reqPacket.rcpPayload, err = ReadBytes(context.sockReader, rpcSize) svc.wg.Wait()
if err != nil { return err
return Err(err)
}
return Err(err)
} }
func (context *Context) BinWriter() io.Writer { func (svc *Service) handleTCPConn(conn *net.TCPConn, wg *sync.WaitGroup) {
return context.sockWriter var err error
if svc.keepalive {
err = conn.SetKeepAlive(true)
if err != nil {
err = fmt.Errorf("unable to set keepalive: %s", err)
return
}
if svc.kaTime > 0 {
err = conn.SetKeepAlivePeriod(svc.kaTime)
if err != nil {
err = fmt.Errorf("unable to set keepalive period: %s", err)
return
}
}
}
svc.handleConn(conn, wg)
} }
func (context *Context) BinReader() io.Reader { func (svc *Service) handleConn(conn net.Conn, wg *sync.WaitGroup) {
return context.sockReader var err error
content := CreateContent(conn)
remoteAddr := conn.RemoteAddr().String()
remoteHost, _, _ := net.SplitHostPort(remoteAddr)
content.remoteHost = remoteHost
content.binReader = conn
content.binWriter = io.Discard
exitFunc := func() {
conn.Close()
wg.Done()
if err != nil {
logError("conn handler err:", err)
}
}
defer exitFunc()
recovFunc := func() {
panicMsg := recover()
if panicMsg != nil {
logError("handler panic message:", panicMsg)
}
}
defer recovFunc()
err = content.ReadRequest()
if err != nil {
err = err
return
}
err = content.BindMethod()
if err != nil {
err = err
return
}
for _, mw := range svc.preMw {
err = mw(content)
if err != nil {
err = err
return
}
}
err = svc.Route(content)
if err != nil {
err = err
return
}
for _, mw := range svc.postMw {
err = mw(content)
if err != nil {
err = err
return
}
}
return
} }
func (context *Context) BinSize() int64 { func (svc *Service) Route(content *Content) error {
return context.reqHeader.binSize handler, ok := svc.handlers[content.reqBlock.Method]
if ok {
return handler(content)
}
return notFound(content)
} }
func (context *Context) ReadBin(writer io.Writer) error { func (content *Content) ReadRequest() error {
var err error var err error
_, err = CopyBytes(context.sockReader, writer, context.reqHeader.binSize)
return Err(err) content.reqPacket.header, err = ReadBytes(content.sockReader, headerSize)
if err != nil {
return err
}
content.reqHeader, err = UnpackHeader(content.reqPacket.header)
if err != nil {
return err
}
rpcSize := content.reqHeader.rpcSize
content.reqPacket.rcpPayload, err = ReadBytes(content.sockReader, rpcSize)
if err != nil {
return err
}
return err
} }
func (content *Content) BinWriter() io.Writer {
func (context *Context) BindMethod() error { return content.sockWriter
var err error
err = json.Unmarshal(context.reqPacket.rcpPayload, context.reqRPC)
return Err(err)
} }
func (context *Context) BindParams(params any) error { func (content *Content) BinReader() io.Reader {
var err error return content.sockReader
context.reqRPC.Params = params
err = json.Unmarshal(context.reqPacket.rcpPayload, context.reqRPC)
if err != nil {
return Err(err)
}
return Err(err)
} }
func (context *Context) SendResult(result any, binSize int64) error { func (content *Content) BinSize() int64 {
var err error return content.reqHeader.binSize
context.resRPC.Result = result
context.resPacket.rcpPayload, err = context.resRPC.Pack()
if err != nil {
return Err(err)
}
context.resHeader.rpcSize = int64(len(context.resPacket.rcpPayload))
context.resHeader.binSize = binSize
context.resPacket.header, err = context.resHeader.Pack()
if err != nil {
return Err(err)
}
_, err = context.sockWriter.Write(context.resPacket.header)
if err != nil {
return Err(err)
}
_, err = context.sockWriter.Write(context.resPacket.rcpPayload)
if err != nil {
return Err(err)
}
return Err(err)
} }
func (content *Content) ReadBin(ctx context.Context, writer io.Writer) error {
func (context *Context) SendError(execErr error) error { var err error
var err error _, err = CopyBytes(ctx, content.sockReader, writer, content.reqHeader.binSize)
return err
context.resRPC.Error = execErr.Error() }
context.resRPC.Result = NewEmpty()
func (content *Content) BindMethod() error {
context.resPacket.rcpPayload, err = context.resRPC.Pack() var err error
if err != nil { err = encoder.Unmarshal(content.reqPacket.rcpPayload, content.reqBlock)
return Err(err) return err
} }
context.resHeader.rpcSize = int64(len(context.resPacket.rcpPayload))
context.resPacket.header, err = context.resHeader.Pack() func (content *Content) BindParams(params any) error {
if err != nil { var err error
return Err(err) content.reqBlock.Params = params
} err = encoder.Unmarshal(content.reqPacket.rcpPayload, content.reqBlock)
_, err = context.sockWriter.Write(context.resPacket.header) if err != nil {
if err != nil { return err
return Err(err) }
} return err
_, err = context.sockWriter.Write(context.resPacket.rcpPayload) }
if err != nil {
return Err(err) func (content *Content) SendResult(result any, binSize int64) error {
} var err error
return Err(err) content.resBlock.Result = result
content.resPacket.rcpPayload, err = content.resBlock.Pack()
if err != nil {
return err
}
content.resHeader.rpcSize = int64(len(content.resPacket.rcpPayload))
content.resHeader.binSize = binSize
content.resPacket.header, err = content.resHeader.Pack()
if err != nil {
return err
}
_, err = content.sockWriter.Write(content.resPacket.header)
if err != nil {
return err
}
_, err = content.sockWriter.Write(content.resPacket.rcpPayload)
if err != nil {
return err
}
return err
}
func (content *Content) SendError(execErr error) error {
var err error
content.resBlock.Error = execErr.Error()
content.resBlock.Result = NewEmptyResult()
content.resPacket.rcpPayload, err = content.resBlock.Pack()
if err != nil {
return err
}
content.resHeader.rpcSize = int64(len(content.resPacket.rcpPayload))
content.resPacket.header, err = content.resHeader.Pack()
if err != nil {
return err
}
_, err = content.sockWriter.Write(content.resPacket.header)
if err != nil {
return err
}
_, err = content.sockWriter.Write(content.resPacket.rcpPayload)
if err != nil {
return err
}
return err
} }

View File

@@ -5,53 +5,60 @@
package dsrpc package dsrpc
import ( import (
"errors" "context"
"io" "errors"
"fmt" "fmt"
"io"
) )
func ReadBytes(reader io.Reader, size int64) ([]byte, error) { func ReadBytes(reader io.Reader, size int64) ([]byte, error) {
buffer := make([]byte, size) buffer := make([]byte, size)
read, err := io.ReadFull(reader, buffer) read, err := io.ReadFull(reader, buffer)
return buffer[0:read], Err(err) return buffer[0:read], err
} }
func CopyBytes(reader io.Reader, writer io.Writer, dataSize int64) (int64, error) { func CopyBytes(ctx context.Context, reader io.Reader, writer io.Writer, dataSize int64) (int64, error) {
var err error var err error
var bSize int64 = 1024 * 4 var bSize int64 = 1024 * 16
var total int64 = 0 var total int64 = 0
var remains int64 = dataSize var remains int64 = dataSize
buffer := make([]byte, bSize) buffer := make([]byte, bSize)
for { for {
if reader == nil { select {
return total, errors.New("reader is nil") case <-ctx.Done():
} return total, errors.New("break by context")
if writer == nil { default:
return total, errors.New("writer is nil") }
}
if remains == 0 { if reader == nil {
return total, err return total, errors.New("reader is nil")
} }
if remains < bSize { if writer == nil {
bSize = remains return total, errors.New("writer is nil")
} }
received, err := reader.Read(buffer[0:bSize]) if remains == 0 {
if err != nil { return total, err
err = fmt.Errorf("read error: %v", err) }
return total, Err(err) if remains < bSize {
} bSize = remains
recorded, err := writer.Write(buffer[0:received]) }
if err != nil { received, err := reader.Read(buffer[0:bSize])
err = fmt.Errorf("write error: %v", err) if err != nil {
return total, Err(err) err = fmt.Errorf("read error: %v", err)
} return total, err
if recorded != received { }
err = errors.New("size mismatch") recorded, err := writer.Write(buffer[0:received])
return total, Err(err) if err != nil {
} err = fmt.Errorf("write error: %v", err)
total += int64(recorded) return total, err
remains -= int64(recorded) }
} if recorded != received {
return total, Err(err) err = errors.New("size mismatch")
return total, err
}
total += int64(recorded)
remains -= int64(recorded)
}
return total, err
} }

View File

@@ -4,161 +4,172 @@
package dsrpc package dsrpc
import ( import (
"io" "context"
"net" "io"
"net"
) )
func LocalExec(method string, param any, result any, auth *Auth, handler HandlerFunc) error { func LocalExec(method string, param, result any, auth *Auth, handler HandlerFunc) error {
var err error var err error
cliConn, srvConn := NewFConn() cliConn, srvConn := NewFConn()
context := CreateContext(cliConn) content := CreateContent(cliConn)
context.reqRPC.Method = method content.reqBlock.Method = method
context.reqRPC.Params = param
context.reqRPC.Auth = auth
context.resRPC.Result = result
if context.reqRPC.Params == nil { if param != nil {
context.reqRPC.Params = NewEmpty() content.reqBlock.Params = param
} }
err = context.CreateRequest() if auth != nil {
if err != nil { content.reqBlock.Auth = auth
return err }
} if result != nil {
err = context.WriteRequest() content.resBlock.Result = result
if err != nil { }
return err
}
err = LocalService(srvConn, handler)
if err != nil {
return err
}
err = context.ReadResponse()
if err != nil {
return err
}
err = context.BindResponse()
if err != nil {
return err
}
return err err = content.createRequest()
if err != nil {
return err
}
err = content.writeRequest()
if err != nil {
return err
}
err = LocalService(srvConn, handler)
if err != nil {
return err
}
err = content.readResponse()
if err != nil {
return err
}
err = content.bindResponse()
if err != nil {
return err
}
return err
} }
func LocalPut(method string, reader io.Reader, size int64, param, result any, auth *Auth, handler HandlerFunc) error { func LocalPut(ctx context.Context, method string, reader io.Reader, size int64, param, result any, auth *Auth, handler HandlerFunc) error {
var err error var err error
cliConn, srvConn := NewFConn() cliConn, srvConn := NewFConn()
context := CreateContext(cliConn) content := CreateContent(cliConn)
context.reqRPC.Method = method content.reqBlock.Method = method
context.reqRPC.Params = param
context.reqRPC.Auth = auth
context.resRPC.Result = result
context.binReader = reader if param != nil {
context.binWriter = cliConn content.reqBlock.Params = param
}
if auth != nil {
content.reqBlock.Auth = auth
}
if result != nil {
content.resBlock.Result = result
}
context.reqHeader.binSize = size content.binReader = reader
content.binWriter = cliConn
if context.reqRPC.Params == nil { content.reqHeader.binSize = size
context.reqRPC.Params = NewEmpty()
} err = content.createRequest()
err = context.CreateRequest() if err != nil {
if err != nil { return err
return err }
} err = content.writeRequest()
err = context.WriteRequest() if err != nil {
if err != nil { return err
return err }
} err = content.uploadBin(ctx)
err = context.UploadBin() if err != nil {
if err != nil { return err
return err }
} err = LocalService(srvConn, handler)
err = LocalService(srvConn, handler) if err != nil {
if err != nil { return err
return err }
} err = content.readResponse()
err = context.ReadResponse() if err != nil {
if err != nil { return err
return err }
} err = content.bindResponse()
err = context.BindResponse() if err != nil {
if err != nil { return err
return err }
} return err
return err
} }
func LocalGet(ctx context.Context, method string, writer io.Writer, param, result any, auth *Auth, handler HandlerFunc) error {
var err error
func LocalGet(method string, writer io.Writer, param, result any, auth *Auth, handler HandlerFunc) error { cliConn, srvConn := NewFConn()
var err error
cliConn, srvConn := NewFConn() content := CreateContent(cliConn)
content.reqBlock.Method = method
context := CreateContext(cliConn) if param != nil {
context.reqRPC.Method = method content.reqBlock.Params = param
context.reqRPC.Params = param }
context.reqRPC.Auth = auth if auth != nil {
context.resRPC.Result = result content.reqBlock.Auth = auth
}
if result != nil {
content.resBlock.Result = result
}
context.binReader = cliConn content.binReader = cliConn
context.binWriter = writer content.binWriter = writer
if context.reqRPC.Params == nil { err = content.createRequest()
context.reqRPC.Params = NewEmpty() if err != nil {
} return err
err = context.CreateRequest() }
if err != nil { err = content.writeRequest()
return err if err != nil {
} return err
err = context.WriteRequest() }
if err != nil {
return err
}
err = LocalService(srvConn, handler) err = LocalService(srvConn, handler)
if err != nil { if err != nil {
return err return err
} }
err = context.ReadResponse() err = content.readResponse()
if err != nil { if err != nil {
return err return err
} }
err = context.DownloadBin() err = content.downloadBin(ctx)
if err != nil { if err != nil {
return err return err
} }
err = context.BindResponse() err = content.bindResponse()
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
func LocalService(conn net.Conn, handler HandlerFunc) error { func LocalService(conn net.Conn, handler HandlerFunc) error {
var err error var err error
context := CreateContext(conn) content := CreateContent(conn)
remoteAddr := conn.RemoteAddr().String() remoteAddr := conn.RemoteAddr().String()
remoteHost, _, _ := net.SplitHostPort(remoteAddr) remoteHost, _, _ := net.SplitHostPort(remoteAddr)
context.remoteHost = remoteHost content.remoteHost = remoteHost
context.binReader = conn content.binReader = conn
context.binWriter = io.Discard content.binWriter = io.Discard
err = context.ReadRequest() err = content.ReadRequest()
if err != nil { if err != nil {
return err return err
} }
err = context.BindMethod() err = content.BindMethod()
if err != nil { if err != nil {
return err return err
} }
return handler(context) return handler(content)
} }

View File

@@ -5,60 +5,60 @@
package dsrpc package dsrpc
import ( import (
"encoding/json" "bytes"
"bytes" "crypto/sha256"
"math/rand" "encoding/json"
"time" "math/rand"
"crypto/sha256" "time"
) )
func init() { func init() {
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
} }
type Auth struct { type Auth struct {
Ident []byte `json:"ident,omitempty"` Ident []byte `msgpack:"ident" json:"ident"`
Salt []byte `json:"salt,omitempty"` Salt []byte `msgpack:"salt" json:"salt"`
Hash []byte `json:"hash,omitempty"` Hash []byte `msgpack:"hash" json:"hash"`
} }
func NewAuth() *Auth { func NewAuth() *Auth {
return &Auth{} return &Auth{}
} }
func (this *Auth) JSON() []byte { func (this *Auth) Json() []byte {
jBytes, _ := json.Marshal(this) jBytes, _ := json.Marshal(this)
return jBytes return jBytes
} }
func CreateAuth(ident, pass []byte) *Auth { func CreateAuth(ident, pass []byte) *Auth {
salt := CreateSalt() salt := CreateSalt()
hash := CreateHash(ident, pass, salt) hash := CreateHash(ident, pass, salt)
auth := &Auth{} auth := &Auth{}
auth.Ident = ident auth.Ident = ident
auth.Salt = salt auth.Salt = salt
auth.Hash = hash auth.Hash = hash
return auth return auth
} }
func CreateSalt() []byte { func CreateSalt() []byte {
const saltSize = 16 const saltSize = 16
randBytes := make([]byte, saltSize) randBytes := make([]byte, saltSize)
rand.Read(randBytes) rand.Read(randBytes)
return randBytes return randBytes
} }
func CreateHash(ident, pass, salt []byte) []byte { func CreateHash(ident, pass, salt []byte) []byte {
vec := make([]byte, 0, len(ident) + len(salt) + len(pass)) vec := make([]byte, 0, len(ident)+len(salt)+len(pass))
vec = append(vec, ident...) vec = append(vec, ident...)
vec = append(vec, salt...) vec = append(vec, salt...)
vec = append(vec, pass...) vec = append(vec, pass...)
hasher := sha256.New() hasher := sha256.New()
hash := hasher.Sum(vec) hash := hasher.Sum(vec)
return hash return hash
} }
func CheckHash(ident, pass, reqSalt, reqHash []byte) bool { func CheckHash(ident, pass, reqSalt, reqHash []byte) bool {
localHash := CreateHash(ident, pass, reqSalt) localHash := CreateHash(ident, pass, reqSalt)
return bytes.Equal(reqHash, localHash) return bytes.Equal(reqHash, localHash)
} }