4 Commits

Author SHA1 Message Date
498abc18b4 delete timestamp from logger 2023-05-18 13:44:39 +02:00
5ea1af7fb1 disable default log timestamp 2023-05-18 13:34:54 +02:00
bc578cf25c fixed mistakes 2023-05-18 12:56:37 +02:00
cb6d26a7cc added listener close() 2023-05-18 12:53:37 +02:00
5 changed files with 63 additions and 48 deletions

View File

@@ -44,7 +44,7 @@ func server() error {
serv := dsrpc.NewService() serv := dsrpc.NewService()
cont := NewController() cont := NewController()
serv.Handler(api.HelloMethod, cont.HelloHandler) serv.Handle(api.HelloMethod, cont.HelloHandler)
serv.PreMiddleware(dsrpc.LogRequest) serv.PreMiddleware(dsrpc.LogRequest)
serv.PostMiddleware(dsrpc.LogResponse) serv.PostMiddleware(dsrpc.LogResponse)
@@ -220,9 +220,9 @@ func sampleServ(quiet bool) error {
serv.PreMiddleware(authMiddleware) serv.PreMiddleware(authMiddleware)
serv.PreMiddleware(dsrpc.LogRequest) serv.PreMiddleware(dsrpc.LogRequest)
serv.Handler(HelloMethod, helloHandler) serv.Handle(HelloMethod, helloHandler)
serv.Handler(SaveMethod, saveHandler) serv.Handle(SaveMethod, saveHandler)
serv.Handler(LoadMethod, loadHandler) serv.Handle(LoadMethod, loadHandler)
serv.PostMiddleware(dsrpc.LogResponse) serv.PostMiddleware(dsrpc.LogResponse)
serv.PostMiddleware(dsrpc.LogAccess) serv.PostMiddleware(dsrpc.LogAccess)

View File

@@ -26,7 +26,7 @@ func server() error {
serv := dsrpc.NewService() serv := dsrpc.NewService()
cont := NewController() cont := NewController()
serv.Handler(api.HelloMethod, cont.HelloHandler) serv.Handle(api.HelloMethod, cont.HelloHandler)
serv.PreMiddleware(dsrpc.LogRequest) serv.PreMiddleware(dsrpc.LogRequest)
serv.PostMiddleware(dsrpc.LogResponse) serv.PostMiddleware(dsrpc.LogResponse)

View File

@@ -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, &params, &result, auth, saveHandler) err = LocalPut(ctx, SaveMethod, reader, binSize, &params, &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, &params, &result, auth, loadHandler) err = LocalGet(ctx, LoadMethod, writer, &params, &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, &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
@@ -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, &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
@@ -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, &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
@@ -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)

View File

@@ -10,30 +10,27 @@ import (
"fmt" "fmt"
"io" "io"
"os" "os"
"time"
) )
var messageWriter io.Writer = os.Stdout var (
var accessWriter io.Writer = os.Stdout messageWriter io.Writer = os.Stdout
accessWriter io.Writer = os.Stdout
)
func logDebug(messages ...any) { func logDebug(messages ...any) {
stamp := time.Now().Format(time.RFC3339) fmt.Fprintln(messageWriter, "debug:", messages)
fmt.Fprintln(messageWriter, stamp, "debug", messages)
} }
func logInfo(messages ...any) { func logInfo(messages ...any) {
stamp := time.Now().Format(time.RFC3339) fmt.Fprintln(messageWriter, "info:", messages)
fmt.Fprintln(messageWriter, stamp, "info", messages)
} }
func logError(messages ...any) { func logError(messages ...any) {
stamp := time.Now().Format(time.RFC3339) fmt.Fprintln(messageWriter, "error:", messages)
fmt.Fprintln(messageWriter, stamp, "error", messages)
} }
func logAccess(messages ...any) { func logAccess(messages ...any) {
stamp := time.Now().Format(time.RFC3339) fmt.Fprintln(accessWriter, "access:", messages)
fmt.Fprintln(accessWriter, stamp, "access", messages)
} }
func SetAccessWriter(writer io.Writer) { func SetAccessWriter(writer io.Writer) {

View File

@@ -20,15 +20,17 @@ import (
type HandlerFunc = func(*Content) error type HandlerFunc = func(*Content) error
type Service struct { type Service struct {
handlers map[string]HandlerFunc handlers map[string]HandlerFunc
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
wg *sync.WaitGroup wg *sync.WaitGroup
preMw []HandlerFunc preMw []HandlerFunc
postMw []HandlerFunc postMw []HandlerFunc
keepalive bool keepalive bool
kaTime time.Duration kaTime time.Duration
kaMtx sync.Mutex kaMtx sync.Mutex
listener net.Listener
tcpListener *net.TCPListener
} }
func NewService() *Service { func NewService() *Service {
@@ -78,14 +80,14 @@ func (svc *Service) Listen(address string) error {
err = fmt.Errorf("unable to resolve adddress: %s", err) err = fmt.Errorf("unable to resolve adddress: %s", err)
return err return err
} }
listener, err := net.ListenTCP("tcp", addr) svc.tcpListener, err = net.ListenTCP("tcp", addr)
if err != nil { if err != nil {
err = fmt.Errorf("unable to start listener: %s", err) err = fmt.Errorf("unable to start listener: %s", err)
return err return err
} }
for { for {
conn, err := listener.AcceptTCP() conn, err := svc.tcpListener.AcceptTCP()
if err != nil { if err != nil {
logError("conn accept err:", err) logError("conn accept err:", err)
} }
@@ -104,19 +106,20 @@ func (svc *Service) ListenTLS(address string, tlsConfig *tls.Config) error {
var err error var err error
logInfo("server listen:", address) logInfo("server listen:", address)
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 := 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")
return err return err
default: default:
} }
@@ -136,6 +139,12 @@ func (svc *Service) Stop() error {
var err error var err error
// Disable new connection // Disable new connection
logInfo("cancel rpc accept loop") logInfo("cancel rpc accept loop")
if svc.listener != nil {
svc.listener.Close()
}
if svc.tcpListener != nil {
svc.tcpListener.Close()
}
svc.cancel() svc.cancel()
// Wait handlers // Wait handlers
logInfo("wait rpc handlers") logInfo("wait rpc handlers")