Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f814bb7d24 | |||
| aed740c933 | |||
| 4716c25cf6 | |||
| 9bc3aef167 | |||
| 0e5321d51a | |||
| 6f9835f399 | |||
| ef8db1d198 |
112
client.go
112
client.go
@@ -5,16 +5,18 @@
|
||||
package dsrpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"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
|
||||
|
||||
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()
|
||||
|
||||
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
|
||||
content := CreateContent(conn)
|
||||
|
||||
@@ -51,11 +65,11 @@ func ConnPut(conn net.Conn, method string, reader io.Reader, binSize int64, para
|
||||
|
||||
content.reqHeader.binSize = binSize
|
||||
|
||||
err = content.CreateRequest()
|
||||
err = content.createRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.WriteRequest()
|
||||
err = content.writeRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -64,24 +78,24 @@ func ConnPut(conn net.Conn, method string, reader io.Reader, binSize int64, para
|
||||
errChan := make(chan error, 1)
|
||||
|
||||
wg.Add(1)
|
||||
go content.ReadResponseAsync(&wg, errChan)
|
||||
go content.readResponseAsync(&wg, errChan)
|
||||
|
||||
wg.Add(1)
|
||||
go content.UploadBinAsync(&wg)
|
||||
go content.uploadBinAsync(ctx, &wg)
|
||||
|
||||
wg.Wait()
|
||||
err = <-errChan
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.BindResponse()
|
||||
err = content.bindResponse()
|
||||
if err != nil {
|
||||
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
|
||||
|
||||
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()
|
||||
|
||||
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
|
||||
|
||||
content := CreateContent(conn)
|
||||
@@ -116,30 +142,30 @@ func ConnGet(conn net.Conn, method string, writer io.Writer, param, result any,
|
||||
content.binReader = conn
|
||||
content.binWriter = writer
|
||||
|
||||
err = content.CreateRequest()
|
||||
err = content.createRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.WriteRequest()
|
||||
err = content.writeRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.ReadResponse()
|
||||
err = content.readResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.DownloadBin()
|
||||
err = content.downloadBin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.BindResponse()
|
||||
err = content.bindResponse()
|
||||
if err != nil {
|
||||
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
|
||||
|
||||
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()
|
||||
|
||||
err = ConnExec(conn, method, param, result, auth)
|
||||
err = ConnExec(ctx, conn, method, param, result, auth)
|
||||
if err != nil {
|
||||
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
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
err = content.CreateRequest()
|
||||
err = content.createRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.WriteRequest()
|
||||
err = content.writeRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.ReadResponse()
|
||||
err = content.readResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.BindResponse()
|
||||
err = content.bindResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (content *Content) CreateRequest() error {
|
||||
func (content *Content) createRequest() error {
|
||||
var err error
|
||||
|
||||
content.reqPacket.rcpPayload, err = content.reqBlock.Pack()
|
||||
@@ -212,7 +254,7 @@ func (content *Content) CreateRequest() error {
|
||||
return err
|
||||
}
|
||||
|
||||
func (content *Content) WriteRequest() error {
|
||||
func (content *Content) writeRequest() error {
|
||||
var err error
|
||||
_, err = content.sockWriter.Write(content.reqPacket.header)
|
||||
if err != nil {
|
||||
@@ -225,13 +267,13 @@ func (content *Content) WriteRequest() error {
|
||||
return err
|
||||
}
|
||||
|
||||
func (content *Content) UploadBin() error {
|
||||
func (content *Content) uploadBin(ctx context.Context) 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
|
||||
}
|
||||
|
||||
func (content *Content) ReadResponse() error {
|
||||
func (content *Content) readResponse() error {
|
||||
var err error
|
||||
|
||||
content.resPacket.header, err = ReadBytes(content.sockReader, headerSize)
|
||||
@@ -250,16 +292,16 @@ func (content *Content) ReadResponse() error {
|
||||
return err
|
||||
}
|
||||
|
||||
func (content *Content) UploadBinAsync(wg *sync.WaitGroup) {
|
||||
func (content *Content) uploadBinAsync(ctx context.Context, wg *sync.WaitGroup) {
|
||||
exitFunc := func() {
|
||||
wg.Done()
|
||||
}
|
||||
defer exitFunc()
|
||||
_, _ = CopyBytes(content.binReader, content.binWriter, content.reqHeader.binSize)
|
||||
_, _ = CopyBytes(ctx, content.binReader, content.binWriter, content.reqHeader.binSize)
|
||||
return
|
||||
}
|
||||
|
||||
func (content *Content) ReadResponseAsync(wg *sync.WaitGroup, errChan chan error) {
|
||||
func (content *Content) readResponseAsync(wg *sync.WaitGroup, errChan chan error) {
|
||||
var err error
|
||||
exitFunc := func() {
|
||||
errChan <- err
|
||||
@@ -285,13 +327,13 @@ func (content *Content) ReadResponseAsync(wg *sync.WaitGroup, errChan chan error
|
||||
return
|
||||
}
|
||||
|
||||
func (content *Content) DownloadBin() error {
|
||||
func (content *Content) downloadBin(ctx context.Context) 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
|
||||
}
|
||||
|
||||
func (content *Content) BindResponse() error {
|
||||
func (content *Content) bindResponse() error {
|
||||
var err error
|
||||
|
||||
err = encoder.Unmarshal(content.resPacket.rcpPayload, content.resBlock)
|
||||
|
||||
@@ -7,33 +7,40 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/kindsoldier/dsrpc"
|
||||
"netsrv/api"
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/kindsoldier/dsrpc"
|
||||
|
||||
"netsrv/api"
|
||||
)
|
||||
|
||||
func main() {
|
||||
|
||||
err := exec()
|
||||
if err != nil {
|
||||
fmt.Println("exec err:", err)
|
||||
}
|
||||
err := exec()
|
||||
if err != nil {
|
||||
fmt.Println("exec err:", err)
|
||||
}
|
||||
}
|
||||
|
||||
func exec() error {
|
||||
var err error
|
||||
var err error
|
||||
|
||||
params := api.HelloParams{
|
||||
Message: "hello, server!",
|
||||
}
|
||||
params := api.HelloParams{
|
||||
Message: "hello, server!",
|
||||
}
|
||||
|
||||
result := api.HelloResult{}
|
||||
result := api.HelloResult{}
|
||||
|
||||
err = dsrpc.Exec("127.0.0.1:8081", api.HelloMethod, ¶ms, &result, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
fmt.Println("result:", result.Message)
|
||||
return err
|
||||
err = dsrpc.Exec(ctx, "127.0.0.1:8081", api.HelloMethod, ¶ms, &result, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fmt.Println("result:", result.Message)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -2,9 +2,4 @@ module netsrv
|
||||
|
||||
go 1.19
|
||||
|
||||
require github.com/kindsoldier/dsrpc v1.1.2
|
||||
|
||||
require (
|
||||
github.com/vmihailenco/msgpack/v5 v5.3.5 // indirect
|
||||
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
|
||||
)
|
||||
require github.com/kindsoldier/dsrpc v1.2.1
|
||||
|
||||
@@ -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/kindsoldier/dsrpc v1.1.2 h1:bFTIGpRSMq5OK1a3dHQxLPMxf6R+Ik15slkqNN0QrNE=
|
||||
github.com/kindsoldier/dsrpc v1.1.2/go.mod h1:KG8x2ZPid/hPJdhkUHtt1mDulWPVhj9fh/1XL3Z2xT8=
|
||||
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/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
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.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
|
||||
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=
|
||||
|
||||
44
exec_test.go
44
exec_test.go
@@ -6,6 +6,7 @@ package dsrpc
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
@@ -66,7 +67,10 @@ func TestLocalSave(t *testing.T) {
|
||||
|
||||
reader := bytes.NewReader(binBytes)
|
||||
|
||||
err = LocalPut(SaveMethod, reader, binSize, ¶ms, &result, auth, saveHandler)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
err = LocalPut(ctx, SaveMethod, reader, binSize, ¶ms, &result, auth, saveHandler)
|
||||
require.NoError(t, err)
|
||||
|
||||
resultJson, _ := json.Marshal(result)
|
||||
@@ -84,7 +88,10 @@ func TestLocalLoad(t *testing.T) {
|
||||
binBytes := make([]byte, 0)
|
||||
writer := bytes.NewBuffer(binBytes)
|
||||
|
||||
err = LocalGet(LoadMethod, writer, ¶ms, &result, auth, loadHandler)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
err = LocalGet(ctx, LoadMethod, writer, ¶ms, &result, auth, loadHandler)
|
||||
require.NoError(t, err)
|
||||
|
||||
resultJson, _ := json.Marshal(result)
|
||||
@@ -141,7 +148,10 @@ func clientHello() error {
|
||||
binBytes := make([]byte, binSize)
|
||||
rand.Read(binBytes)
|
||||
|
||||
err = Exec("127.0.0.1:8081", HelloMethod, ¶ms, &result, auth)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
err = Exec(ctx, "127.0.0.1:8081", HelloMethod, ¶ms, &result, auth)
|
||||
if err != nil {
|
||||
logError("method err:", err)
|
||||
return err
|
||||
@@ -166,7 +176,10 @@ func clientSave() error {
|
||||
|
||||
reader := bytes.NewReader(binBytes)
|
||||
|
||||
err = Put("127.0.0.1:8081", SaveMethod, reader, binSize, ¶ms, &result, auth)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
err = Put(ctx, "127.0.0.1:8081", SaveMethod, reader, binSize, ¶ms, &result, auth)
|
||||
if err != nil {
|
||||
logError("method err:", err)
|
||||
return err
|
||||
@@ -187,7 +200,10 @@ func clientLoad() error {
|
||||
binBytes := make([]byte, 0)
|
||||
writer := bytes.NewBuffer(binBytes)
|
||||
|
||||
err = Get("127.0.0.1:8081", LoadMethod, writer, ¶ms, &result, auth)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
err = Get(ctx, "127.0.0.1:8081", LoadMethod, writer, ¶ms, &result, auth)
|
||||
if err != nil {
|
||||
logError("method err:", err)
|
||||
return err
|
||||
@@ -261,7 +277,10 @@ func helloHandler(content *Content) error {
|
||||
return err
|
||||
}
|
||||
|
||||
err = content.ReadBin(io.Discard)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
err = content.ReadBin(ctx, io.Discard)
|
||||
if err != nil {
|
||||
content.SendError(err)
|
||||
return err
|
||||
@@ -289,7 +308,10 @@ func saveHandler(content *Content) error {
|
||||
bufferBytes := make([]byte, 0, 1024)
|
||||
binWriter := bytes.NewBuffer(bufferBytes)
|
||||
|
||||
err = content.ReadBin(binWriter)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
err = content.ReadBin(ctx, binWriter)
|
||||
if err != nil {
|
||||
content.SendError(err)
|
||||
return err
|
||||
@@ -314,7 +336,10 @@ func loadHandler(content *Content) error {
|
||||
return err
|
||||
}
|
||||
|
||||
err = content.ReadBin(io.Discard)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
defer cancel()
|
||||
|
||||
err = content.ReadBin(ctx, io.Discard)
|
||||
if err != nil {
|
||||
content.SendError(err)
|
||||
return err
|
||||
@@ -334,8 +359,9 @@ func loadHandler(content *Content) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
binWriter := content.BinWriter()
|
||||
_, err = CopyBytes(binReader, binWriter, binSize)
|
||||
_, err = CopyBytes(ctx, binReader, binWriter, binSize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
6
go.mod
6
go.mod
@@ -2,14 +2,10 @@ module github.com/kindsoldier/dsrpc
|
||||
|
||||
go 1.19
|
||||
|
||||
require (
|
||||
github.com/stretchr/testify v1.8.2
|
||||
github.com/vmihailenco/msgpack/v5 v5.3.5
|
||||
)
|
||||
require github.com/stretchr/testify v1.8.2
|
||||
|
||||
require (
|
||||
github.com/davecgh/go-spew v1.1.1 // 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
|
||||
)
|
||||
|
||||
5
go.sum
5
go.sum
@@ -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.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.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.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=
|
||||
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/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
||||
20
header.go
20
header.go
@@ -44,16 +44,16 @@ func (hdr *Header) Pack() ([]byte, error) {
|
||||
headerBytes := make([]byte, 0, headerSize)
|
||||
headerBuffer := bytes.NewBuffer(headerBytes)
|
||||
|
||||
magicCodeABytes := encoderI64(hdr.magicCodeA)
|
||||
magicCodeABytes := EncoderI64(hdr.magicCodeA)
|
||||
headerBuffer.Write(magicCodeABytes)
|
||||
|
||||
rpcSizeBytes := encoderI64(hdr.rpcSize)
|
||||
rpcSizeBytes := EncoderI64(hdr.rpcSize)
|
||||
headerBuffer.Write(rpcSizeBytes)
|
||||
|
||||
binSizeBytes := encoderI64(hdr.binSize)
|
||||
binSizeBytes := EncoderI64(hdr.binSize)
|
||||
headerBuffer.Write(binSizeBytes)
|
||||
|
||||
magicCodeBBytes := encoderI64(hdr.magicCodeB)
|
||||
magicCodeBBytes := EncoderI64(hdr.magicCodeB)
|
||||
headerBuffer.Write(magicCodeBBytes)
|
||||
|
||||
return headerBuffer.Bytes(), err
|
||||
@@ -77,10 +77,10 @@ func UnpackHeader(headerBytes []byte) (*Header, error) {
|
||||
headerReader.Read(magicCodeBBytes)
|
||||
|
||||
header := &Header{
|
||||
magicCodeA: decoderI64(magicCodeABytes),
|
||||
rpcSize: decoderI64(rpcSizeBytes),
|
||||
binSize: decoderI64(binSizeBytes),
|
||||
magicCodeB: decoderI64(magicCodeBBytes),
|
||||
magicCodeA: DecoderI64(magicCodeABytes),
|
||||
rpcSize: DecoderI64(rpcSizeBytes),
|
||||
binSize: DecoderI64(binSizeBytes),
|
||||
magicCodeB: DecoderI64(magicCodeBBytes),
|
||||
}
|
||||
|
||||
if header.magicCodeA != magicCodeA || header.magicCodeB != magicCodeB {
|
||||
@@ -90,12 +90,12 @@ func UnpackHeader(headerBytes []byte) (*Header, error) {
|
||||
return header, err
|
||||
}
|
||||
|
||||
func encoderI64(i int64) []byte {
|
||||
func EncoderI64(i int64) []byte {
|
||||
buffer := make([]byte, sizeOfInt64)
|
||||
binary.BigEndian.PutUint64(buffer, uint64(i))
|
||||
return buffer
|
||||
}
|
||||
|
||||
func decoderI64(b []byte) int64 {
|
||||
func DecoderI64(b []byte) int64 {
|
||||
return int64(binary.BigEndian.Uint64(b))
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@ package dsrpc
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
encoder "github.com/vmihailenco/msgpack/v5"
|
||||
encoder "encoding/json"
|
||||
)
|
||||
|
||||
type EmptyParams struct{}
|
||||
|
||||
@@ -7,7 +7,7 @@ package dsrpc
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
encoder "github.com/vmihailenco/msgpack/v5"
|
||||
encoder "encoding/json"
|
||||
)
|
||||
|
||||
type EmptyResult struct{}
|
||||
|
||||
44
server.go
44
server.go
@@ -6,6 +6,7 @@ package dsrpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -13,7 +14,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
encoder "github.com/vmihailenco/msgpack/v5"
|
||||
encoder "encoding/json"
|
||||
)
|
||||
|
||||
type HandlerFunc = func(*Content) error
|
||||
@@ -52,7 +53,7 @@ func (svc *Service) PostMiddleware(mw HandlerFunc) {
|
||||
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
|
||||
}
|
||||
|
||||
@@ -94,6 +95,32 @@ func (svc *Service) Listen(address string) error {
|
||||
default:
|
||||
}
|
||||
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)
|
||||
|
||||
listener, err := tls.Listen("tcp", address, tlsConfig)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("unable to start listener: %s", err)
|
||||
return err
|
||||
}
|
||||
|
||||
for {
|
||||
conn, err := listener.Accept()
|
||||
if err != nil {
|
||||
logError("conn accept err:", err)
|
||||
}
|
||||
select {
|
||||
case <-svc.ctx.Done():
|
||||
return err
|
||||
default:
|
||||
}
|
||||
svc.wg.Add(1)
|
||||
go svc.handleConn(conn, svc.wg)
|
||||
}
|
||||
return err
|
||||
@@ -116,9 +143,8 @@ func (svc *Service) Stop() error {
|
||||
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
|
||||
|
||||
if svc.keepalive {
|
||||
err = conn.SetKeepAlive(true)
|
||||
if err != nil {
|
||||
@@ -133,6 +159,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)
|
||||
|
||||
remoteAddr := conn.RemoteAddr().String()
|
||||
@@ -232,9 +264,9 @@ func (content *Content) BinSize() int64 {
|
||||
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
|
||||
_, err = CopyBytes(content.sockReader, writer, content.reqHeader.binSize)
|
||||
_, err = CopyBytes(ctx, content.sockReader, writer, content.reqHeader.binSize)
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
9
tools.go
9
tools.go
@@ -5,6 +5,7 @@
|
||||
package dsrpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -16,7 +17,7 @@ func ReadBytes(reader io.Reader, size int64) ([]byte, error) {
|
||||
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 bSize int64 = 1024 * 16
|
||||
var total int64 = 0
|
||||
@@ -24,6 +25,12 @@ func CopyBytes(reader io.Reader, writer io.Writer, dataSize int64) (int64, error
|
||||
buffer := make([]byte, bSize)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return total, errors.New("break by context")
|
||||
default:
|
||||
}
|
||||
|
||||
if reader == nil {
|
||||
return total, errors.New("reader is nil")
|
||||
}
|
||||
|
||||
33
validate.go
33
validate.go
@@ -5,6 +5,7 @@
|
||||
package dsrpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net"
|
||||
)
|
||||
@@ -27,11 +28,11 @@ func LocalExec(method string, param, result any, auth *Auth, handler HandlerFunc
|
||||
content.resBlock.Result = result
|
||||
}
|
||||
|
||||
err = content.CreateRequest()
|
||||
err = content.createRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.WriteRequest()
|
||||
err = content.writeRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -39,11 +40,11 @@ func LocalExec(method string, param, result any, auth *Auth, handler HandlerFunc
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.ReadResponse()
|
||||
err = content.readResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.BindResponse()
|
||||
err = content.bindResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -51,7 +52,7 @@ func LocalExec(method string, param, result any, auth *Auth, handler HandlerFunc
|
||||
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
|
||||
|
||||
@@ -75,15 +76,15 @@ func LocalPut(method string, reader io.Reader, size int64, param, result any, au
|
||||
|
||||
content.reqHeader.binSize = size
|
||||
|
||||
err = content.CreateRequest()
|
||||
err = content.createRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.WriteRequest()
|
||||
err = content.writeRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.UploadBin()
|
||||
err = content.uploadBin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -91,18 +92,18 @@ func LocalPut(method string, reader io.Reader, size int64, param, result any, au
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.ReadResponse()
|
||||
err = content.readResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.BindResponse()
|
||||
err = content.bindResponse()
|
||||
if err != nil {
|
||||
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
|
||||
|
||||
cliConn, srvConn := NewFConn()
|
||||
@@ -123,11 +124,11 @@ func LocalGet(method string, writer io.Writer, param, result any, auth *Auth, ha
|
||||
content.binReader = cliConn
|
||||
content.binWriter = writer
|
||||
|
||||
err = content.CreateRequest()
|
||||
err = content.createRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.WriteRequest()
|
||||
err = content.writeRequest()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -136,15 +137,15 @@ func LocalGet(method string, writer io.Writer, param, result any, auth *Auth, ha
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.ReadResponse()
|
||||
err = content.readResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.DownloadBin()
|
||||
err = content.downloadBin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = content.BindResponse()
|
||||
err = content.bindResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user