Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 498abc18b4 | |||
| 5ea1af7fb1 | |||
| bc578cf25c | |||
| cb6d26a7cc | |||
| f814bb7d24 | |||
| aed740c933 | |||
| 4716c25cf6 | |||
| 9bc3aef167 |
@@ -44,7 +44,7 @@ func server() error {
|
||||
serv := dsrpc.NewService()
|
||||
|
||||
cont := NewController()
|
||||
serv.Handler(api.HelloMethod, cont.HelloHandler)
|
||||
serv.Handle(api.HelloMethod, cont.HelloHandler)
|
||||
|
||||
serv.PreMiddleware(dsrpc.LogRequest)
|
||||
serv.PostMiddleware(dsrpc.LogResponse)
|
||||
@@ -220,9 +220,9 @@ func sampleServ(quiet bool) error {
|
||||
serv.PreMiddleware(authMiddleware)
|
||||
serv.PreMiddleware(dsrpc.LogRequest)
|
||||
|
||||
serv.Handler(HelloMethod, helloHandler)
|
||||
serv.Handler(SaveMethod, saveHandler)
|
||||
serv.Handler(LoadMethod, loadHandler)
|
||||
serv.Handle(HelloMethod, helloHandler)
|
||||
serv.Handle(SaveMethod, saveHandler)
|
||||
serv.Handle(LoadMethod, loadHandler)
|
||||
|
||||
serv.PostMiddleware(dsrpc.LogResponse)
|
||||
serv.PostMiddleware(dsrpc.LogAccess)
|
||||
|
||||
41
client.go
41
client.go
@@ -6,6 +6,7 @@ package dsrpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -32,6 +33,18 @@ func Put(ctx context.Context, address string, method string, reader io.Reader, b
|
||||
return ConnPut(ctx, conn, method, reader, binSize, 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
|
||||
|
||||
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)
|
||||
@@ -99,6 +112,18 @@ func Get(ctx context.Context, address string, method string, writer io.Writer, p
|
||||
return ConnGet(ctx, conn, method, writer, param, result, auth)
|
||||
}
|
||||
|
||||
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
|
||||
|
||||
@@ -161,6 +186,22 @@ func Exec(ctx context.Context, address, method string, param any, result any, au
|
||||
return err
|
||||
}
|
||||
|
||||
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
|
||||
|
||||
|
||||
@@ -2,9 +2,4 @@ module netsrv
|
||||
|
||||
go 1.19
|
||||
|
||||
require github.com/kindsoldier/dsrpc v1.2.0
|
||||
|
||||
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.2.0 h1:5ZAFeqo6qyFm0OzPZEpabgf49/yMSA1ZKnOhHr4R/5c=
|
||||
github.com/kindsoldier/dsrpc v1.2.0/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=
|
||||
|
||||
@@ -26,7 +26,7 @@ func server() error {
|
||||
serv := dsrpc.NewService()
|
||||
|
||||
cont := NewController()
|
||||
serv.Handler(api.HelloMethod, cont.HelloHandler)
|
||||
serv.Handle(api.HelloMethod, cont.HelloHandler)
|
||||
|
||||
serv.PreMiddleware(dsrpc.LogRequest)
|
||||
serv.PostMiddleware(dsrpc.LogResponse)
|
||||
|
||||
47
exec_test.go
47
exec_test.go
@@ -67,7 +67,8 @@ func TestLocalSave(t *testing.T) {
|
||||
|
||||
reader := bytes.NewReader(binBytes)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
timeout := time.Duration(5 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
err = LocalPut(ctx, SaveMethod, reader, binSize, ¶ms, &result, auth, saveHandler)
|
||||
@@ -88,7 +89,8 @@ func TestLocalLoad(t *testing.T) {
|
||||
binBytes := make([]byte, 0)
|
||||
writer := bytes.NewBuffer(binBytes)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
timeout := time.Duration(5 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
err = LocalGet(ctx, LoadMethod, writer, ¶ms, &result, auth, loadHandler)
|
||||
@@ -101,7 +103,7 @@ func TestLocalLoad(t *testing.T) {
|
||||
|
||||
func TestNetExec(t *testing.T) {
|
||||
go testServ(false)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
err := clientHello()
|
||||
|
||||
require.NoError(t, err)
|
||||
@@ -109,21 +111,21 @@ func TestNetExec(t *testing.T) {
|
||||
|
||||
func TestNetSave(t *testing.T) {
|
||||
go testServ(false)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
err := clientSave()
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestNetLoad(t *testing.T) {
|
||||
go testServ(false)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
err := clientLoad()
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func BenchmarkNetPut(b *testing.B) {
|
||||
go testServ(true)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
time.Sleep(1000 * time.Millisecond)
|
||||
clientSave()
|
||||
|
||||
pBench := func(pb *testing.PB) {
|
||||
@@ -148,10 +150,11 @@ func clientHello() error {
|
||||
binBytes := make([]byte, binSize)
|
||||
rand.Read(binBytes)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
timeout := time.Duration(5 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
err = Exec(ctx, "127.0.0.1:8081", HelloMethod, ¶ms, &result, auth)
|
||||
err = Exec(ctx, "127.0.0.1:18081", HelloMethod, ¶ms, &result, auth)
|
||||
if err != nil {
|
||||
logError("method err:", err)
|
||||
return err
|
||||
@@ -176,10 +179,11 @@ func clientSave() error {
|
||||
|
||||
reader := bytes.NewReader(binBytes)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
timeout := time.Duration(5 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
err = Put(ctx, "127.0.0.1:8081", SaveMethod, reader, binSize, ¶ms, &result, auth)
|
||||
err = Put(ctx, "127.0.0.1:18081", SaveMethod, reader, binSize, ¶ms, &result, auth)
|
||||
if err != nil {
|
||||
logError("method err:", err)
|
||||
return err
|
||||
@@ -200,10 +204,11 @@ func clientLoad() error {
|
||||
binBytes := make([]byte, 0)
|
||||
writer := bytes.NewBuffer(binBytes)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
timeout := time.Duration(5 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
err = Get(ctx, "127.0.0.1:8081", LoadMethod, writer, ¶ms, &result, auth)
|
||||
err = Get(ctx, "127.0.0.1:18081", LoadMethod, writer, ¶ms, &result, auth)
|
||||
if err != nil {
|
||||
logError("method err:", err)
|
||||
return err
|
||||
@@ -228,10 +233,11 @@ func testServ(quiet bool) error {
|
||||
SetAccessWriter(io.Discard)
|
||||
SetMessageWriter(io.Discard)
|
||||
}
|
||||
|
||||
serv := NewService()
|
||||
serv.Handler(HelloMethod, helloHandler)
|
||||
serv.Handler(SaveMethod, saveHandler)
|
||||
serv.Handler(LoadMethod, loadHandler)
|
||||
serv.Handle(HelloMethod, helloHandler)
|
||||
serv.Handle(SaveMethod, saveHandler)
|
||||
serv.Handle(LoadMethod, loadHandler)
|
||||
|
||||
serv.PreMiddleware(LogRequest)
|
||||
serv.PreMiddleware(auth)
|
||||
@@ -239,7 +245,7 @@ func testServ(quiet bool) error {
|
||||
serv.PostMiddleware(LogResponse)
|
||||
serv.PostMiddleware(LogAccess)
|
||||
|
||||
err = serv.Listen(":8081")
|
||||
err = serv.Listen(":18081")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -277,7 +283,8 @@ func helloHandler(content *Content) error {
|
||||
return err
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
timeout := time.Duration(5 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
err = content.ReadBin(ctx, io.Discard)
|
||||
@@ -308,7 +315,8 @@ func saveHandler(content *Content) error {
|
||||
bufferBytes := make([]byte, 0, 1024)
|
||||
binWriter := bytes.NewBuffer(bufferBytes)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
timeout := time.Duration(5 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
err = content.ReadBin(ctx, binWriter)
|
||||
@@ -336,7 +344,8 @@ func loadHandler(content *Content) error {
|
||||
return err
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
|
||||
timeout := time.Duration(5 * time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
|
||||
err = content.ReadBin(ctx, io.Discard)
|
||||
|
||||
19
logger.go
19
logger.go
@@ -10,30 +10,27 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"time"
|
||||
)
|
||||
|
||||
var messageWriter io.Writer = os.Stdout
|
||||
var accessWriter io.Writer = os.Stdout
|
||||
var (
|
||||
messageWriter io.Writer = os.Stdout
|
||||
accessWriter io.Writer = os.Stdout
|
||||
)
|
||||
|
||||
func logDebug(messages ...any) {
|
||||
stamp := time.Now().Format(time.RFC3339)
|
||||
fmt.Fprintln(messageWriter, stamp, "debug", messages)
|
||||
fmt.Fprintln(messageWriter, "debug:", messages)
|
||||
}
|
||||
|
||||
func logInfo(messages ...any) {
|
||||
stamp := time.Now().Format(time.RFC3339)
|
||||
fmt.Fprintln(messageWriter, stamp, "info", messages)
|
||||
fmt.Fprintln(messageWriter, "info:", messages)
|
||||
}
|
||||
|
||||
func logError(messages ...any) {
|
||||
stamp := time.Now().Format(time.RFC3339)
|
||||
fmt.Fprintln(messageWriter, stamp, "error", messages)
|
||||
fmt.Fprintln(messageWriter, "error:", messages)
|
||||
}
|
||||
|
||||
func logAccess(messages ...any) {
|
||||
stamp := time.Now().Format(time.RFC3339)
|
||||
fmt.Fprintln(accessWriter, stamp, "access", messages)
|
||||
fmt.Fprintln(accessWriter, "access:", messages)
|
||||
}
|
||||
|
||||
func SetAccessWriter(writer io.Writer) {
|
||||
|
||||
69
server.go
69
server.go
@@ -6,6 +6,7 @@ package dsrpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -19,15 +20,17 @@ import (
|
||||
type HandlerFunc = func(*Content) error
|
||||
|
||||
type Service struct {
|
||||
handlers map[string]HandlerFunc
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg *sync.WaitGroup
|
||||
preMw []HandlerFunc
|
||||
postMw []HandlerFunc
|
||||
keepalive bool
|
||||
kaTime time.Duration
|
||||
kaMtx sync.Mutex
|
||||
handlers map[string]HandlerFunc
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg *sync.WaitGroup
|
||||
preMw []HandlerFunc
|
||||
postMw []HandlerFunc
|
||||
keepalive bool
|
||||
kaTime time.Duration
|
||||
kaMtx sync.Mutex
|
||||
listener net.Listener
|
||||
tcpListener *net.TCPListener
|
||||
}
|
||||
|
||||
func NewService() *Service {
|
||||
@@ -52,7 +55,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
|
||||
}
|
||||
|
||||
@@ -77,14 +80,14 @@ func (svc *Service) Listen(address string) error {
|
||||
err = fmt.Errorf("unable to resolve adddress: %s", err)
|
||||
return err
|
||||
}
|
||||
listener, err := net.ListenTCP("tcp", addr)
|
||||
svc.tcpListener, err = net.ListenTCP("tcp", addr)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("unable to start listener: %s", err)
|
||||
return err
|
||||
}
|
||||
|
||||
for {
|
||||
conn, err := listener.AcceptTCP()
|
||||
conn, err := svc.tcpListener.AcceptTCP()
|
||||
if err != nil {
|
||||
logError("conn accept err:", err)
|
||||
}
|
||||
@@ -94,6 +97,33 @@ 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)
|
||||
|
||||
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)
|
||||
}
|
||||
return err
|
||||
@@ -109,6 +139,12 @@ func (svc *Service) Stop() error {
|
||||
var err error
|
||||
// Disable new connection
|
||||
logInfo("cancel rpc accept loop")
|
||||
if svc.listener != nil {
|
||||
svc.listener.Close()
|
||||
}
|
||||
if svc.tcpListener != nil {
|
||||
svc.tcpListener.Close()
|
||||
}
|
||||
svc.cancel()
|
||||
// Wait handlers
|
||||
logInfo("wait rpc handlers")
|
||||
@@ -116,9 +152,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 +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)
|
||||
|
||||
remoteAddr := conn.RemoteAddr().String()
|
||||
|
||||
Reference in New Issue
Block a user