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

disable default log timestamp

Browse Source
This commit is contained in:
Oleg Borodin
2023-05-18 13:34:54 +02:00
parent bc578cf25c
commit 5ea1af7fb1
3 changed files with 520 additions and 495 deletions
Download Patch File Download Diff File Expand all files Collapse all files

487
exec_test.go
View File

@@ -5,26 +5,26 @@
package dsrpc package dsrpc
import ( import (
"bytes" "bytes"
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"io" "io"
"math/rand" "math/rand"
"testing" "testing"
"time" "time"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
const HelloMethod string = "hello" const HelloMethod string = "hello"
type HelloParams struct { type HelloParams struct {
Message string `json:"message" msgpack:"message"` Message string `json:"message" msgpack:"message"`
} }
type HelloResult struct { type HelloResult struct {
Message string `json:"message" msgpack:"message"` Message string `json:"message" msgpack:"message"`
} }
const SaveMethod string = "save" const SaveMethod string = "save"
@@ -38,333 +38,342 @@ type LoadParams HelloParams
type LoadResult HelloResult type LoadResult HelloResult
func TestLocalExec(t *testing.T) { func TestLocalExec(t *testing.T) {
var err error var err error
params := HelloParams{} params := HelloParams{}
params.Message = "hello server!" params.Message = "hello server!"
result := HelloResult{} 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) resultJson, _ := json.Marshal(result)
logDebug("method result:", string(resultJson)) logDebug("method result:", string(resultJson))
} }
func TestLocalSave(t *testing.T) { func TestLocalSave(t *testing.T) {
var err error var err error
params := SaveParams{} params := SaveParams{}
params.Message = "save data!" params.Message = "save data!"
result := SaveResult{} result := SaveResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16 var binSize int64 = 16
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize) binBytes := make([]byte, binSize)
rand.Read(binBytes) rand.Read(binBytes)
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)
defer cancel() ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = LocalPut(ctx, SaveMethod, reader, binSize, &params, &result, auth, saveHandler) 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 := LoadParams{} params := LoadParams{}
params.Message = "load data!" params.Message = "load data!"
result := LoadResult{} 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)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second)) timeout := time.Duration(5 * time.Second)
defer cancel() ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = LocalGet(ctx, LoadMethod, writer, &params, &result, auth, loadHandler) 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)
} }
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) {
for pb.Next() { for pb.Next() {
clientSave() clientSave()
} }
} }
b.SetParallelism(2000) b.SetParallelism(2000)
b.RunParallel(pBench) b.RunParallel(pBench)
} }
func clientHello() error { func clientHello() error {
var err error var err error
params := HelloParams{} params := HelloParams{}
params.Message = "hello server!" params.Message = "hello server!"
result := HelloResult{} result := HelloResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16 var binSize int64 = 16
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
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)
defer cancel() ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = Exec(ctx, "127.0.0.1:8081", HelloMethod, &params, &result, auth) 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 := SaveParams{} params := SaveParams{}
params.Message = "save data!" params.Message = "save data!"
result := SaveResult{} result := SaveResult{}
auth := CreateAuth([]byte("qwert"), []byte("12345")) auth := CreateAuth([]byte("qwert"), []byte("12345"))
var binSize int64 = 16 var binSize int64 = 16
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize) binBytes := make([]byte, binSize)
rand.Read(binBytes) rand.Read(binBytes)
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)
defer cancel() ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = Put(ctx, "127.0.0.1:8081", SaveMethod, reader, binSize, &params, &result, auth) 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 := LoadParams{} params := LoadParams{}
params.Message = "load data!" params.Message = "load data!"
result := LoadResult{} 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)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second)) timeout := time.Duration(5 * time.Second)
defer cancel() ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = Get(ctx, "127.0.0.1:8081", LoadMethod, writer, &params, &result, auth) 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
} }
var testServRun bool = false var testServRun bool = false
func testServ(quiet bool) error { func testServ(quiet bool) error {
var err error var err error
if testServRun { if testServRun {
return err return err
} }
testServRun = true testServRun = true
if quiet { if quiet {
SetAccessWriter(io.Discard) SetAccessWriter(io.Discard)
SetMessageWriter(io.Discard) SetMessageWriter(io.Discard)
} }
serv := NewService()
serv.Handle(HelloMethod, helloHandler)
serv.Handle(SaveMethod, saveHandler)
serv.Handle(LoadMethod, loadHandler)
serv.PreMiddleware(LogRequest) serv := NewService()
serv.PreMiddleware(auth) serv.Handle(HelloMethod, helloHandler)
serv.Handle(SaveMethod, saveHandler)
serv.Handle(LoadMethod, loadHandler)
serv.PostMiddleware(LogResponse) serv.PreMiddleware(LogRequest)
serv.PostMiddleware(LogAccess) serv.PreMiddleware(auth)
err = serv.Listen(":8081") serv.PostMiddleware(LogResponse)
if err != nil { serv.PostMiddleware(LogAccess)
return err
} err = serv.Listen(":18081")
return err if err != nil {
return err
}
return err
} }
func auth(content *Content) error { func auth(content *Content) error {
var err error var err error
reqIdent := content.AuthIdent() reqIdent := content.AuthIdent()
reqSalt := content.AuthSalt() reqSalt := content.AuthSalt()
reqHash := content.AuthHash() reqHash := content.AuthHash()
ident := reqIdent ident := reqIdent
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)
if !ok { if !ok {
err = errors.New("auth ident or pass missmatch") err = errors.New("auth ident or pass missmatch")
content.SendError(err) content.SendError(err)
return err return err
} }
return err return err
} }
func helloHandler(content *Content) error { func helloHandler(content *Content) error {
var err error var err error
params := HelloParams{} params := HelloParams{}
err = content.BindParams(&params) err = content.BindParams(&params)
if err != nil { if err != nil {
return err return err
} }
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second)) timeout := time.Duration(5 * time.Second)
defer cancel() ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, io.Discard) err = content.ReadBin(ctx, io.Discard)
if err != nil { if err != nil {
content.SendError(err) content.SendError(err)
return err return err
} }
result := HelloResult{} result := HelloResult{}
result.Message = "hello, client!" result.Message = "hello, client!"
err = content.SendResult(result, 0) err = content.SendResult(result, 0)
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
func saveHandler(content *Content) error { func saveHandler(content *Content) error {
var err error var err error
params := SaveParams{} params := SaveParams{}
err = content.BindParams(&params) err = content.BindParams(&params)
if err != nil { if err != nil {
return err return err
} }
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)
defer cancel() ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, binWriter) err = content.ReadBin(ctx, binWriter)
if err != nil { if err != nil {
content.SendError(err) content.SendError(err)
return err return err
} }
result := SaveResult{} result := SaveResult{}
result.Message = "saved successfully!" result.Message = "saved successfully!"
err = content.SendResult(result, 0) err = content.SendResult(result, 0)
if err != nil { if err != nil {
return err return err
} }
return err return err
} }
func loadHandler(content *Content) error { func loadHandler(content *Content) error {
var err error var err error
params := SaveParams{} params := SaveParams{}
err = content.BindParams(&params) err = content.BindParams(&params)
if err != nil { if err != nil {
return err return err
} }
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second)) timeout := time.Duration(5 * time.Second)
defer cancel() ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
err = content.ReadBin(ctx, io.Discard) err = content.ReadBin(ctx, io.Discard)
if err != nil { if err != nil {
content.SendError(err) content.SendError(err)
return err return err
} }
var binSize int64 = 1024 var binSize int64 = 1024
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
binBytes := make([]byte, binSize) binBytes := make([]byte, binSize)
rand.Read(binBytes) rand.Read(binBytes)
binReader := bytes.NewReader(binBytes) binReader := bytes.NewReader(binBytes)
result := SaveResult{} 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(ctx, binReader, binWriter, binSize) _, err = CopyBytes(ctx, binReader, binWriter, binSize)
if err != nil { if err != nil {
return err return err
} }
return err return err
} }

