Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 498abc18b4 | |||
| 5ea1af7fb1 | |||
| bc578cf25c |
@@ -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)
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
47
exec_test.go
47
exec_test.go
@@ -67,7 +67,8 @@ func TestLocalSave(t *testing.T) {
|
|||||||
|
|
||||||
reader := bytes.NewReader(binBytes)
|
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()
|
defer cancel()
|
||||||
|
|
||||||
err = LocalPut(ctx, SaveMethod, reader, binSize, ¶ms, &result, auth, saveHandler)
|
err = LocalPut(ctx, SaveMethod, reader, binSize, ¶ms, &result, auth, saveHandler)
|
||||||
@@ -88,7 +89,8 @@ func TestLocalLoad(t *testing.T) {
|
|||||||
binBytes := make([]byte, 0)
|
binBytes := make([]byte, 0)
|
||||||
writer := bytes.NewBuffer(binBytes)
|
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()
|
defer cancel()
|
||||||
|
|
||||||
err = LocalGet(ctx, LoadMethod, writer, ¶ms, &result, auth, loadHandler)
|
err = LocalGet(ctx, LoadMethod, writer, ¶ms, &result, auth, loadHandler)
|
||||||
@@ -101,7 +103,7 @@ func TestLocalLoad(t *testing.T) {
|
|||||||
|
|
||||||
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)
|
||||||
@@ -109,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) {
|
||||||
@@ -148,10 +150,11 @@ func clientHello() error {
|
|||||||
binBytes := make([]byte, binSize)
|
binBytes := make([]byte, binSize)
|
||||||
rand.Read(binBytes)
|
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()
|
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 {
|
if err != nil {
|
||||||
logError("method err:", err)
|
logError("method err:", err)
|
||||||
return err
|
return err
|
||||||
@@ -176,10 +179,11 @@ func clientSave() error {
|
|||||||
|
|
||||||
reader := bytes.NewReader(binBytes)
|
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()
|
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 {
|
if err != nil {
|
||||||
logError("method err:", err)
|
logError("method err:", err)
|
||||||
return err
|
return err
|
||||||
@@ -200,10 +204,11 @@ func clientLoad() error {
|
|||||||
binBytes := make([]byte, 0)
|
binBytes := make([]byte, 0)
|
||||||
writer := bytes.NewBuffer(binBytes)
|
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()
|
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 {
|
if err != nil {
|
||||||
logError("method err:", err)
|
logError("method err:", err)
|
||||||
return err
|
return err
|
||||||
@@ -228,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)
|
||||||
@@ -239,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
|
||||||
}
|
}
|
||||||
@@ -277,7 +283,8 @@ func helloHandler(content *Content) error {
|
|||||||
return err
|
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()
|
defer cancel()
|
||||||
|
|
||||||
err = content.ReadBin(ctx, io.Discard)
|
err = content.ReadBin(ctx, io.Discard)
|
||||||
@@ -308,7 +315,8 @@ func saveHandler(content *Content) error {
|
|||||||
bufferBytes := make([]byte, 0, 1024)
|
bufferBytes := make([]byte, 0, 1024)
|
||||||
binWriter := bytes.NewBuffer(bufferBytes)
|
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()
|
defer cancel()
|
||||||
|
|
||||||
err = content.ReadBin(ctx, binWriter)
|
err = content.ReadBin(ctx, binWriter)
|
||||||
@@ -336,7 +344,8 @@ func loadHandler(content *Content) error {
|
|||||||
return err
|
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()
|
defer cancel()
|
||||||
|
|
||||||
err = content.ReadBin(ctx, io.Discard)
|
err = content.ReadBin(ctx, io.Discard)
|
||||||
|
|||||||
19
logger.go
19
logger.go
@@ -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) {
|
||||||
|
|||||||
500
server.go
500
server.go
@@ -5,344 +5,344 @@
|
|||||||
package dsrpc
|
package dsrpc
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"crypto/tls"
|
"crypto/tls"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net"
|
"net"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
encoder "encoding/json"
|
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
|
listener net.Listener
|
||||||
tcpListener *net.TCPListener
|
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 (svc *Service) PreMiddleware(mw HandlerFunc) {
|
func (svc *Service) PreMiddleware(mw HandlerFunc) {
|
||||||
svc.preMw = append(svc.preMw, mw)
|
svc.preMw = append(svc.preMw, mw)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) PostMiddleware(mw HandlerFunc) {
|
func (svc *Service) PostMiddleware(mw HandlerFunc) {
|
||||||
svc.postMw = append(svc.postMw, mw)
|
svc.postMw = append(svc.postMw, mw)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) Handle(method string, handler HandlerFunc) {
|
func (svc *Service) Handle(method string, handler HandlerFunc) {
|
||||||
svc.handlers[method] = handler
|
svc.handlers[method] = handler
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) SetKeepAlive(flag bool) {
|
func (svc *Service) SetKeepAlive(flag bool) {
|
||||||
svc.kaMtx.Lock()
|
svc.kaMtx.Lock()
|
||||||
defer svc.kaMtx.Unlock()
|
defer svc.kaMtx.Unlock()
|
||||||
svc.keepalive = true
|
svc.keepalive = true
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) SetKeepAlivePeriod(interval time.Duration) {
|
func (svc *Service) SetKeepAlivePeriod(interval time.Duration) {
|
||||||
svc.kaMtx.Lock()
|
svc.kaMtx.Lock()
|
||||||
defer svc.kaMtx.Unlock()
|
defer svc.kaMtx.Unlock()
|
||||||
svc.kaTime = interval
|
svc.kaTime = interval
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) Listen(address string) error {
|
func (svc *Service) Listen(address string) error {
|
||||||
var err error
|
var err error
|
||||||
logInfo("server listen:", address)
|
logInfo("server listen:", address)
|
||||||
|
|
||||||
addr, err := net.ResolveTCPAddr("tcp", address)
|
addr, err := net.ResolveTCPAddr("tcp", address)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
err = fmt.Errorf("unable to resolve adddress: %s", err)
|
err = fmt.Errorf("unable to resolve adddress: %s", err)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
svc.tcpListener, 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 := svc.tcpListener.AcceptTCP()
|
conn, err := svc.tcpListener.AcceptTCP()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logError("conn accept err:", err)
|
logError("conn accept err:", err)
|
||||||
}
|
}
|
||||||
select {
|
select {
|
||||||
case <-svc.ctx.Done():
|
case <-svc.ctx.Done():
|
||||||
return err
|
return err
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
svc.wg.Add(1)
|
svc.wg.Add(1)
|
||||||
go svc.handleTCPConn(conn, svc.wg)
|
go svc.handleTCPConn(conn, svc.wg)
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) ListenTLS(address string, tlsConfig *tls.Config) error {
|
func (svc *Service) ListenTLS(address string, tlsConfig *tls.Config) error {
|
||||||
var err error
|
var err error
|
||||||
logInfo("server listen:", address)
|
logInfo("server listen:", address)
|
||||||
|
|
||||||
svc.listener, err = tls.Listen("tcp", address, tlsConfig)
|
svc.listener, err = tls.Listen("tcp", address, tlsConfig)
|
||||||
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 := svc.listener.Accept()
|
conn, err := svc.listener.Accept()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logError("conn accept err:", err)
|
logError("conn accept err:", err)
|
||||||
}
|
}
|
||||||
select {
|
select {
|
||||||
case <-svc.ctx.Done():
|
case <-svc.ctx.Done():
|
||||||
logMessage("accept loop done")
|
logInfo("accept loop done")
|
||||||
return err
|
return err
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
svc.wg.Add(1)
|
svc.wg.Add(1)
|
||||||
go svc.handleConn(conn, svc.wg)
|
go svc.handleConn(conn, svc.wg)
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func notFound(content *Content) error {
|
func notFound(content *Content) error {
|
||||||
execErr := errors.New("method not found")
|
execErr := errors.New("method not found")
|
||||||
err := content.SendError(execErr)
|
err := content.SendError(execErr)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) Stop() error {
|
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 {
|
if svc.listener != nil {
|
||||||
svc.listener.Close()
|
svc.listener.Close()
|
||||||
}
|
}
|
||||||
if svc.tcpListener != nil {
|
if svc.tcpListener != nil {
|
||||||
svc.tcpListener.Close()
|
svc.tcpListener.Close()
|
||||||
}
|
}
|
||||||
svc.cancel()
|
svc.cancel()
|
||||||
// Wait handlers
|
// Wait handlers
|
||||||
logInfo("wait rpc handlers")
|
logInfo("wait rpc handlers")
|
||||||
svc.wg.Wait()
|
svc.wg.Wait()
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) handleTCPConn(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 {
|
||||||
err = fmt.Errorf("unable to set keepalive: %s", err)
|
err = fmt.Errorf("unable to set keepalive: %s", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if svc.kaTime > 0 {
|
if svc.kaTime > 0 {
|
||||||
err = conn.SetKeepAlivePeriod(svc.kaTime)
|
err = conn.SetKeepAlivePeriod(svc.kaTime)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
err = fmt.Errorf("unable to set keepalive period: %s", err)
|
err = fmt.Errorf("unable to set keepalive period: %s", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
svc.handleConn(conn, wg)
|
svc.handleConn(conn, wg)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) handleConn(conn net.Conn, wg *sync.WaitGroup) {
|
func (svc *Service) handleConn(conn net.Conn, wg *sync.WaitGroup) {
|
||||||
var err error
|
var err error
|
||||||
|
|
||||||
content := CreateContent(conn)
|
content := CreateContent(conn)
|
||||||
|
|
||||||
remoteAddr := conn.RemoteAddr().String()
|
remoteAddr := conn.RemoteAddr().String()
|
||||||
remoteHost, _, _ := net.SplitHostPort(remoteAddr)
|
remoteHost, _, _ := net.SplitHostPort(remoteAddr)
|
||||||
content.remoteHost = remoteHost
|
content.remoteHost = remoteHost
|
||||||
|
|
||||||
content.binReader = conn
|
content.binReader = conn
|
||||||
content.binWriter = io.Discard
|
content.binWriter = io.Discard
|
||||||
|
|
||||||
exitFunc := func() {
|
exitFunc := func() {
|
||||||
conn.Close()
|
conn.Close()
|
||||||
wg.Done()
|
wg.Done()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logError("conn handler err:", err)
|
logError("conn handler err:", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
defer exitFunc()
|
defer exitFunc()
|
||||||
|
|
||||||
recovFunc := func() {
|
recovFunc := func() {
|
||||||
panicMsg := recover()
|
panicMsg := recover()
|
||||||
if panicMsg != nil {
|
if panicMsg != nil {
|
||||||
logError("handler panic message:", panicMsg)
|
logError("handler panic message:", panicMsg)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
defer recovFunc()
|
defer recovFunc()
|
||||||
|
|
||||||
err = content.ReadRequest()
|
err = content.ReadRequest()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
err = err
|
err = err
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
err = content.BindMethod()
|
err = content.BindMethod()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
err = err
|
err = err
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
for _, mw := range svc.preMw {
|
for _, mw := range svc.preMw {
|
||||||
err = mw(content)
|
err = mw(content)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
err = err
|
err = err
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
err = svc.Route(content)
|
err = svc.Route(content)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
err = err
|
err = err
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
for _, mw := range svc.postMw {
|
for _, mw := range svc.postMw {
|
||||||
err = mw(content)
|
err = mw(content)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
err = err
|
err = err
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (svc *Service) Route(content *Content) error {
|
func (svc *Service) Route(content *Content) error {
|
||||||
handler, ok := svc.handlers[content.reqBlock.Method]
|
handler, ok := svc.handlers[content.reqBlock.Method]
|
||||||
if ok {
|
if ok {
|
||||||
return handler(content)
|
return handler(content)
|
||||||
}
|
}
|
||||||
return notFound(content)
|
return notFound(content)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) ReadRequest() error {
|
func (content *Content) ReadRequest() error {
|
||||||
var err error
|
var err error
|
||||||
|
|
||||||
content.reqPacket.header, err = ReadBytes(content.sockReader, headerSize)
|
content.reqPacket.header, err = ReadBytes(content.sockReader, headerSize)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
content.reqHeader, err = UnpackHeader(content.reqPacket.header)
|
content.reqHeader, err = UnpackHeader(content.reqPacket.header)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
rpcSize := content.reqHeader.rpcSize
|
rpcSize := content.reqHeader.rpcSize
|
||||||
content.reqPacket.rcpPayload, err = ReadBytes(content.sockReader, rpcSize)
|
content.reqPacket.rcpPayload, err = ReadBytes(content.sockReader, rpcSize)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) BinWriter() io.Writer {
|
func (content *Content) BinWriter() io.Writer {
|
||||||
return content.sockWriter
|
return content.sockWriter
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) BinReader() io.Reader {
|
func (content *Content) BinReader() io.Reader {
|
||||||
return content.sockReader
|
return content.sockReader
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) BinSize() int64 {
|
func (content *Content) BinSize() int64 {
|
||||||
return content.reqHeader.binSize
|
return content.reqHeader.binSize
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) ReadBin(ctx context.Context, writer io.Writer) error {
|
func (content *Content) ReadBin(ctx context.Context, writer io.Writer) error {
|
||||||
var err error
|
var err error
|
||||||
_, err = CopyBytes(ctx, content.sockReader, writer, content.reqHeader.binSize)
|
_, err = CopyBytes(ctx, content.sockReader, writer, content.reqHeader.binSize)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) BindMethod() error {
|
func (content *Content) BindMethod() error {
|
||||||
var err error
|
var err error
|
||||||
err = encoder.Unmarshal(content.reqPacket.rcpPayload, content.reqBlock)
|
err = encoder.Unmarshal(content.reqPacket.rcpPayload, content.reqBlock)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) BindParams(params any) error {
|
func (content *Content) BindParams(params any) error {
|
||||||
var err error
|
var err error
|
||||||
content.reqBlock.Params = params
|
content.reqBlock.Params = params
|
||||||
err = encoder.Unmarshal(content.reqPacket.rcpPayload, content.reqBlock)
|
err = encoder.Unmarshal(content.reqPacket.rcpPayload, content.reqBlock)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) SendResult(result any, binSize int64) error {
|
func (content *Content) SendResult(result any, binSize int64) error {
|
||||||
var err error
|
var err error
|
||||||
content.resBlock.Result = result
|
content.resBlock.Result = result
|
||||||
|
|
||||||
content.resPacket.rcpPayload, err = content.resBlock.Pack()
|
content.resPacket.rcpPayload, err = content.resBlock.Pack()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
content.resHeader.rpcSize = int64(len(content.resPacket.rcpPayload))
|
content.resHeader.rpcSize = int64(len(content.resPacket.rcpPayload))
|
||||||
content.resHeader.binSize = binSize
|
content.resHeader.binSize = binSize
|
||||||
|
|
||||||
content.resPacket.header, err = content.resHeader.Pack()
|
content.resPacket.header, err = content.resHeader.Pack()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
_, err = content.sockWriter.Write(content.resPacket.header)
|
_, err = content.sockWriter.Write(content.resPacket.header)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
_, err = content.sockWriter.Write(content.resPacket.rcpPayload)
|
_, err = content.sockWriter.Write(content.resPacket.rcpPayload)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (content *Content) SendError(execErr error) error {
|
func (content *Content) SendError(execErr error) error {
|
||||||
var err error
|
var err error
|
||||||
|
|
||||||
content.resBlock.Error = execErr.Error()
|
content.resBlock.Error = execErr.Error()
|
||||||
content.resBlock.Result = NewEmptyResult()
|
content.resBlock.Result = NewEmptyResult()
|
||||||
|
|
||||||
content.resPacket.rcpPayload, err = content.resBlock.Pack()
|
content.resPacket.rcpPayload, err = content.resBlock.Pack()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
content.resHeader.rpcSize = int64(len(content.resPacket.rcpPayload))
|
content.resHeader.rpcSize = int64(len(content.resPacket.rcpPayload))
|
||||||
content.resPacket.header, err = content.resHeader.Pack()
|
content.resPacket.header, err = content.resHeader.Pack()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
_, err = content.sockWriter.Write(content.resPacket.header)
|
_, err = content.sockWriter.Write(content.resPacket.header)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
_, err = content.sockWriter.Write(content.resPacket.rcpPayload)
|
_, err = content.sockWriter.Write(content.resPacket.rcpPayload)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user