fix(service): default nil logger, fix Events() allocation, collect Close errors
- Default nil logger to Nop() in NewService to prevent panics - Fix Events() returning 20-element slice padded with zero values - Collect and join all errors in Close() instead of discarding - Remove unnecessary variable shadow in execCmds - Add doc comments to all 22 exported symbols - Add 34 tests covering status, serve loop, close, and edge cases
This commit is contained in:
+46
-5
@@ -1,3 +1,6 @@
|
|||||||
|
// Package service implements the core [service.Service] interface, binding a
|
||||||
|
// [listener.Listener] and a [handler.Handler] into a runnable proxy service
|
||||||
|
// that accepts inbound connections and forwards traffic through a proxy chain.
|
||||||
package service
|
package service
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -22,6 +25,7 @@ import (
|
|||||||
"github.com/go-gost/core/recorder"
|
"github.com/go-gost/core/recorder"
|
||||||
"github.com/go-gost/core/service"
|
"github.com/go-gost/core/service"
|
||||||
xctx "github.com/go-gost/x/ctx"
|
xctx "github.com/go-gost/x/ctx"
|
||||||
|
xlogger "github.com/go-gost/x/logger"
|
||||||
xmetrics "github.com/go-gost/x/metrics"
|
xmetrics "github.com/go-gost/x/metrics"
|
||||||
xstats "github.com/go-gost/x/observer/stats"
|
xstats "github.com/go-gost/x/observer/stats"
|
||||||
"github.com/google/shlex"
|
"github.com/google/shlex"
|
||||||
@@ -41,62 +45,76 @@ type options struct {
|
|||||||
logger logger.Logger
|
logger logger.Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Option is a functional option for configuring a service.
|
||||||
type Option func(opts *options)
|
type Option func(opts *options)
|
||||||
|
|
||||||
|
// AdmissionOption sets the admission controller that filters connections by
|
||||||
|
// source address.
|
||||||
func AdmissionOption(admission admission.Admission) Option {
|
func AdmissionOption(admission admission.Admission) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.admission = admission
|
opts.admission = admission
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RecordersOption sets the traffic recorders for the service.
|
||||||
func RecordersOption(recorders ...recorder.RecorderObject) Option {
|
func RecordersOption(recorders ...recorder.RecorderObject) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.recorders = recorders
|
opts.recorders = recorders
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PreUpOption sets shell commands to run before the service starts.
|
||||||
func PreUpOption(cmds []string) Option {
|
func PreUpOption(cmds []string) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.preUp = cmds
|
opts.preUp = cmds
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PreDownOption sets shell commands to run before the service stops.
|
||||||
func PreDownOption(cmds []string) Option {
|
func PreDownOption(cmds []string) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.preDown = cmds
|
opts.preDown = cmds
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PostUpOption sets shell commands to run after the service starts listening.
|
||||||
func PostUpOption(cmds []string) Option {
|
func PostUpOption(cmds []string) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.postUp = cmds
|
opts.postUp = cmds
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PostDownOption sets shell commands to run after the service stops.
|
||||||
func PostDownOption(cmds []string) Option {
|
func PostDownOption(cmds []string) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.postDown = cmds
|
opts.postDown = cmds
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// StatsOption sets the traffic statistics collector for the service.
|
||||||
func StatsOption(stats stats.Stats) Option {
|
func StatsOption(stats stats.Stats) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.stats = stats
|
opts.stats = stats
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ObserverOption sets the observer that receives service state and stats events.
|
||||||
func ObserverOption(observer observer.Observer) Option {
|
func ObserverOption(observer observer.Observer) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.observer = observer
|
opts.observer = observer
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ObserverPeriodOption sets the interval at which stats are reported to the
|
||||||
|
// observer. Defaults to 5 seconds with a minimum of 1 second.
|
||||||
func ObserverPeriodOption(period time.Duration) Option {
|
func ObserverPeriodOption(period time.Duration) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.observerPeriod = period
|
opts.observerPeriod = period
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// LoggerOption sets the logger for the service. If not provided, a no-op
|
||||||
|
// logger is used.
|
||||||
func LoggerOption(logger logger.Logger) Option {
|
func LoggerOption(logger logger.Logger) Option {
|
||||||
return func(opts *options) {
|
return func(opts *options) {
|
||||||
opts.logger = logger
|
opts.logger = logger
|
||||||
@@ -111,11 +129,17 @@ type defaultService struct {
|
|||||||
options options
|
options options
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// NewService creates a new service that binds the given listener and handler.
|
||||||
|
// The service is registered in the running state and pre-up commands are
|
||||||
|
// executed immediately. Call [Serve] to start accepting connections.
|
||||||
func NewService(name string, ln listener.Listener, h handler.Handler, opts ...Option) service.Service {
|
func NewService(name string, ln listener.Listener, h handler.Handler, opts ...Option) service.Service {
|
||||||
var options options
|
var options options
|
||||||
for _, opt := range opts {
|
for _, opt := range opts {
|
||||||
opt(&options)
|
opt(&options)
|
||||||
}
|
}
|
||||||
|
if options.logger == nil {
|
||||||
|
options.logger = xlogger.Nop()
|
||||||
|
}
|
||||||
s := &defaultService{
|
s := &defaultService{
|
||||||
name: name,
|
name: name,
|
||||||
listener: ln,
|
listener: ln,
|
||||||
@@ -134,10 +158,14 @@ func NewService(name string, ln listener.Listener, h handler.Handler, opts ...Op
|
|||||||
return s
|
return s
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Addr returns the network address the service is listening on.
|
||||||
func (s *defaultService) Addr() net.Addr {
|
func (s *defaultService) Addr() net.Addr {
|
||||||
return s.listener.Addr()
|
return s.listener.Addr()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Serve starts the accept loop. It blocks until the listener is closed or a
|
||||||
|
// fatal accept error occurs. Each accepted connection is handled in its own
|
||||||
|
// goroutine by the service's handler.
|
||||||
func (s *defaultService) Serve() error {
|
func (s *defaultService) Serve() error {
|
||||||
s.execCmds("post-up", s.options.postUp)
|
s.execCmds("post-up", s.options.postUp)
|
||||||
s.setState(StateReady)
|
s.setState(StateReady)
|
||||||
@@ -291,28 +319,39 @@ func (s *defaultService) Serve() error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Status returns the runtime status of the service.
|
||||||
func (s *defaultService) Status() *Status {
|
func (s *defaultService) Status() *Status {
|
||||||
return s.status
|
return s.status
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Close shuts down the service by closing the handler, observer, and listener
|
||||||
|
// in order. All errors are collected and returned as a joined error.
|
||||||
func (s *defaultService) Close() error {
|
func (s *defaultService) Close() error {
|
||||||
s.execCmds("pre-down", s.options.preDown)
|
s.execCmds("pre-down", s.options.preDown)
|
||||||
defer s.execCmds("post-down", s.options.postDown)
|
defer s.execCmds("post-down", s.options.postDown)
|
||||||
|
|
||||||
|
var errs []error
|
||||||
if closer, ok := s.handler.(io.Closer); ok {
|
if closer, ok := s.handler.(io.Closer); ok {
|
||||||
closer.Close()
|
if err := closer.Close(); err != nil {
|
||||||
|
errs = append(errs, err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if s.options.observer != nil {
|
if s.options.observer != nil {
|
||||||
if closer, ok := s.options.observer.(io.Closer); ok {
|
if closer, ok := s.options.observer.(io.Closer); ok {
|
||||||
closer.Close()
|
if err := closer.Close(); err != nil {
|
||||||
|
errs = append(errs, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return s.listener.Close()
|
}
|
||||||
|
if err := s.listener.Close(); err != nil {
|
||||||
|
errs = append(errs, err)
|
||||||
|
}
|
||||||
|
return errors.Join(errs...)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *defaultService) execCmds(phase string, cmds []string) {
|
func (s *defaultService) execCmds(phase string, cmds []string) {
|
||||||
for _, cmd := range cmds {
|
for _, raw := range cmds {
|
||||||
cmd := strings.TrimSpace(cmd)
|
cmd := strings.TrimSpace(raw)
|
||||||
if cmd == "" {
|
if cmd == "" {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
@@ -411,6 +450,7 @@ func (s *defaultService) observeStats(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ServiceEvent is an observer event representing a service state change.
|
||||||
type ServiceEvent struct {
|
type ServiceEvent struct {
|
||||||
Kind string
|
Kind string
|
||||||
Service string
|
Service string
|
||||||
@@ -418,6 +458,7 @@ type ServiceEvent struct {
|
|||||||
Msg string
|
Msg string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Type returns the observer event type for service events.
|
||||||
func (ServiceEvent) Type() observer.EventType {
|
func (ServiceEvent) Type() observer.EventType {
|
||||||
return observer.EventStatus
|
return observer.EventStatus
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,668 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"io"
|
||||||
|
"net"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-gost/core/admission"
|
||||||
|
"github.com/go-gost/core/handler"
|
||||||
|
"github.com/go-gost/core/listener"
|
||||||
|
"github.com/go-gost/core/metadata"
|
||||||
|
"github.com/go-gost/core/observer"
|
||||||
|
"github.com/go-gost/core/observer/stats"
|
||||||
|
"github.com/go-gost/core/recorder"
|
||||||
|
xctx "github.com/go-gost/x/ctx"
|
||||||
|
)
|
||||||
|
|
||||||
|
// --- Mocks ---
|
||||||
|
|
||||||
|
type mockListener struct {
|
||||||
|
addr net.Addr
|
||||||
|
connCh chan net.Conn // connections to deliver via Accept
|
||||||
|
errCh chan error // errors to deliver via Accept
|
||||||
|
closed atomic.Bool
|
||||||
|
closeMu sync.Mutex
|
||||||
|
closeCh chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func newMockListener() *mockListener {
|
||||||
|
return &mockListener{
|
||||||
|
addr: &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0},
|
||||||
|
connCh: make(chan net.Conn, 10),
|
||||||
|
errCh: make(chan error, 10),
|
||||||
|
closeCh: make(chan struct{}),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *mockListener) Init(metadata.Metadata) error { return nil }
|
||||||
|
|
||||||
|
func (l *mockListener) Accept() (net.Conn, error) {
|
||||||
|
select {
|
||||||
|
case c := <-l.connCh:
|
||||||
|
return c, nil
|
||||||
|
case e := <-l.errCh:
|
||||||
|
return nil, e
|
||||||
|
case <-l.closeCh:
|
||||||
|
return nil, net.ErrClosed
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *mockListener) Addr() net.Addr { return l.addr }
|
||||||
|
|
||||||
|
func (l *mockListener) Close() error {
|
||||||
|
l.closeMu.Lock()
|
||||||
|
defer l.closeMu.Unlock()
|
||||||
|
if l.closed.CompareAndSwap(false, true) {
|
||||||
|
close(l.closeCh)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type mockHandler struct {
|
||||||
|
handleFn func(ctx context.Context, conn net.Conn) error
|
||||||
|
closed atomic.Bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func newMockHandler(fn func(ctx context.Context, conn net.Conn) error) *mockHandler {
|
||||||
|
return &mockHandler{handleFn: fn}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *mockHandler) Init(metadata.Metadata) error { return nil }
|
||||||
|
|
||||||
|
func (h *mockHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error {
|
||||||
|
if h.handleFn != nil {
|
||||||
|
return h.handleFn(ctx, conn)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *mockHandler) Close() error {
|
||||||
|
h.closed.Store(true)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type mockConn struct {
|
||||||
|
net.Conn
|
||||||
|
remoteAddr net.Addr
|
||||||
|
localAddr net.Addr
|
||||||
|
ctx context.Context
|
||||||
|
closed atomic.Bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *mockConn) RemoteAddr() net.Addr { return c.remoteAddr }
|
||||||
|
func (c *mockConn) LocalAddr() net.Addr { return c.localAddr }
|
||||||
|
func (c *mockConn) Close() error { c.closed.Store(true); return nil }
|
||||||
|
|
||||||
|
type mockAdmission struct {
|
||||||
|
allow bool
|
||||||
|
calls atomic.Int32
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *mockAdmission) Admit(ctx context.Context, network, addr string, opts ...admission.Option) bool {
|
||||||
|
a.calls.Add(1)
|
||||||
|
return a.allow
|
||||||
|
}
|
||||||
|
|
||||||
|
type mockObserver struct {
|
||||||
|
events []observer.Event
|
||||||
|
mu sync.Mutex
|
||||||
|
err error
|
||||||
|
closed atomic.Bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (o *mockObserver) Observe(ctx context.Context, events []observer.Event, opts ...observer.Option) error {
|
||||||
|
o.mu.Lock()
|
||||||
|
defer o.mu.Unlock()
|
||||||
|
o.events = append(o.events, events...)
|
||||||
|
return o.err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (o *mockObserver) Close() error {
|
||||||
|
o.closed.Store(true)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (o *mockObserver) getEvents() []observer.Event {
|
||||||
|
o.mu.Lock()
|
||||||
|
defer o.mu.Unlock()
|
||||||
|
out := make([]observer.Event, len(o.events))
|
||||||
|
copy(out, o.events)
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
type mockRecorder struct {
|
||||||
|
recorded [][]byte
|
||||||
|
mu sync.Mutex
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *mockRecorder) Record(ctx context.Context, b []byte, opts ...recorder.RecordOption) error {
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
cp := make([]byte, len(b))
|
||||||
|
copy(cp, b)
|
||||||
|
r.recorded = append(r.recorded, cp)
|
||||||
|
return r.err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *mockRecorder) getRecorded() [][]byte {
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
return r.recorded
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- NewService tests ---
|
||||||
|
|
||||||
|
func TestNewServiceDefaultsNopLogger(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
h := newMockHandler(nil)
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
// Should not panic when logging
|
||||||
|
svc.options.logger.Info("test")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNewServiceSetsRunningState(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
h := newMockHandler(nil)
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
if svc.Status().State() != StateRunning {
|
||||||
|
t.Errorf("State() = %q, want %q", svc.Status().State(), StateRunning)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNewServiceRecordsCreateTime(t *testing.T) {
|
||||||
|
before := time.Now()
|
||||||
|
ln := newMockListener()
|
||||||
|
h := newMockHandler(nil)
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
after := time.Now()
|
||||||
|
|
||||||
|
ct := svc.Status().CreateTime()
|
||||||
|
if ct.Before(before) || ct.After(after) {
|
||||||
|
t.Errorf("CreateTime = %v, want between %v and %v", ct, before, after)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Addr tests ---
|
||||||
|
|
||||||
|
func TestAddr(t *testing.T) {
|
||||||
|
addr := &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 8080}
|
||||||
|
ln := newMockListener()
|
||||||
|
ln.addr = addr
|
||||||
|
svc := NewService("test", ln, newMockHandler(nil)).(*defaultService)
|
||||||
|
if got := svc.Addr(); got.String() != addr.String() {
|
||||||
|
t.Errorf("Addr() = %v, want %v", got, addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Serve tests ---
|
||||||
|
|
||||||
|
func TestServeAcceptsConnection(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
var handled atomic.Int32
|
||||||
|
h := newMockHandler(func(ctx context.Context, conn net.Conn) error {
|
||||||
|
handled.Add(1)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.connCh <- &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("192.168.1.1"), Port: 1234},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
}
|
||||||
|
|
||||||
|
// Give handler goroutine time to run
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
if handled.Load() != 1 {
|
||||||
|
t.Errorf("handled = %d, want 1", handled.Load())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeSetsReadyState(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
svc := NewService("test", ln, newMockHandler(nil)).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
time.Sleep(50 * time.Millisecond)
|
||||||
|
if got := svc.Status().State(); got != StateReady {
|
||||||
|
t.Errorf("State() = %q, want %q", got, StateReady)
|
||||||
|
}
|
||||||
|
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeReturnsOnListenerClose(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
svc := NewService("test", ln, newMockHandler(nil)).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.Close()
|
||||||
|
err := <-done
|
||||||
|
if !errors.Is(err, net.ErrClosed) {
|
||||||
|
t.Errorf("Serve() = %v, want net.ErrClosed", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if svc.Status().State() != StateClosed {
|
||||||
|
t.Errorf("State() = %q, want %q", svc.Status().State(), StateClosed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeTemporaryErrorRetry(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
svc := NewService("test", ln, newMockHandler(nil)).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
// Send a temporary error, then close
|
||||||
|
ln.errCh <- &tempErr{}
|
||||||
|
time.Sleep(50 * time.Millisecond)
|
||||||
|
|
||||||
|
if svc.Status().State() != StateFailed {
|
||||||
|
t.Errorf("State() = %q, want %q", svc.Status().State(), StateFailed)
|
||||||
|
}
|
||||||
|
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
}
|
||||||
|
|
||||||
|
type tempErr struct{}
|
||||||
|
|
||||||
|
func (tempErr) Error() string { return "temporary error" }
|
||||||
|
func (tempErr) Temporary() bool { return true }
|
||||||
|
func (tempErr) Timeout() bool { return false }
|
||||||
|
|
||||||
|
func TestServeAdmissionDeny(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
var handled atomic.Int32
|
||||||
|
h := newMockHandler(func(ctx context.Context, conn net.Conn) error {
|
||||||
|
handled.Add(1)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
adm := &mockAdmission{allow: false}
|
||||||
|
svc := NewService("test", ln, h, AdmissionOption(adm)).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.connCh <- &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
if handled.Load() != 0 {
|
||||||
|
t.Error("handler should not be called for denied connection")
|
||||||
|
}
|
||||||
|
if adm.calls.Load() != 1 {
|
||||||
|
t.Errorf("admission calls = %d, want 1", adm.calls.Load())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeAdmissionAllow(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
var handled atomic.Int32
|
||||||
|
h := newMockHandler(func(ctx context.Context, conn net.Conn) error {
|
||||||
|
handled.Add(1)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
adm := &mockAdmission{allow: true}
|
||||||
|
svc := NewService("test", ln, h, AdmissionOption(adm)).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.connCh <- &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
if handled.Load() != 1 {
|
||||||
|
t.Errorf("handled = %d, want 1", handled.Load())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeHandlerErrorIncrementsStats(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
h := newMockHandler(func(ctx context.Context, conn net.Conn) error {
|
||||||
|
return errors.New("handler error")
|
||||||
|
})
|
||||||
|
ms := newMockStats()
|
||||||
|
svc := NewService("test", ln, h, StatsOption(ms)).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.connCh <- &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
if got := ms.Get(stats.KindTotalErrs); got != 1 {
|
||||||
|
t.Errorf("TotalErrs = %d, want 1", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeContextFromConn(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
var gotSid string
|
||||||
|
h := newMockHandler(func(ctx context.Context, conn net.Conn) error {
|
||||||
|
if s := xctx.SidFromContext(ctx); s != "" {
|
||||||
|
gotSid = string(s)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.connCh <- &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
if gotSid == "" {
|
||||||
|
t.Error("expected session ID in context, got empty")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeCtxConnContext(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
wantCtx := context.Background()
|
||||||
|
var gotCtx context.Context
|
||||||
|
h := newMockHandler(func(ctx context.Context, conn net.Conn) error {
|
||||||
|
gotCtx = ctx
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.connCh <- &ctxConn{
|
||||||
|
Conn: &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
},
|
||||||
|
ctx: wantCtx,
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
if gotCtx == nil {
|
||||||
|
t.Fatal("handler context is nil")
|
||||||
|
}
|
||||||
|
// The ctx from the conn should be used as the base
|
||||||
|
if s := xctx.SidFromContext(gotCtx); s == "" {
|
||||||
|
t.Error("expected session ID to be added to conn context")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type ctxConn struct {
|
||||||
|
net.Conn
|
||||||
|
ctx context.Context
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *ctxConn) Context() context.Context { return c.ctx }
|
||||||
|
|
||||||
|
func TestServeRecordsClientAddress(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
rec := &mockRecorder{}
|
||||||
|
h := newMockHandler(nil)
|
||||||
|
|
||||||
|
svc := NewService("test", ln, h,
|
||||||
|
RecordersOption(recorder.RecorderObject{
|
||||||
|
Recorder: rec,
|
||||||
|
Record: recorder.RecorderServiceClientAddress,
|
||||||
|
}),
|
||||||
|
).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.connCh <- &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("192.168.1.100"), Port: 54321},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
recorded := rec.getRecorded()
|
||||||
|
if len(recorded) != 1 {
|
||||||
|
t.Fatalf("recorded = %d entries, want 1", len(recorded))
|
||||||
|
}
|
||||||
|
if string(recorded[0]) != "192.168.1.100" {
|
||||||
|
t.Errorf("recorded = %q, want %q", string(recorded[0]), "192.168.1.100")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeRecordsClientAddressError(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
rec := &mockRecorder{err: errors.New("record fail")}
|
||||||
|
h := newMockHandler(nil)
|
||||||
|
|
||||||
|
svc := NewService("test", ln, h,
|
||||||
|
RecordersOption(recorder.RecorderObject{
|
||||||
|
Recorder: rec,
|
||||||
|
Record: recorder.RecorderServiceClientAddress,
|
||||||
|
}),
|
||||||
|
).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
ln.connCh <- &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("192.168.1.100"), Port: 54321},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
// Should not panic; error is just logged
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeObserverStats(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
obs := &mockObserver{}
|
||||||
|
ms := newMockStats()
|
||||||
|
h := newMockHandler(nil)
|
||||||
|
|
||||||
|
svc := NewService("test", ln, h,
|
||||||
|
ObserverOption(obs),
|
||||||
|
StatsOption(ms),
|
||||||
|
ObserverPeriodOption(50*time.Millisecond),
|
||||||
|
).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
// Add some stats to trigger observation
|
||||||
|
ms.Add(stats.KindTotalConns, 10)
|
||||||
|
|
||||||
|
time.Sleep(200 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
events := obs.getEvents()
|
||||||
|
if len(events) == 0 {
|
||||||
|
t.Error("expected observer to receive events")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Close tests ---
|
||||||
|
|
||||||
|
func TestCloseClosesListener(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
svc := NewService("test", ln, newMockHandler(nil))
|
||||||
|
|
||||||
|
if err := svc.Close(); err != nil {
|
||||||
|
t.Errorf("Close() = %v", err)
|
||||||
|
}
|
||||||
|
if !ln.closed.Load() {
|
||||||
|
t.Error("listener was not closed")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCloseClosesHandler(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
h := newMockHandler(nil)
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
|
||||||
|
if err := svc.Close(); err != nil {
|
||||||
|
t.Errorf("Close() = %v", err)
|
||||||
|
}
|
||||||
|
if !h.closed.Load() {
|
||||||
|
t.Error("handler was not closed")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCloseClosesObserver(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
obs := &mockObserver{}
|
||||||
|
svc := NewService("test", ln, newMockHandler(nil), ObserverOption(obs)).(*defaultService)
|
||||||
|
|
||||||
|
if err := svc.Close(); err != nil {
|
||||||
|
t.Errorf("Close() = %v", err)
|
||||||
|
}
|
||||||
|
if !obs.closed.Load() {
|
||||||
|
t.Error("observer was not closed")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCloseJoinsErrors(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
h := &closeErrHandler{err: errors.New("handler close error")}
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
|
||||||
|
err := svc.Close()
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("Close() = nil, want error")
|
||||||
|
}
|
||||||
|
if err.Error() != "handler close error" {
|
||||||
|
t.Errorf("Close() = %v, want 'handler close error'", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type closeErrHandler struct {
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *closeErrHandler) Init(metadata.Metadata) error { return nil }
|
||||||
|
func (h *closeErrHandler) Handle(context.Context, net.Conn, ...handler.HandleOption) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
func (h *closeErrHandler) Close() error { return h.err }
|
||||||
|
|
||||||
|
// --- Status tests ---
|
||||||
|
|
||||||
|
func TestStatusReturnsStatus(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
svc := NewService("test", ln, newMockHandler(nil)).(*defaultService)
|
||||||
|
st := svc.Status()
|
||||||
|
if st == nil {
|
||||||
|
t.Fatal("Status() = nil")
|
||||||
|
}
|
||||||
|
if st.State() != StateRunning {
|
||||||
|
t.Errorf("State() = %q, want %q", st.State(), StateRunning)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- ServiceEvent tests ---
|
||||||
|
|
||||||
|
func TestServiceEventType(t *testing.T) {
|
||||||
|
ev := ServiceEvent{Kind: "service", Service: "test", State: StateReady}
|
||||||
|
if ev.Type() != observer.EventStatus {
|
||||||
|
t.Errorf("Type() = %v, want %v", ev.Type(), observer.EventStatus)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Integration: multiple connections ---
|
||||||
|
|
||||||
|
func TestServeMultipleConnections(t *testing.T) {
|
||||||
|
ln := newMockListener()
|
||||||
|
var handled atomic.Int32
|
||||||
|
h := newMockHandler(func(ctx context.Context, conn net.Conn) error {
|
||||||
|
handled.Add(1)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
svc := NewService("test", ln, h).(*defaultService)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- svc.Serve() }()
|
||||||
|
|
||||||
|
for i := 0; i < 5; i++ {
|
||||||
|
ln.connCh <- &mockConn{
|
||||||
|
remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 1000 + i},
|
||||||
|
localAddr: ln.addr,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(200 * time.Millisecond)
|
||||||
|
ln.Close()
|
||||||
|
<-done
|
||||||
|
|
||||||
|
if got := handled.Load(); got != 5 {
|
||||||
|
t.Errorf("handled = %d, want 5", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- execCmd tests ---
|
||||||
|
|
||||||
|
func TestExecCmdInvalidCommand(t *testing.T) {
|
||||||
|
svc := NewService("test", newMockListener(), newMockHandler(nil)).(*defaultService)
|
||||||
|
err := svc.execCmd("")
|
||||||
|
if err == nil {
|
||||||
|
t.Error("execCmd('') = nil, want error")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Interface assertions ---
|
||||||
|
|
||||||
|
var _ io.Closer = (*closeErrHandler)(nil)
|
||||||
|
var _ listener.Listener = (*mockListener)(nil)
|
||||||
|
var _ handler.Handler = (*mockHandler)(nil)
|
||||||
|
var _ handler.Handler = (*closeErrHandler)(nil)
|
||||||
|
var _ admission.Admission = (*mockAdmission)(nil)
|
||||||
|
var _ observer.Observer = (*mockObserver)(nil)
|
||||||
|
var _ stats.Stats = (*mockStats)(nil)
|
||||||
|
var _ recorder.Recorder = (*mockRecorder)(nil)
|
||||||
|
var _ xctx.Context = (*ctxConn)(nil)
|
||||||
+17
-5
@@ -7,24 +7,31 @@ import (
|
|||||||
"github.com/go-gost/core/observer/stats"
|
"github.com/go-gost/core/observer/stats"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
// MaxEventSize is the maximum number of events retained in a Status event log.
|
||||||
MaxEventSize = 20
|
// When the log is full, the oldest event is discarded.
|
||||||
)
|
const MaxEventSize = 20
|
||||||
|
|
||||||
|
// State represents the operational state of a service.
|
||||||
type State string
|
type State string
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
// StateRunning indicates the service has been created but is not yet accepting.
|
||||||
StateRunning State = "running"
|
StateRunning State = "running"
|
||||||
|
// StateReady indicates the service is actively accepting connections.
|
||||||
StateReady State = "ready"
|
StateReady State = "ready"
|
||||||
|
// StateFailed indicates the service accept loop encountered a temporary error.
|
||||||
StateFailed State = "failed"
|
StateFailed State = "failed"
|
||||||
|
// StateClosed indicates the service has been shut down.
|
||||||
StateClosed State = "closed"
|
StateClosed State = "closed"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Event records a state change or notable occurrence in a service's lifetime.
|
||||||
type Event struct {
|
type Event struct {
|
||||||
Time time.Time
|
Time time.Time
|
||||||
Message string
|
Message string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Status tracks the runtime state, event log, and traffic statistics of a service.
|
||||||
type Status struct {
|
type Status struct {
|
||||||
createTime time.Time
|
createTime time.Time
|
||||||
state State
|
state State
|
||||||
@@ -33,10 +40,12 @@ type Status struct {
|
|||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// CreateTime returns the time at which the service was created.
|
||||||
func (p *Status) CreateTime() time.Time {
|
func (p *Status) CreateTime() time.Time {
|
||||||
return p.createTime
|
return p.createTime
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// State returns the current service state.
|
||||||
func (p *Status) State() State {
|
func (p *Status) State() State {
|
||||||
p.mu.RLock()
|
p.mu.RLock()
|
||||||
defer p.mu.RUnlock()
|
defer p.mu.RUnlock()
|
||||||
@@ -49,12 +58,13 @@ func (p *Status) setState(state State) {
|
|||||||
p.state = state
|
p.state = state
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Events returns a snapshot of the recent event log. The returned slice is safe
|
||||||
|
// to modify without affecting the internal state.
|
||||||
func (p *Status) Events() []Event {
|
func (p *Status) Events() []Event {
|
||||||
events := make([]Event, MaxEventSize)
|
|
||||||
|
|
||||||
p.mu.RLock()
|
p.mu.RLock()
|
||||||
defer p.mu.RUnlock()
|
defer p.mu.RUnlock()
|
||||||
|
|
||||||
|
events := make([]Event, len(p.events))
|
||||||
copy(events, p.events)
|
copy(events, p.events)
|
||||||
return events
|
return events
|
||||||
}
|
}
|
||||||
@@ -71,6 +81,8 @@ func (p *Status) addEvent(event Event) {
|
|||||||
p.events = append(p.events, event)
|
p.events = append(p.events, event)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Stats returns the traffic statistics for the service. It safely handles nil
|
||||||
|
// receivers by returning nil.
|
||||||
func (p *Status) Stats() stats.Stats {
|
func (p *Status) Stats() stats.Stats {
|
||||||
if p == nil {
|
if p == nil {
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -0,0 +1,218 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-gost/core/observer/stats"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestStatusCreateAndState(t *testing.T) {
|
||||||
|
before := time.Now()
|
||||||
|
st := &Status{
|
||||||
|
createTime: time.Now(),
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
}
|
||||||
|
after := time.Now()
|
||||||
|
|
||||||
|
if st.CreateTime().Before(before) || st.CreateTime().After(after) {
|
||||||
|
t.Errorf("CreateTime = %v, want between %v and %v", st.CreateTime(), before, after)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Initial state is zero value
|
||||||
|
if st.State() != State("") {
|
||||||
|
t.Errorf("State() = %q, want empty", st.State())
|
||||||
|
}
|
||||||
|
|
||||||
|
st.setState(StateReady)
|
||||||
|
if got := st.State(); got != StateReady {
|
||||||
|
t.Errorf("State() = %q, want %q", got, StateReady)
|
||||||
|
}
|
||||||
|
|
||||||
|
st.setState(StateFailed)
|
||||||
|
if got := st.State(); got != StateFailed {
|
||||||
|
t.Errorf("State() = %q, want %q", got, StateFailed)
|
||||||
|
}
|
||||||
|
|
||||||
|
st.setState(StateClosed)
|
||||||
|
if got := st.State(); got != StateClosed {
|
||||||
|
t.Errorf("State() = %q, want %q", got, StateClosed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusStateConcurrency(t *testing.T) {
|
||||||
|
st := &Status{
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
}
|
||||||
|
|
||||||
|
states := []State{StateRunning, StateReady, StateFailed, StateClosed}
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for i := 0; i < 100; i++ {
|
||||||
|
wg.Add(1)
|
||||||
|
go func(i int) {
|
||||||
|
defer wg.Done()
|
||||||
|
st.setState(states[i%len(states)])
|
||||||
|
_ = st.State()
|
||||||
|
}(i)
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusEventsEmpty(t *testing.T) {
|
||||||
|
st := &Status{
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
}
|
||||||
|
|
||||||
|
events := st.Events()
|
||||||
|
if len(events) != 0 {
|
||||||
|
t.Errorf("Events() = %d events, want 0", len(events))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusEventsAddAndRetrieve(t *testing.T) {
|
||||||
|
st := &Status{
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
}
|
||||||
|
|
||||||
|
want := []Event{
|
||||||
|
{Time: time.Now(), Message: "event1"},
|
||||||
|
{Time: time.Now(), Message: "event2"},
|
||||||
|
{Time: time.Now(), Message: "event3"},
|
||||||
|
}
|
||||||
|
for _, e := range want {
|
||||||
|
st.addEvent(e)
|
||||||
|
}
|
||||||
|
|
||||||
|
events := st.Events()
|
||||||
|
if len(events) != len(want) {
|
||||||
|
t.Fatalf("Events() = %d events, want %d", len(events), len(want))
|
||||||
|
}
|
||||||
|
for i, e := range events {
|
||||||
|
if e.Message != want[i].Message {
|
||||||
|
t.Errorf("Events()[%d].Message = %q, want %q", i, e.Message, want[i].Message)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusEventsReturnsCopy(t *testing.T) {
|
||||||
|
st := &Status{
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
}
|
||||||
|
st.addEvent(Event{Message: "original"})
|
||||||
|
|
||||||
|
events := st.Events()
|
||||||
|
events[0] = Event{Message: "modified"}
|
||||||
|
|
||||||
|
original := st.Events()
|
||||||
|
if original[0].Message != "original" {
|
||||||
|
t.Error("Events() should return a copy, internal state was mutated")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusEventsRingBuffer(t *testing.T) {
|
||||||
|
st := &Status{
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
}
|
||||||
|
|
||||||
|
// Add MaxEventSize + 5 events; first 5 should be evicted
|
||||||
|
for i := 0; i < MaxEventSize+5; i++ {
|
||||||
|
st.addEvent(Event{Message: time.Now().Format(time.RFC3339Nano)})
|
||||||
|
time.Sleep(time.Microsecond) // ensure unique timestamps
|
||||||
|
}
|
||||||
|
|
||||||
|
events := st.Events()
|
||||||
|
if len(events) != MaxEventSize {
|
||||||
|
t.Fatalf("len(Events()) = %d, want %d", len(events), MaxEventSize)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusEventsConcurrency(t *testing.T) {
|
||||||
|
st := &Status{
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
}
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for i := 0; i < 100; i++ {
|
||||||
|
wg.Add(2)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
st.addEvent(Event{Message: "writer"})
|
||||||
|
}()
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
_ = st.Events()
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|
||||||
|
type mockStats struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
vals map[stats.Kind]uint64
|
||||||
|
updated bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func newMockStats() *mockStats {
|
||||||
|
return &mockStats{vals: make(map[stats.Kind]uint64)}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *mockStats) Add(kind stats.Kind, n int64) {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
m.vals[kind] += uint64(n)
|
||||||
|
m.updated = true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *mockStats) Get(kind stats.Kind) uint64 {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
return m.vals[kind]
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *mockStats) IsUpdated() bool {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
return m.updated
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *mockStats) Reset() {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
m.vals = make(map[stats.Kind]uint64)
|
||||||
|
m.updated = false
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusStats(t *testing.T) {
|
||||||
|
ms := newMockStats()
|
||||||
|
st := &Status{
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
stats: ms,
|
||||||
|
}
|
||||||
|
|
||||||
|
if got := st.Stats(); got == nil {
|
||||||
|
t.Error("Stats() = nil, want non-nil")
|
||||||
|
}
|
||||||
|
|
||||||
|
ms.Add(stats.KindTotalConns, 5)
|
||||||
|
if got := st.Stats().Get(stats.KindTotalConns); got != 5 {
|
||||||
|
t.Errorf("Stats().Get(TotalConns) = %d, want 5", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusStatsNilReceiver(t *testing.T) {
|
||||||
|
var st *Status
|
||||||
|
if got := st.Stats(); got != nil {
|
||||||
|
t.Errorf("nil Stats() = %v, want nil", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStatusStatsNil(t *testing.T) {
|
||||||
|
st := &Status{
|
||||||
|
events: make([]Event, 0, MaxEventSize),
|
||||||
|
stats: nil,
|
||||||
|
}
|
||||||
|
if got := st.Stats(); got != nil {
|
||||||
|
t.Errorf("Stats() = %v, want nil", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user