28
logger.go
View File

@@ -13,26 +13,37 @@ import (
"time" "time"
) )
var messageWriter io.Writer = os.Stdout var (
var accessWriter io.Writer = os.Stdout messageWriter io.Writer = os.Stdout
accessWriter io.Writer = os.Stdout
logTimestamp bool = false
)
func getLogStamp() string {
var stamp string
if logTimestamp {
stamp = time.Now().Format(time.RFC3339)
}
return stamp
}
func logDebug(messages ...any) { func logDebug(messages ...any) {
stamp := time.Now().Format(time.RFC3339) stamp := getLogStamp()
fmt.Fprintln(messageWriter, stamp, "debug", messages) fmt.Fprintln(messageWriter, stamp, "debug", messages)
} }
func logInfo(messages ...any) { func logInfo(messages ...any) {
stamp := time.Now().Format(time.RFC3339) stamp := getLogStamp()
fmt.Fprintln(messageWriter, stamp, "info", messages) fmt.Fprintln(messageWriter, stamp, "info", messages)
} }
func logError(messages ...any) { func logError(messages ...any) {
stamp := time.Now().Format(time.RFC3339) stamp := getLogStamp()
fmt.Fprintln(messageWriter, stamp, "error", messages) fmt.Fprintln(messageWriter, stamp, "error", messages)
} }
func logAccess(messages ...any) { func logAccess(messages ...any) {
stamp := time.Now().Format(time.RFC3339) stamp := getLogStamp()
fmt.Fprintln(accessWriter, stamp, "access", messages) fmt.Fprintln(accessWriter, stamp, "access", messages)
} }
@@ -43,3 +54,8 @@ func SetAccessWriter(writer io.Writer) {
func SetMessageWriter(writer io.Writer) { func SetMessageWriter(writer io.Writer) {
messageWriter = writer messageWriter = writer
} }
func EnableLogTimestamp(enable bool) {
logTimestamp = enable
}

500
server.go
View File

@@ -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():
logInfo("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
} }