ziggi/dsrpc
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: ziggi:v1.1.2
Branches Tags
ziggi:master
ziggi:v1.3.4 ziggi:v1.3.3 ziggi:v1.3.2 ziggi:v1.3.1 ziggi:v1.3.0 ziggi:v1.2.2 ziggi:v1.2.1 ziggi:v1.2.0 ziggi:v1.1.3 ziggi:v1.1.4 ziggi:v1.1.2 ziggi:v1.1.1 ziggi:v1.1.0 ziggi:v1.0.6 ziggi:v1.0.1 ziggi:v1.0.2 ziggi:v1.0.3 ziggi:v1.0.4 ziggi:v1.0.5
...
compare: ziggi:master
Branches Tags
ziggi:master
ziggi:v1.3.4 ziggi:v1.3.3 ziggi:v1.3.2 ziggi:v1.3.1 ziggi:v1.3.0 ziggi:v1.2.2 ziggi:v1.2.1 ziggi:v1.2.0 ziggi:v1.1.3 ziggi:v1.1.4 ziggi:v1.1.2 ziggi:v1.1.1 ziggi:v1.1.0 ziggi:v1.0.6 ziggi:v1.0.1 ziggi:v1.0.2 ziggi:v1.0.3 ziggi:v1.0.4 ziggi:v1.0.5

12 Commits
v1.1.2 ... master

Author SHA1 Message Date
Oleg Borodin
498abc18b4 delete timestamp from logger 2023-05-18 13:44:39 +02:00
Oleg Borodin
5ea1af7fb1 disable default log timestamp 2023-05-18 13:34:54 +02:00
Oleg Borodin
bc578cf25c fixed mistakes 2023-05-18 12:56:37 +02:00
Oleg Borodin
cb6d26a7cc added listener close() 2023-05-18 12:53:37 +02:00
Oleg Borodin
f814bb7d24 go fmt 2023-05-18 11:12:35 +02:00
Oleg Borodin
aed740c933 added TLS transport 2023-05-18 11:12:16 +02:00
Oleg Borodin
4716c25cf6 rename Handler to Handle 2023-05-17 22:35:35 +02:00
Oleg Borodin
9bc3aef167 updated samples 2023-05-17 13:06:13 +02:00
Oleg Borodin
0e5321d51a encoder changed to json 2023-05-17 13:03:52 +02:00
Oleg Borodin
6f9835f399 context added 2023-05-17 12:59:01 +02:00
Oleg Borodin
ef8db1d198 context added 2023-05-17 12:52:41 +02:00
Oleg Borodin
c20faaecb3 Little refactoring 2023-04-01 01:03:38 +02:00
17 changed files with 325 additions and 240 deletions
Expand all files Collapse all files

8
README.md
View File

@@ -44,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)
@@ -220,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)

112
client.go
View File

@@ -5,16 +5,18 @@
package dsrpc package dsrpc
import ( import (
"context"
"crypto/tls"
"errors" "errors"
"fmt" "fmt"
"io" "io"
"net" "net"
"sync" "sync"
encoder "github.com/vmihailenco/msgpack/v5" encoder "encoding/json"
) )
func Put(address string, method string, reader io.Reader, binSize int64, param, result any, auth *Auth) error { func Put(ctx context.Context, address string, method string, reader io.Reader, binSize int64, param, result any, auth *Auth) error {
var err error var err error
addr, err := net.ResolveTCPAddr("tcp", address) addr, err := net.ResolveTCPAddr("tcp", address)
@@ -28,10 +30,22 @@ func Put(address string, method string, reader io.Reader, binSize int64, param,
} }
defer conn.Close() defer conn.Close()
return ConnPut(conn, method, reader, binSize, param, result, auth) return ConnPut(ctx, conn, method, reader, binSize, param, result, auth)
} }
func ConnPut(conn net.Conn, method string, reader io.Reader, binSize int64, param, result any, auth *Auth) error { 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
conn, err := tls.Dial("tcp", address, tlsConfig)
if err != nil {
return err
}
defer conn.Close()
return ConnPut(ctx, conn, method, reader, binSize, param, result, auth)
}
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) content := CreateContent(conn)
@@ -51,11 +65,11 @@ func ConnPut(conn net.Conn, method string, reader io.Reader, binSize int64, para
content.reqHeader.binSize = binSize content.reqHeader.binSize = binSize
err = content.CreateRequest() err = content.createRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.WriteRequest() err = content.writeRequest()
if err != nil { if err != nil {
return err return err
} }
@@ -64,24 +78,24 @@ func ConnPut(conn net.Conn, method string, reader io.Reader, binSize int64, para
errChan := make(chan error, 1) errChan := make(chan error, 1)
wg.Add(1) wg.Add(1)
go content.ReadResponseAsync(&wg, errChan) go content.readResponseAsync(&wg, errChan)
wg.Add(1) wg.Add(1)
go content.UploadBinAsync(&wg) go content.uploadBinAsync(ctx, &wg)
wg.Wait() wg.Wait()
err = <-errChan err = <-errChan
if err != nil { if err != nil {
return err return err
} }
err = content.BindResponse() err = content.bindResponse()
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
func Get(address string, 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
addr, err := net.ResolveTCPAddr("tcp", address) addr, err := net.ResolveTCPAddr("tcp", address)
@@ -95,10 +109,22 @@ func Get(address string, method string, writer io.Writer, param, result any, aut
} }
defer conn.Close() defer conn.Close()
return ConnGet(conn, method, writer, param, result, auth) return ConnGet(ctx, conn, method, writer, param, result, auth)
} }
func ConnGet(conn net.Conn, method string, writer io.Writer, param, 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
conn, err := tls.Dial("tcp", address, tlsConfig)
if err != nil {
return 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 var err error
content := CreateContent(conn) content := CreateContent(conn)
@@ -116,30 +142,30 @@ func ConnGet(conn net.Conn, method string, writer io.Writer, param, result any,
content.binReader = conn content.binReader = conn
content.binWriter = writer content.binWriter = writer
err = content.CreateRequest() err = content.createRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.WriteRequest() err = content.writeRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.ReadResponse() err = content.readResponse()
if err != nil { if err != nil {
return err return err
} }
err = content.DownloadBin() err = content.downloadBin(ctx)
if err != nil { if err != nil {
return err return err
} }
err = content.BindResponse() err = content.bindResponse()
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
func Exec(address, method string, param any, result any, auth *Auth) error { func Exec(ctx context.Context, address, method string, param any, result any, auth *Auth) error {
var err error var err error
addr, err := net.ResolveTCPAddr("tcp", address) addr, err := net.ResolveTCPAddr("tcp", address)
@@ -153,14 +179,30 @@ func Exec(address, method string, param any, result any, auth *Auth) error {
} }
defer conn.Close() defer conn.Close()
err = ConnExec(conn, method, param, result, auth) err = ConnExec(ctx, conn, method, param, result, auth)
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
func ConnExec(conn net.Conn, method string, param any, result any, auth *Auth) error { func ExecTLS(ctx context.Context, tlsConfig *tls.Config, address, method string, param any, result any, auth *Auth) error {
var err error
conn, err := tls.Dial("tcp", address, tlsConfig)
if err != nil {
return err
}
defer conn.Close()
err = ConnExec(ctx, conn, method, param, result, auth)
if err != nil {
return err
}
return err
}
func ConnExec(ctx context.Context, conn net.Conn, method string, param any, result any, auth *Auth) error {
var err error var err error
content := CreateContent(conn) content := CreateContent(conn)
@@ -176,26 +218,26 @@ func ConnExec(conn net.Conn, method string, param any, result any, auth *Auth) e
content.reqBlock.Auth = auth content.reqBlock.Auth = auth
} }
err = content.CreateRequest() err = content.createRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.WriteRequest() err = content.writeRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.ReadResponse() err = content.readResponse()
if err != nil { if err != nil {
return err return err
} }
err = content.BindResponse() err = content.bindResponse()
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
func (content *Content) CreateRequest() error { func (content *Content) createRequest() error {
var err error var err error
content.reqPacket.rcpPayload, err = content.reqBlock.Pack() content.reqPacket.rcpPayload, err = content.reqBlock.Pack()
@@ -212,7 +254,7 @@ func (content *Content) CreateRequest() error {
return err return err
} }
func (content *Content) WriteRequest() error { func (content *Content) writeRequest() error {
var err error var err error
_, err = content.sockWriter.Write(content.reqPacket.header) _, err = content.sockWriter.Write(content.reqPacket.header)
if err != nil { if err != nil {
@@ -225,13 +267,13 @@ func (content *Content) WriteRequest() error {
return err return err
} }
func (content *Content) UploadBin() error { func (content *Content) uploadBin(ctx context.Context) error {
var err error var err error
_, err = CopyBytes(content.binReader, content.binWriter, content.reqHeader.binSize) _, err = CopyBytes(ctx, content.binReader, content.binWriter, content.reqHeader.binSize)
return err return err
} }
func (content *Content) ReadResponse() error { func (content *Content) readResponse() error {
var err error var err error
content.resPacket.header, err = ReadBytes(content.sockReader, headerSize) content.resPacket.header, err = ReadBytes(content.sockReader, headerSize)
@@ -250,16 +292,16 @@ func (content *Content) ReadResponse() error {
return err return err
} }
func (content *Content) UploadBinAsync(wg *sync.WaitGroup) { func (content *Content) uploadBinAsync(ctx context.Context, wg *sync.WaitGroup) {
exitFunc := func() { exitFunc := func() {
wg.Done() wg.Done()
} }
defer exitFunc() defer exitFunc()
_, _ = CopyBytes(content.binReader, content.binWriter, content.reqHeader.binSize) _, _ = CopyBytes(ctx, content.binReader, content.binWriter, content.reqHeader.binSize)
return return
} }
func (content *Content) ReadResponseAsync(wg *sync.WaitGroup, errChan chan error) { func (content *Content) readResponseAsync(wg *sync.WaitGroup, errChan chan error) {
var err error var err error
exitFunc := func() { exitFunc := func() {
errChan <- err errChan <- err
@@ -285,13 +327,13 @@ func (content *Content) ReadResponseAsync(wg *sync.WaitGroup, errChan chan error
return return
} }
func (content *Content) DownloadBin() error { func (content *Content) downloadBin(ctx context.Context) error {
var err error var err error
_, err = CopyBytes(content.binReader, content.binWriter, content.resHeader.binSize) _, err = CopyBytes(ctx, content.binReader, content.binWriter, content.resHeader.binSize)
return err return err
} }
func (content *Content) BindResponse() error { func (content *Content) bindResponse() error {
var err error var err error
err = encoder.Unmarshal(content.resPacket.rcpPayload, content.resBlock) err = encoder.Unmarshal(content.resPacket.rcpPayload, content.resBlock)

43
example/cli/netcli.go
View File

@@ -7,33 +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.HelloParams{ params := api.HelloParams{
Message: "hello, server!", Message: "hello, server!",
} }
result := api.HelloResult{} 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
} }

7
example/go.mod
View File

@@ -2,9 +2,4 @@ module netsrv
go 1.19 go 1.19
require github.com/kindsoldier/dsrpc v1.1.1 require github.com/kindsoldier/dsrpc v1.2.1
require (
github.com/vmihailenco/msgpack/v5 v5.3.5 // indirect
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
)

16
example/go.sum
View File

@@ -1,16 +1,6 @@
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 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/kindsoldier/dsrpc v1.1.1 h1:BZWuZexJVjhORsu30u3IRx5Piz1Za0TaIIBEXDOAqAM= github.com/kindsoldier/dsrpc v1.2.1 h1:sw1a3MAD83Do1Fu+Dh+AHArrwVMgZ/KTLUWWTkQ6vj8=
github.com/kindsoldier/dsrpc v1.1.1/go.mod h1:HgludPEdSw8leTlAt9N5qzj3/u7xdMpHSvop/TapHKY= 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=
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/vmihailenco/msgpack/v5 v5.3.5 h1:5gO0H1iULLWGhs2H5tbAHIZTV8/cYafcFOr9znI5mJU=
github.com/vmihailenco/msgpack/v5 v5.3.5/go.mod h1:7xyJ9e+0+9SaZT0Wt1RGleJXzli6Q/V5KbhBonMG9jc=
github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
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 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

2
example/srv/netsrv.go
View File

@@ -26,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)

204
exec_test.go
View File

@@ -6,6 +6,7 @@ package dsrpc
import ( import (
"bytes" "bytes"
"context"
"encoding/json" "encoding/json"
"errors" "errors"
"io" "io"
@@ -16,26 +17,47 @@ import (
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
const HelloMethod string = "hello"
type HelloParams struct {
Message string `json:"message" msgpack:"message"`
}
type HelloResult struct {
Message string `json:"message" msgpack:"message"`
}
const SaveMethod string = "save"
type SaveParams HelloParams
type SaveResult HelloResult
const LoadMethod string = "load"
type LoadParams HelloParams
type LoadResult HelloResult
func TestLocalExec(t *testing.T) { func TestLocalExec(t *testing.T) {
var err error var err error
params := NewHelloParams() params := HelloParams{}
params.Message = "hello server!" params.Message = "hello server!"
result := NewHelloResult() result := HelloResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
err = LocalExec(HelloMethod, params, result, auth, helloHandler) err = LocalExec(HelloMethod, &params, &result, auth, helloHandler)
require.NoError(t, err) require.NoError(t, err)
resultJSON, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON)) resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJson))
} }
func TestLocalSave(t *testing.T) { func TestLocalSave(t *testing.T) {
var err error var err error
params := NewSaveParams() params := SaveParams{}
params.Message = "save data!" params.Message = "save data!"
result := NewHelloResult() result := SaveResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16 var binSize int64 = 16
@@ -45,35 +67,43 @@ func TestLocalSave(t *testing.T) {
reader := bytes.NewReader(binBytes) reader := bytes.NewReader(binBytes)
err = LocalPut(SaveMethod, reader, binSize, params, result, auth, saveHandler) 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) require.NoError(t, err)
resultJSON, _ := json.Marshal(result) resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON)) logDebug("method result:", string(resultJson))
} }
func TestLocalLoad(t *testing.T) { func TestLocalLoad(t *testing.T) {
var err error var err error
params := NewLoadParams() params := LoadParams{}
params.Message = "load data!" params.Message = "load data!"
result := NewHelloResult() result := LoadResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
binBytes := make([]byte, 0) binBytes := make([]byte, 0)
writer := bytes.NewBuffer(binBytes) writer := bytes.NewBuffer(binBytes)
err = LocalGet(LoadMethod, writer, params, result, auth, loadHandler) 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) require.NoError(t, err)
resultJSON, _ := json.Marshal(result) resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON)) logDebug("method result:", string(resultJson))
logDebug("bin size:", len(writer.Bytes())) logDebug("bin size:", len(writer.Bytes()))
} }
func TestNetExec(t *testing.T) { func TestNetExec(t *testing.T) {
go testServ(false) go testServ(false)
time.Sleep(10 * time.Millisecond) time.Sleep(100 * time.Millisecond)
err := clientHello() err := clientHello()
require.NoError(t, err) require.NoError(t, err)
@@ -81,21 +111,21 @@ func TestNetExec(t *testing.T) {
func TestNetSave(t *testing.T) { func TestNetSave(t *testing.T) {
go testServ(false) go testServ(false)
time.Sleep(10 * time.Millisecond) time.Sleep(100 * time.Millisecond)
err := clientSave() err := clientSave()
require.NoError(t, err) require.NoError(t, err)
} }
func TestNetLoad(t *testing.T) { func TestNetLoad(t *testing.T) {
go testServ(false) go testServ(false)
time.Sleep(10 * time.Millisecond) time.Sleep(100 * time.Millisecond)
err := clientLoad() err := clientLoad()
require.NoError(t, err) require.NoError(t, err)
} }
func BenchmarkNetPut(b *testing.B) { func BenchmarkNetPut(b *testing.B) {
go testServ(true) go testServ(true)
time.Sleep(10 * time.Millisecond) time.Sleep(1000 * time.Millisecond)
clientSave() clientSave()
pBench := func(pb *testing.PB) { pBench := func(pb *testing.PB) {
@@ -103,16 +133,16 @@ func BenchmarkNetPut(b *testing.B) {
clientSave() clientSave()
} }
} }
b.SetParallelism(10) b.SetParallelism(2000)
b.RunParallel(pBench) b.RunParallel(pBench)
} }
func clientHello() error { func clientHello() error {
var err error var err error
params := NewHelloParams() params := HelloParams{}
params.Message = "hello server!" params.Message = "hello server!"
result := NewHelloResult() result := HelloResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16 var binSize int64 = 16
@@ -120,22 +150,26 @@ func clientHello() error {
binBytes := make([]byte, binSize) binBytes := make([]byte, binSize)
rand.Read(binBytes) rand.Read(binBytes)
err = Exec("127.0.0.1:8081", HelloMethod, params, result, auth) 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 { if err != nil {
logError("method err:", err) logError("method err:", err)
return err return err
} }
resultJSON, _ := json.Marshal(result) resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON)) logDebug("method result:", string(resultJson))
return err return err
} }
func clientSave() error { func clientSave() error {
var err error var err error
params := NewSaveParams() params := SaveParams{}
params.Message = "save data!" params.Message = "save data!"
result := NewHelloResult() result := SaveResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16 var binSize int64 = 16
@@ -145,34 +179,42 @@ func clientSave() error {
reader := bytes.NewReader(binBytes) reader := bytes.NewReader(binBytes)
err = Put("127.0.0.1:8081", SaveMethod, reader, binSize, params, result, auth) 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 { if err != nil {
logError("method err:", err) logError("method err:", err)
return err return err
} }
resultJSON, _ := json.Marshal(result) resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON)) logDebug("method result:", string(resultJson))
return err return err
} }
func clientLoad() error { func clientLoad() error {
var err error var err error
params := NewLoadParams() params := LoadParams{}
params.Message = "load data!" params.Message = "load data!"
result := NewHelloResult() result := LoadResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
binBytes := make([]byte, 0) binBytes := make([]byte, 0)
writer := bytes.NewBuffer(binBytes) writer := bytes.NewBuffer(binBytes)
err = Get("127.0.0.1:8081", LoadMethod, writer, params, result, auth) 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 { if err != nil {
logError("method err:", err) logError("method err:", err)
return err return err
} }
resultJSON, _ := json.Marshal(result) resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJSON)) logDebug("method result:", string(resultJson))
logDebug("bin size:", len(writer.Bytes())) logDebug("bin size:", len(writer.Bytes()))
return err return err
} }
@@ -191,10 +233,11 @@ func testServ(quiet bool) error {
SetAccessWriter(io.Discard) SetAccessWriter(io.Discard)
SetMessageWriter(io.Discard) SetMessageWriter(io.Discard)
} }
serv := NewService() serv := NewService()
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.PreMiddleware(LogRequest) serv.PreMiddleware(LogRequest)
serv.PreMiddleware(auth) serv.PreMiddleware(auth)
@@ -202,7 +245,7 @@ func testServ(quiet bool) error {
serv.PostMiddleware(LogResponse) serv.PostMiddleware(LogResponse)
serv.PostMiddleware(LogAccess) serv.PostMiddleware(LogAccess)
err = serv.Listen(":8081") err = serv.Listen(":18081")
if err != nil { if err != nil {
return err return err
} }
@@ -219,7 +262,7 @@ func auth(content *Content) error {
pass := []byte("12345") pass := []byte("12345")
auth := content.Auth() auth := content.Auth()
logDebug("auth ", string(auth.JSON())) logDebug("auth ", string(auth.Json()))
ok := CheckHash(ident, pass, reqSalt, reqHash) ok := CheckHash(ident, pass, reqSalt, reqHash)
logDebug("auth ok:", ok) logDebug("auth ok:", ok)
@@ -233,20 +276,24 @@ func auth(content *Content) error {
func helloHandler(content *Content) error { func helloHandler(content *Content) error {
var err error var err error
params := NewHelloParams() params := HelloParams{}
err = content.BindParams(params) err = content.BindParams(&params)
if err != nil { if err != nil {
return err return err
} }
err = content.ReadBin(io.Discard) timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, io.Discard)
if err != nil { if err != nil {
content.SendError(err) content.SendError(err)
return err return err
} }
result := NewHelloResult() result := HelloResult{}
result.Message = "hello, client!" result.Message = "hello, client!"
err = content.SendResult(result, 0) err = content.SendResult(result, 0)
@@ -258,9 +305,9 @@ func helloHandler(content *Content) error {
func saveHandler(content *Content) error { func saveHandler(content *Content) error {
var err error var err error
params := NewSaveParams() params := SaveParams{}
err = content.BindParams(params) err = content.BindParams(&params)
if err != nil { if err != nil {
return err return err
} }
@@ -268,13 +315,17 @@ func saveHandler(content *Content) error {
bufferBytes := make([]byte, 0, 1024) bufferBytes := make([]byte, 0, 1024)
binWriter := bytes.NewBuffer(bufferBytes) binWriter := bytes.NewBuffer(bufferBytes)
err = content.ReadBin(binWriter) timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, binWriter)
if err != nil { if err != nil {
content.SendError(err) content.SendError(err)
return err return err
} }
result := NewSaveResult() result := SaveResult{}
result.Message = "saved successfully!" result.Message = "saved successfully!"
err = content.SendResult(result, 0) err = content.SendResult(result, 0)
@@ -286,14 +337,18 @@ func saveHandler(content *Content) error {
func loadHandler(content *Content) error { func loadHandler(content *Content) error {
var err error var err error
params := NewSaveParams() params := SaveParams{}
err = content.BindParams(params) err = content.BindParams(&params)
if err != nil { if err != nil {
return err return err
} }
err = content.ReadBin(io.Discard) timeout := time.Duration(5 * time.Second)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, io.Discard)
if err != nil { if err != nil {
content.SendError(err) content.SendError(err)
return err return err
@@ -306,60 +361,19 @@ func loadHandler(content *Content) error {
binReader := bytes.NewReader(binBytes) binReader := bytes.NewReader(binBytes)
result := NewSaveResult() result := SaveResult{}
result.Message = "load successfully!" result.Message = "load successfully!"
err = content.SendResult(result, binSize) err = content.SendResult(result, binSize)
if err != nil { if err != nil {
return err return err
} }
binWriter := content.BinWriter() binWriter := content.BinWriter()
_, err = CopyBytes(binReader, binWriter, binSize) _, err = CopyBytes(ctx, binReader, binWriter, binSize)
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
const HelloMethod string = "hello"
type HelloParams struct {
Message string `json:"message" msgpack:"message"`
}
func NewHelloParams() *HelloParams {
return &HelloParams{}
}
type HelloResult struct {
Message string `json:"message" msgpack:"message"`
}
func NewHelloResult() *HelloResult {
return &HelloResult{}
}
const SaveMethod string = "save"
type SaveParams HelloParams
type SaveResult HelloResult
func NewSaveParams() *SaveParams {
return &SaveParams{}
}
func NewSaveResult() *SaveResult {
return &SaveResult{}
}
const LoadMethod string = "load"
type LoadParams HelloParams
type LoadResult HelloResult
func NewLoadParams() *LoadParams {
return &LoadParams{}
}
func NewLoadResult() *LoadResult {
return &LoadResult{}
}

6
go.mod
View File

@@ -2,14 +2,10 @@ module github.com/kindsoldier/dsrpc
go 1.19 go 1.19
require ( require github.com/stretchr/testify v1.8.2
github.com/stretchr/testify v1.8.2
github.com/vmihailenco/msgpack/v5 v5.3.5
)
require ( require (
github.com/davecgh/go-spew v1.1.1 // 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
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect
) )

5
go.sum
View File

@@ -6,15 +6,10 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
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/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= 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/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/testify v1.6.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.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.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 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/vmihailenco/msgpack/v5 v5.3.5 h1:5gO0H1iULLWGhs2H5tbAHIZTV8/cYafcFOr9znI5mJU=
github.com/vmihailenco/msgpack/v5 v5.3.5/go.mod h1:7xyJ9e+0+9SaZT0Wt1RGleJXzli6Q/V5KbhBonMG9jc=
github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
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/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

20
header.go
View File

@@ -44,16 +44,16 @@ func (hdr *Header) Pack() ([]byte, error) {
headerBytes := make([]byte, 0, headerSize) headerBytes := make([]byte, 0, headerSize)
headerBuffer := bytes.NewBuffer(headerBytes) headerBuffer := bytes.NewBuffer(headerBytes)
magicCodeABytes := encoderI64(hdr.magicCodeA) magicCodeABytes := EncoderI64(hdr.magicCodeA)
headerBuffer.Write(magicCodeABytes) headerBuffer.Write(magicCodeABytes)
rpcSizeBytes := encoderI64(hdr.rpcSize) rpcSizeBytes := EncoderI64(hdr.rpcSize)
headerBuffer.Write(rpcSizeBytes) headerBuffer.Write(rpcSizeBytes)
binSizeBytes := encoderI64(hdr.binSize) binSizeBytes := EncoderI64(hdr.binSize)
headerBuffer.Write(binSizeBytes) headerBuffer.Write(binSizeBytes)
magicCodeBBytes := encoderI64(hdr.magicCodeB) magicCodeBBytes := EncoderI64(hdr.magicCodeB)
headerBuffer.Write(magicCodeBBytes) headerBuffer.Write(magicCodeBBytes)
return headerBuffer.Bytes(), err return headerBuffer.Bytes(), err
@@ -77,10 +77,10 @@ func UnpackHeader(headerBytes []byte) (*Header, error) {
headerReader.Read(magicCodeBBytes) headerReader.Read(magicCodeBBytes)
header := &Header{ header := &Header{
magicCodeA: decoderI64(magicCodeABytes), magicCodeA: DecoderI64(magicCodeABytes),
rpcSize: decoderI64(rpcSizeBytes), rpcSize: DecoderI64(rpcSizeBytes),
binSize: decoderI64(binSizeBytes), binSize: DecoderI64(binSizeBytes),
magicCodeB: decoderI64(magicCodeBBytes), magicCodeB: DecoderI64(magicCodeBBytes),
} }
if header.magicCodeA != magicCodeA || header.magicCodeB != magicCodeB { if header.magicCodeA != magicCodeA || header.magicCodeB != magicCodeB {
@@ -90,12 +90,12 @@ func UnpackHeader(headerBytes []byte) (*Header, error) {
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))
} }

19
logger.go
View File

@@ -10,30 +10,27 @@ 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, "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, "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, "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, "access", messages)
} }
func SetAccessWriter(writer io.Writer) { func SetAccessWriter(writer io.Writer) {

2
request.go
View File

@@ -9,7 +9,7 @@ package dsrpc
import ( import (
"encoding/json" "encoding/json"
encoder "github.com/vmihailenco/msgpack/v5" encoder "encoding/json"
) )
type EmptyParams struct{} type EmptyParams struct{}

2
response.go
View File

@@ -7,7 +7,7 @@ package dsrpc
import ( import (
"encoding/json" "encoding/json"
encoder "github.com/vmihailenco/msgpack/v5" encoder "encoding/json"
) )
type EmptyResult struct{} type EmptyResult struct{}

75
server.go
View File

@@ -6,6 +6,7 @@ package dsrpc
import ( import (
"context" "context"
"crypto/tls"
"errors" "errors"
"fmt" "fmt"
"io" "io"
@@ -13,21 +14,23 @@ import (
"sync" "sync"
"time" "time"
encoder "github.com/vmihailenco/msgpack/v5" encoder "encoding/json"
) )
type HandlerFunc = func(*Content) 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 keepalive bool
kaTime time.Duration kaTime time.Duration
kaMtx sync.Mutex kaMtx sync.Mutex
listener net.Listener
tcpListener *net.TCPListener
} }
func NewService() *Service { func NewService() *Service {
@@ -52,7 +55,7 @@ func (svc *Service) PostMiddleware(mw HandlerFunc) {
svc.postMw = append(svc.postMw, mw) svc.postMw = append(svc.postMw, mw)
} }
func (svc *Service) Handler(method string, handler HandlerFunc) { func (svc *Service) Handle(method string, handler HandlerFunc) {
svc.handlers[method] = handler svc.handlers[method] = handler
} }
@@ -77,14 +80,14 @@ func (svc *Service) Listen(address string) error {
err = fmt.Errorf("unable to resolve adddress: %s", err) err = fmt.Errorf("unable to resolve adddress: %s", err)
return err return err
} }
listener, err := net.ListenTCP("tcp", addr) svc.tcpListener, err = net.ListenTCP("tcp", addr)
if err != nil { if err != nil {
err = fmt.Errorf("unable to start listener: %s", err) err = fmt.Errorf("unable to start listener: %s", err)
return err return err
} }
for { for {
conn, err := listener.AcceptTCP() conn, err := svc.tcpListener.AcceptTCP()
if err != nil { if err != nil {
logError("conn accept err:", err) logError("conn accept err:", err)
} }
@@ -94,6 +97,33 @@ func (svc *Service) Listen(address string) error {
default: default:
} }
svc.wg.Add(1) svc.wg.Add(1)
go svc.handleTCPConn(conn, svc.wg)
}
return err
}
func (svc *Service) ListenTLS(address string, tlsConfig *tls.Config) error {
var err error
logInfo("server listen:", address)
svc.listener, err = tls.Listen("tcp", address, tlsConfig)
if err != nil {
err = fmt.Errorf("unable to start listener: %s", err)
return err
}
for {
conn, err := svc.listener.Accept()
if err != nil {
logError("conn accept err:", err)
}
select {
case <-svc.ctx.Done():
logInfo("accept loop done")
return err
default:
}
svc.wg.Add(1)
go svc.handleConn(conn, svc.wg) go svc.handleConn(conn, svc.wg)
} }
return err return err
@@ -109,6 +139,12 @@ func (svc *Service) Stop() error {
var err error var err error
// Disable new connection // Disable new connection
logInfo("cancel rpc accept loop") logInfo("cancel rpc accept loop")
if svc.listener != nil {
svc.listener.Close()
}
if svc.tcpListener != nil {
svc.tcpListener.Close()
}
svc.cancel() svc.cancel()
// Wait handlers // Wait handlers
logInfo("wait rpc handlers") logInfo("wait rpc handlers")
@@ -116,9 +152,8 @@ func (svc *Service) Stop() error {
return err return err
} }
func (svc *Service) handleConn(conn *net.TCPConn, wg *sync.WaitGroup) { func (svc *Service) handleTCPConn(conn *net.TCPConn, wg *sync.WaitGroup) {
var err error var err error
if svc.keepalive { if svc.keepalive {
err = conn.SetKeepAlive(true) err = conn.SetKeepAlive(true)
if err != nil { if err != nil {
@@ -133,6 +168,12 @@ func (svc *Service) handleConn(conn *net.TCPConn, wg *sync.WaitGroup) {
} }
} }
} }
svc.handleConn(conn, wg)
}
func (svc *Service) handleConn(conn net.Conn, wg *sync.WaitGroup) {
var err error
content := CreateContent(conn) content := CreateContent(conn)
remoteAddr := conn.RemoteAddr().String() remoteAddr := conn.RemoteAddr().String()
@@ -232,9 +273,9 @@ func (content *Content) BinSize() int64 {
return content.reqHeader.binSize return content.reqHeader.binSize
} }
func (content *Content) ReadBin(writer io.Writer) error { func (content *Content) ReadBin(ctx context.Context, writer io.Writer) error {
var err error var err error
_, err = CopyBytes(content.sockReader, writer, content.reqHeader.binSize) _, err = CopyBytes(ctx, content.sockReader, writer, content.reqHeader.binSize)
return err return err
} }

9
tools.go
View File

@@ -5,6 +5,7 @@
package dsrpc package dsrpc
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"io" "io"
@@ -16,7 +17,7 @@ func ReadBytes(reader io.Reader, size int64) ([]byte, error) {
return buffer[0:read], 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 * 16 var bSize int64 = 1024 * 16
var total int64 = 0 var total int64 = 0
@@ -24,6 +25,12 @@ func CopyBytes(reader io.Reader, writer io.Writer, dataSize int64) (int64, error
buffer := make([]byte, bSize) buffer := make([]byte, bSize)
for { for {
select {
case <-ctx.Done():
return total, errors.New("break by context")
default:
}
if reader == nil { if reader == nil {
return total, errors.New("reader is nil") return total, errors.New("reader is nil")
} }

33
validate.go
View File

@@ -5,6 +5,7 @@
package dsrpc package dsrpc
import ( import (
"context"
"io" "io"
"net" "net"
) )
@@ -27,11 +28,11 @@ func LocalExec(method string, param, result any, auth *Auth, handler HandlerFunc
content.resBlock.Result = result content.resBlock.Result = result
} }
err = content.CreateRequest() err = content.createRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.WriteRequest() err = content.writeRequest()
if err != nil { if err != nil {
return err return err
} }
@@ -39,11 +40,11 @@ func LocalExec(method string, param, result any, auth *Auth, handler HandlerFunc
if err != nil { if err != nil {
return err return err
} }
err = content.ReadResponse() err = content.readResponse()
if err != nil { if err != nil {
return err return err
} }
err = content.BindResponse() err = content.bindResponse()
if err != nil { if err != nil {
return err return err
} }
@@ -51,7 +52,7 @@ func LocalExec(method string, param, result any, auth *Auth, handler HandlerFunc
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
@@ -75,15 +76,15 @@ func LocalPut(method string, reader io.Reader, size int64, param, result any, au
content.reqHeader.binSize = size content.reqHeader.binSize = size
err = content.CreateRequest() err = content.createRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.WriteRequest() err = content.writeRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.UploadBin() err = content.uploadBin(ctx)
if err != nil { if err != nil {
return err return err
} }
@@ -91,18 +92,18 @@ func LocalPut(method string, reader io.Reader, size int64, param, result any, au
if err != nil { if err != nil {
return err return err
} }
err = content.ReadResponse() err = content.readResponse()
if err != nil { if err != nil {
return err return err
} }
err = content.BindResponse() err = content.bindResponse()
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
func LocalGet(method string, writer io.Writer, param, result any, auth *Auth, handler HandlerFunc) error { func LocalGet(ctx context.Context, method string, writer io.Writer, param, result any, auth *Auth, handler HandlerFunc) error {
var err error var err error
cliConn, srvConn := NewFConn() cliConn, srvConn := NewFConn()
@@ -123,11 +124,11 @@ func LocalGet(method string, writer io.Writer, param, result any, auth *Auth, ha
content.binReader = cliConn content.binReader = cliConn
content.binWriter = writer content.binWriter = writer
err = content.CreateRequest() err = content.createRequest()
if err != nil { if err != nil {
return err return err
} }
err = content.WriteRequest() err = content.writeRequest()
if err != nil { if err != nil {
return err return err
} }
@@ -136,15 +137,15 @@ func LocalGet(method string, writer io.Writer, param, result any, auth *Auth, ha
if err != nil { if err != nil {
return err return err
} }
err = content.ReadResponse() err = content.readResponse()
if err != nil { if err != nil {
return err return err
} }
err = content.DownloadBin() err = content.downloadBin(ctx)
if err != nil { if err != nil {
return err return err
} }
err = content.BindResponse() err = content.bindResponse()
if err != nil { if err != nil {
return err return err
} }

2
xauth.go
View File

@@ -26,7 +26,7 @@ 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
} }