Naming, layout and comments: descriptive full-word names, one item per line in wrapped calls and literals, and doc comments on exported names
Short and single-letter names in application code become full words (tests keep theirs). Wrapped calls, signatures and composite literals put one item per line, and one-line function bodies are expanded. Exported names and groups of constants get doc comments, and comments refer to the renamed identifiers.
This commit is contained in:
1 parent
043e307663
commit
465358609a
78 files changed
+2374
-1402
No files matched your search
@@ -55,18 +55,18 @@ var errUnsupportedData = errors.New("binary frames are not supported")
|
||||
// runs with; the client calls it when it finishes or is closed. The client's
|
||||
// messages are limited by websocket.message_rate_limit if it is enabled.
|
||||
func newClient(
|
||||
h *Hub, conn msgConn, ip string, cancel context.CancelFunc,
|
||||
hub *Hub, conn msgConn, ip string, cancel context.CancelFunc,
|
||||
) *client {
|
||||
limiter := rate.NewLimiter(rate.Inf, 0)
|
||||
if rule := h.conf.MessageRateLimit; rule.Enabled && rule.PerSecond > 0 {
|
||||
if rule := hub.conf.MessageRateLimit; rule.Enabled && rule.PerSecond > 0 {
|
||||
limiter = rate.NewLimiter(
|
||||
rate.Limit(rule.PerSecond),
|
||||
max(rule.Burst, 1),
|
||||
)
|
||||
}
|
||||
|
||||
c := &client{
|
||||
hub: h,
|
||||
client := &client{
|
||||
hub: hub,
|
||||
conn: conn,
|
||||
id: rand.Text()[:12],
|
||||
ip: ip,
|
||||
@@ -74,9 +74,9 @@ func newClient(
|
||||
cancel: cancel,
|
||||
limiter: limiter,
|
||||
}
|
||||
c.lastSeen.Store(time.Now().UnixNano())
|
||||
client.lastSeen.Store(time.Now().UnixNano())
|
||||
|
||||
return c
|
||||
return client
|
||||
}
|
||||
|
||||
// run starts the writer and keepalive goroutines and reads until the
|
||||
@@ -84,8 +84,14 @@ func newClient(
|
||||
func (c *client) run(ctx context.Context) {
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() { defer wg.Done(); c.writeLoop(ctx) }()
|
||||
go func() { defer wg.Done(); c.pingLoop(ctx) }()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
c.writeLoop(ctx)
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
c.pingLoop(ctx)
|
||||
}()
|
||||
|
||||
c.sendEnvelope(Envelope{
|
||||
Type: TypeWelcome,
|
||||
@@ -134,35 +140,35 @@ func (c *client) readLoop(ctx context.Context) error {
|
||||
continue
|
||||
}
|
||||
// Unpack the envelope and pass to handle to process if valid
|
||||
var env Envelope
|
||||
if err := json.Unmarshal(data, &env); err != nil {
|
||||
var envelope Envelope
|
||||
if err := json.Unmarshal(data, &envelope); err != nil {
|
||||
c.sendError("", "invalid JSON envelope")
|
||||
continue
|
||||
}
|
||||
c.handle(env)
|
||||
c.handle(envelope)
|
||||
}
|
||||
}
|
||||
|
||||
// handle answers one envelope from the client: ping and echo, or an error
|
||||
// for any other type.
|
||||
func (c *client) handle(env Envelope) {
|
||||
switch env.Type {
|
||||
func (c *client) handle(envelope Envelope) {
|
||||
switch envelope.Type {
|
||||
case TypePing:
|
||||
c.sendEnvelope(Envelope{
|
||||
Type: TypePong,
|
||||
ID: env.ID,
|
||||
ID: envelope.ID,
|
||||
Payload: mustRaw(map[string]any{"time": time.Now().UTC()}),
|
||||
})
|
||||
|
||||
case TypeEcho:
|
||||
c.sendEnvelope(Envelope{
|
||||
Type: TypeEcho,
|
||||
ID: env.ID,
|
||||
Payload: env.Payload,
|
||||
ID: envelope.ID,
|
||||
Payload: envelope.Payload,
|
||||
})
|
||||
|
||||
default:
|
||||
c.sendError(env.ID, "unknown message type")
|
||||
c.sendError(envelope.ID, "unknown message type")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -235,10 +241,10 @@ func (c *client) trySend(msg []byte) bool {
|
||||
}
|
||||
}
|
||||
|
||||
// sendEnvelope queues env for the client, closing the client as a slow
|
||||
// sendEnvelope queues envelope for the client, closing the client as a slow
|
||||
// consumer if its queue is full.
|
||||
func (c *client) sendEnvelope(env Envelope) {
|
||||
encoded, err := json.Marshal(env)
|
||||
func (c *client) sendEnvelope(envelope Envelope) {
|
||||
encoded, err := json.Marshal(envelope)
|
||||
if err != nil {
|
||||
c.hub.logger().Error("marshal", "client", c.id, "error", err)
|
||||
return
|
||||
@@ -298,7 +304,8 @@ func withTimeout(
|
||||
ctx context.Context, timeout time.Duration,
|
||||
) (context.Context, context.CancelFunc) {
|
||||
if timeout <= 0 {
|
||||
return ctx, func() {}
|
||||
return ctx, func() {
|
||||
}
|
||||
}
|
||||
return context.WithTimeout(ctx, timeout)
|
||||
}
|
||||
@@ -23,7 +23,11 @@ type fakeConn struct {
|
||||
closed chan ws.StatusCode
|
||||
}
|
||||
|
||||
func newFakeConn() *fakeConn { return &fakeConn{closed: make(chan ws.StatusCode, 1)} }
|
||||
func newFakeConn() *fakeConn {
|
||||
return &fakeConn{
|
||||
closed: make(chan ws.StatusCode, 1),
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeConn) Read(ctx context.Context) (ws.MessageType, []byte, error) {
|
||||
<-ctx.Done()
|
||||
@@ -52,7 +56,10 @@ func newTestClient(f *fakeConn) (*client, context.Context) {
|
||||
// An envelope that can't be marshalled is dropped, not queued.
|
||||
func TestSendEnvelopeBadPayload(t *testing.T) {
|
||||
c, _ := newTestClient(newFakeConn())
|
||||
c.sendEnvelope(Envelope{Type: TypeEcho, Payload: json.RawMessage("{")})
|
||||
c.sendEnvelope(Envelope{
|
||||
Type: TypeEcho,
|
||||
Payload: json.RawMessage("{"),
|
||||
})
|
||||
if len(c.send) != 0 {
|
||||
t.Errorf("queued %d messages; want none", len(c.send))
|
||||
}
|
||||
@@ -63,7 +70,9 @@ func TestSendEnvelopeSlowConsumer(t *testing.T) {
|
||||
f := newFakeConn()
|
||||
c, ctx := newTestClient(f)
|
||||
for range clientSendBuffer {
|
||||
c.sendEnvelope(Envelope{Type: TypePong})
|
||||
c.sendEnvelope(Envelope{
|
||||
Type: TypePong,
|
||||
})
|
||||
}
|
||||
select {
|
||||
case code := <-f.closed:
|
||||
@@ -71,7 +80,9 @@ func TestSendEnvelopeSlowConsumer(t *testing.T) {
|
||||
default:
|
||||
}
|
||||
|
||||
c.sendEnvelope(Envelope{Type: TypePong})
|
||||
c.sendEnvelope(Envelope{
|
||||
Type: TypePong,
|
||||
})
|
||||
select {
|
||||
case code := <-f.closed:
|
||||
if code != ws.StatusPolicyViolation {
|
||||
@@ -109,13 +120,31 @@ func TestClientMessageRateLimit(t *testing.T) {
|
||||
rule config.JSCRateLimitRule
|
||||
limited bool
|
||||
}{
|
||||
{config.JSCRateLimitRule{Enabled: false, PerSecond: 1, Burst: 1}, false},
|
||||
{config.JSCRateLimitRule{Enabled: true, PerSecond: 0, Burst: 1}, false},
|
||||
{config.JSCRateLimitRule{Enabled: true, PerSecond: 1, Burst: 1}, true},
|
||||
{config.JSCRateLimitRule{
|
||||
Enabled: false,
|
||||
PerSecond: 1,
|
||||
Burst: 1,
|
||||
}, false},
|
||||
{config.JSCRateLimitRule{
|
||||
Enabled: true,
|
||||
PerSecond: 0,
|
||||
Burst: 1,
|
||||
}, false},
|
||||
{config.JSCRateLimitRule{
|
||||
Enabled: true,
|
||||
PerSecond: 1,
|
||||
Burst: 1,
|
||||
}, true},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
h := NewHub(config.JSCServerWebSocket{MessageRateLimit: tt.rule}, WithLogger(discardLog))
|
||||
c := newClient(h, newFakeConn(), "", func() {})
|
||||
h := NewHub(
|
||||
config.JSCServerWebSocket{
|
||||
MessageRateLimit: tt.rule,
|
||||
},
|
||||
WithLogger(discardLog),
|
||||
)
|
||||
c := newClient(h, newFakeConn(), "", func() {
|
||||
})
|
||||
c.limiter.Allow()
|
||||
if limited := !c.limiter.Allow(); limited != tt.limited {
|
||||
t.Errorf("rule %+v: second message limited = %v; want %v", tt.rule, limited, tt.limited)
|
||||
@@ -133,7 +162,8 @@ func TestMustRaw(t *testing.T) {
|
||||
{"nil", nil, ""},
|
||||
{"raw message as is", raw, `{"a":1}`},
|
||||
{"marshalled", map[string]int{"b": 2}, `{"b":2}`},
|
||||
{"unmarshallable", func() {}, ""},
|
||||
{"unmarshallable", func() {
|
||||
}, ""},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
if got := string(mustRaw(tt.v)); got != tt.want {
|
||||
@@ -187,13 +217,20 @@ func (b *syncBuffer) entries(t *testing.T) []map[string]any {
|
||||
// buffer.
|
||||
func jsonLog(level slog.Level) (*slog.Logger, *syncBuffer) {
|
||||
buf := &syncBuffer{}
|
||||
return slog.New(slog.NewJSONHandler(buf, &slog.HandlerOptions{Level: level})), buf
|
||||
return slog.New(slog.NewJSONHandler(buf, &slog.HandlerOptions{
|
||||
Level: level,
|
||||
})), buf
|
||||
}
|
||||
|
||||
// At debug level, messages received from and sent to a client are logged.
|
||||
func TestLogMessages(t *testing.T) {
|
||||
log, buf := jsonLog(slog.LevelDebug)
|
||||
h := NewHub(config.JSCServerWebSocket{WriteTimeout: time.Second}, WithLogger(log))
|
||||
h := NewHub(
|
||||
config.JSCServerWebSocket{
|
||||
WriteTimeout: time.Second,
|
||||
},
|
||||
WithLogger(log),
|
||||
)
|
||||
peer, _ := pipeClient(t, h)
|
||||
|
||||
ping := `{"type":"ping","id":"1"}`
|
||||
@@ -236,7 +273,8 @@ func TestLogMessages(t *testing.T) {
|
||||
func TestLogMessageLimit(t *testing.T) {
|
||||
log, buf := jsonLog(slog.LevelDebug)
|
||||
h := NewHub(config.JSCServerWebSocket{}, WithLogger(log))
|
||||
c := newClient(h, newFakeConn(), "", func() {})
|
||||
c := newClient(h, newFakeConn(), "", func() {
|
||||
})
|
||||
long := strings.Repeat("x", logMessageLimit+100)
|
||||
c.logMessage(context.Background(), "message sent", []byte(long))
|
||||
|
||||
|
||||
@@ -30,16 +30,16 @@ func (h *Hub) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
ip := middleware.ClientIP(r)
|
||||
ctx, cancel := context.WithCancel(r.Context())
|
||||
c := newClient(h, conn, ip, cancel)
|
||||
client := newClient(h, conn, ip, cancel)
|
||||
|
||||
if !h.register(c) {
|
||||
if !h.register(client) {
|
||||
cancel()
|
||||
conn.Close(ws.StatusGoingAway, "server shutting down") //nolint:errcheck,gosec // turned away
|
||||
return
|
||||
}
|
||||
defer h.unregister(c)
|
||||
defer h.unregister(client)
|
||||
|
||||
h.logger().Info("client connected", "client", c.id, "ip", ip)
|
||||
c.run(ctx)
|
||||
h.logger().Info("client disconnected", "client", c.id, "ip", ip)
|
||||
h.logger().Info("client connected", "client", client.id, "ip", ip)
|
||||
client.run(ctx)
|
||||
h.logger().Info("client disconnected", "client", client.id, "ip", ip)
|
||||
}
|
||||
@@ -35,7 +35,9 @@ func dialWS(t *testing.T, ctx context.Context, url string) *ws.Conn {
|
||||
if err != nil {
|
||||
t.Fatalf("dial: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { conn.CloseNow() })
|
||||
t.Cleanup(func() {
|
||||
conn.CloseNow()
|
||||
})
|
||||
var e Envelope
|
||||
if err := wsjson.Read(ctx, conn, &e); err != nil || e.Type != TypeWelcome {
|
||||
t.Fatalf("want welcome, got %+v, %v", e, err)
|
||||
@@ -46,13 +48,21 @@ func dialWS(t *testing.T, ctx context.Context, url string) *ws.Conn {
|
||||
// A client upgraded over HTTP gets replies. A max_message_size of 0 uses the
|
||||
// default limit rather than refusing every message.
|
||||
func TestHTTPMessages(t *testing.T) {
|
||||
h := NewHub(config.JSCServerWebSocket{WriteTimeout: time.Second}, WithLogger(discardLog))
|
||||
h := NewHub(
|
||||
config.JSCServerWebSocket{
|
||||
WriteTimeout: time.Second,
|
||||
},
|
||||
WithLogger(discardLog),
|
||||
)
|
||||
url := wsHub(t, h)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
conn := dialWS(t, ctx, url)
|
||||
|
||||
if err := wsjson.Write(ctx, conn, Envelope{Type: TypePing, ID: "1"}); err != nil {
|
||||
if err := wsjson.Write(ctx, conn, Envelope{
|
||||
Type: TypePing,
|
||||
ID: "1",
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var e Envelope
|
||||
@@ -89,7 +99,8 @@ func TestPushBadPayload(t *testing.T) {
|
||||
c.read() // welcome
|
||||
waitCount(t, h, 1)
|
||||
|
||||
h.Push("bad", func() {})
|
||||
h.Push("bad", func() {
|
||||
})
|
||||
h.Push("good", nil)
|
||||
if e := c.read(); e.Type != "good" || e.Payload != nil {
|
||||
t.Fatalf("want good with no payload, got %+v", e)
|
||||
@@ -98,7 +109,12 @@ func TestPushBadPayload(t *testing.T) {
|
||||
|
||||
// A binary frame closes the connection with 1003 (unsupported data).
|
||||
func TestHTTPBinaryFrame(t *testing.T) {
|
||||
h := NewHub(config.JSCServerWebSocket{WriteTimeout: time.Second}, WithLogger(discardLog))
|
||||
h := NewHub(
|
||||
config.JSCServerWebSocket{
|
||||
WriteTimeout: time.Second,
|
||||
},
|
||||
WithLogger(discardLog),
|
||||
)
|
||||
url := wsHub(t, h)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
@@ -117,8 +133,12 @@ func TestHTTPBinaryFrame(t *testing.T) {
|
||||
// handled; the connection stays open.
|
||||
func TestHTTPMessageRateLimit(t *testing.T) {
|
||||
conf := config.JSCServerWebSocket{
|
||||
WriteTimeout: time.Second,
|
||||
MessageRateLimit: config.JSCRateLimitRule{Enabled: true, PerSecond: 1, Burst: 1},
|
||||
WriteTimeout: time.Second,
|
||||
MessageRateLimit: config.JSCRateLimitRule{
|
||||
Enabled: true,
|
||||
PerSecond: 1,
|
||||
Burst: 1,
|
||||
},
|
||||
}
|
||||
h := NewHub(conf, WithLogger(discardLog))
|
||||
url := wsHub(t, h)
|
||||
@@ -127,7 +147,10 @@ func TestHTTPMessageRateLimit(t *testing.T) {
|
||||
conn := dialWS(t, ctx, url)
|
||||
|
||||
for _, id := range []string{"1", "2"} {
|
||||
if err := wsjson.Write(ctx, conn, Envelope{Type: TypePing, ID: id}); err != nil {
|
||||
if err := wsjson.Write(ctx, conn, Envelope{
|
||||
Type: TypePing,
|
||||
ID: id,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
+18
-16
@@ -17,7 +17,7 @@ type Hub struct {
|
||||
conf config.JSCServerWebSocket
|
||||
log *slog.Logger
|
||||
|
||||
m sync.RWMutex
|
||||
mu sync.RWMutex
|
||||
clients map[*client]struct{}
|
||||
closed bool
|
||||
wg sync.WaitGroup
|
||||
@@ -82,13 +82,13 @@ func (h *Hub) Push(msgType string, payload any) {
|
||||
}
|
||||
|
||||
var slow []*client
|
||||
h.m.RLock()
|
||||
h.mu.RLock()
|
||||
for client := range h.clients {
|
||||
if !client.trySend(msg) {
|
||||
slow = append(slow, client)
|
||||
}
|
||||
}
|
||||
h.m.RUnlock()
|
||||
h.mu.RUnlock()
|
||||
|
||||
for _, client := range slow {
|
||||
h.logger().Warn("dropping slow client", "client", client.id, "ip", client.ip)
|
||||
@@ -98,8 +98,8 @@ func (h *Hub) Push(msgType string, payload any) {
|
||||
|
||||
// Count returns the number of connected clients.
|
||||
func (h *Hub) Count() int {
|
||||
h.m.RLock()
|
||||
defer h.m.RUnlock()
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
return len(h.clients)
|
||||
}
|
||||
|
||||
@@ -108,13 +108,13 @@ func (h *Hub) Count() int {
|
||||
// closed at once and ctx's error is returned. New connections are refused
|
||||
// afterwards.
|
||||
func (h *Hub) Shutdown(ctx context.Context) error {
|
||||
h.m.Lock()
|
||||
h.mu.Lock()
|
||||
h.closed = true
|
||||
clientShutdowns := make([]*client, 0, len(h.clients))
|
||||
for client := range h.clients {
|
||||
clientShutdowns = append(clientShutdowns, client)
|
||||
}
|
||||
h.m.Unlock()
|
||||
h.mu.Unlock()
|
||||
|
||||
if len(clientShutdowns) == 0 {
|
||||
// Nothing to close. Clients that just left are only finishing their
|
||||
@@ -151,7 +151,9 @@ func (h *Hub) Shutdown(ctx context.Context) error {
|
||||
}
|
||||
|
||||
// maxMessageSize is the largest message read from a client: max_message_size,
|
||||
// or defaultMaxMessageSize if that is 0 or less.
|
||||
// or defaultMaxMessageSize if that is 0 or less. The configuration rejects -1
|
||||
// (no limit) for this setting, as a client could then exhaust the server's
|
||||
// memory; a -1 set in code also gets the default, never no limit.
|
||||
func (h *Hub) maxMessageSize() int {
|
||||
if h.conf.MaxMessageSize <= 0 {
|
||||
return defaultMaxMessageSize
|
||||
@@ -161,21 +163,21 @@ func (h *Hub) maxMessageSize() int {
|
||||
|
||||
// register adds a client to the hub. Hub clients receive all push messages and
|
||||
// have their lifecycles tracked by the hub.
|
||||
func (h *Hub) register(c *client) bool {
|
||||
h.m.Lock()
|
||||
defer h.m.Unlock()
|
||||
func (h *Hub) register(client *client) bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.closed {
|
||||
return false
|
||||
}
|
||||
h.clients[c] = struct{}{}
|
||||
h.clients[client] = struct{}{}
|
||||
h.wg.Add(1)
|
||||
return true
|
||||
}
|
||||
|
||||
// unregister removes a client from the hub.
|
||||
func (h *Hub) unregister(c *client) {
|
||||
h.m.Lock()
|
||||
delete(h.clients, c)
|
||||
h.m.Unlock()
|
||||
func (h *Hub) unregister(client *client) {
|
||||
h.mu.Lock()
|
||||
delete(h.clients, client)
|
||||
h.mu.Unlock()
|
||||
h.wg.Done()
|
||||
}
|
||||
@@ -17,9 +17,15 @@ import (
|
||||
func pipeClient(t *testing.T, h *Hub) (net.Conn, chan struct{}) {
|
||||
t.Helper()
|
||||
server, peer := net.Pipe()
|
||||
t.Cleanup(func() { server.Close(); peer.Close() })
|
||||
t.Cleanup(func() {
|
||||
server.Close()
|
||||
peer.Close()
|
||||
})
|
||||
done := make(chan struct{})
|
||||
go func() { h.ServeTCP(context.Background(), server); close(done) }()
|
||||
go func() {
|
||||
h.ServeTCP(context.Background(), server)
|
||||
close(done)
|
||||
}()
|
||||
peer.SetReadDeadline(time.Now().Add(time.Second))
|
||||
if _, err := bufio.NewReader(peer).ReadString('\n'); err != nil {
|
||||
t.Fatalf("welcome: %v", err)
|
||||
@@ -61,10 +67,13 @@ func TestShutdownNoClients(t *testing.T) {
|
||||
|
||||
// A ping that can't be sent within the pong timeout drops the client.
|
||||
func TestPingFailureDropsClient(t *testing.T) {
|
||||
h := NewHub(config.JSCServerWebSocket{
|
||||
PingInterval: 10 * time.Millisecond,
|
||||
PongTimeout: 20 * time.Millisecond,
|
||||
}, WithLogger(discardLog))
|
||||
h := NewHub(
|
||||
config.JSCServerWebSocket{
|
||||
PingInterval: 10 * time.Millisecond,
|
||||
PongTimeout: 20 * time.Millisecond,
|
||||
},
|
||||
WithLogger(discardLog),
|
||||
)
|
||||
_, done := pipeClient(t, h)
|
||||
|
||||
select {
|
||||
@@ -81,8 +90,12 @@ func TestMaxMessageSize(t *testing.T) {
|
||||
want int
|
||||
}{
|
||||
{config.JSCServerWebSocket{}, defaultMaxMessageSize},
|
||||
{config.JSCServerWebSocket{MaxMessageSize: -1}, defaultMaxMessageSize},
|
||||
{config.JSCServerWebSocket{MaxMessageSize: 100}, 100},
|
||||
{config.JSCServerWebSocket{
|
||||
MaxMessageSize: -1,
|
||||
}, defaultMaxMessageSize},
|
||||
{config.JSCServerWebSocket{
|
||||
MaxMessageSize: 100,
|
||||
}, 100},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
h := NewHub(tt.conf, WithLogger(discardLog))
|
||||
|
||||
@@ -37,28 +37,36 @@ type Envelope struct {
|
||||
// Envelope types. TypeEvent, TypeRequest and TypeAck are reserved: nothing
|
||||
// sends or handles them yet.
|
||||
const (
|
||||
// TypeWelcome is sent to a client when it connects, with its client ID.
|
||||
TypeWelcome = "welcome"
|
||||
TypePing = "ping"
|
||||
TypePong = "pong"
|
||||
TypeEcho = "echo"
|
||||
TypeEvent = "event"
|
||||
// TypePing asks the server for a TypePong.
|
||||
TypePing = "ping"
|
||||
// TypePong answers a TypePing, with the server's time.
|
||||
TypePong = "pong"
|
||||
// TypeEcho is sent back to the client with the same ID and payload.
|
||||
TypeEcho = "echo"
|
||||
// TypeEvent is reserved for events pushed to clients.
|
||||
TypeEvent = "event"
|
||||
// TypeRequest is reserved for requests from clients.
|
||||
TypeRequest = "request"
|
||||
TypeAck = "ack"
|
||||
TypeError = "error"
|
||||
// TypeAck is reserved for acknowledging a request.
|
||||
TypeAck = "ack"
|
||||
// TypeError reports an error, with the ID of the message it answers.
|
||||
TypeError = "error"
|
||||
)
|
||||
|
||||
// mustRaw returns v as raw JSON: v itself if it is already json.RawMessage,
|
||||
// nil if v is nil or can't be marshalled.
|
||||
func mustRaw(v any) json.RawMessage {
|
||||
if v == nil {
|
||||
// mustRaw returns value as raw JSON: value itself if it is already
|
||||
// json.RawMessage, nil if value is nil or can't be marshalled.
|
||||
func mustRaw(value any) json.RawMessage {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
if raw, ok := v.(json.RawMessage); ok {
|
||||
if raw, ok := value.(json.RawMessage); ok {
|
||||
return raw
|
||||
}
|
||||
b, err := json.Marshal(v)
|
||||
encoded, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return b
|
||||
return encoded
|
||||
}
|
||||
@@ -178,7 +178,10 @@ func (t *tcpConn) Ping(ctx context.Context) error {
|
||||
// connection. Only the first Close or CloseNow does anything; later calls
|
||||
// return nil. A CloseNow while Close is sending its message cuts it short.
|
||||
func (t *tcpConn) Close(code ws.StatusCode, reason string) error {
|
||||
if !t.markClosed(ws.CloseError{Code: code, Reason: reason}) {
|
||||
if !t.markClosed(ws.CloseError{
|
||||
Code: code,
|
||||
Reason: reason,
|
||||
}) {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -206,7 +209,9 @@ func (t *tcpConn) Close(code ws.StatusCode, reason string) error {
|
||||
// CloseNow closes the connection without a message. It doesn't wait for a
|
||||
// Close that is sending its message.
|
||||
func (t *tcpConn) CloseNow() error {
|
||||
t.markClosed(ws.CloseError{Code: ws.StatusAbnormalClosure})
|
||||
t.markClosed(ws.CloseError{
|
||||
Code: ws.StatusAbnormalClosure,
|
||||
})
|
||||
if err := t.conn.Close(); err != nil && !errors.Is(err, net.ErrClosed) {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -69,8 +69,14 @@ func dialTCP(t *testing.T, addr string) *tcpClient {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { conn.Close() })
|
||||
return &tcpClient{t: t, conn: conn, r: bufio.NewReader(conn)}
|
||||
t.Cleanup(func() {
|
||||
conn.Close()
|
||||
})
|
||||
return &tcpClient{
|
||||
t: t,
|
||||
conn: conn,
|
||||
r: bufio.NewReader(conn),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *tcpClient) send(line string) {
|
||||
@@ -155,7 +161,9 @@ func TestTCPMessages(t *testing.T) {
|
||||
// With no write timeout, messages are still sent: a zero timeout must not
|
||||
// mean one that has already expired.
|
||||
func TestTCPNoWriteTimeout(t *testing.T) {
|
||||
_, addr := tcpHub(t, func(c *config.JSCServerWebSocket) { c.WriteTimeout = 0 })
|
||||
_, addr := tcpHub(t, func(c *config.JSCServerWebSocket) {
|
||||
c.WriteTimeout = 0
|
||||
})
|
||||
c := dialTCP(t, addr)
|
||||
if e := c.read(); e.Type != TypeWelcome {
|
||||
t.Fatalf("want welcome, got %+v", e)
|
||||
@@ -177,7 +185,9 @@ func TestTCPClientDisconnect(t *testing.T) {
|
||||
|
||||
// A line longer than max_message_size drops the connection.
|
||||
func TestTCPLineTooLong(t *testing.T) {
|
||||
h, addr := tcpHub(t, func(c *config.JSCServerWebSocket) { c.MaxMessageSize = 64 })
|
||||
h, addr := tcpHub(t, func(c *config.JSCServerWebSocket) {
|
||||
c.MaxMessageSize = 64
|
||||
})
|
||||
c := dialTCP(t, addr)
|
||||
c.read() // welcome
|
||||
c.send(`{"type":"echo","payload":"` + strings.Repeat("x", 100) + `"}` + "\n")
|
||||
@@ -244,12 +254,21 @@ func TestTCPPing(t *testing.T) {
|
||||
// full. net.Pipe has no buffer, so writes to it block until the peer reads.
|
||||
func TestTCPSlowConsumer(t *testing.T) {
|
||||
h := NewHub(
|
||||
config.JSCServerWebSocket{WriteTimeout: 50 * time.Millisecond}, WithLogger(discardLog),
|
||||
config.JSCServerWebSocket{
|
||||
WriteTimeout: 50 * time.Millisecond,
|
||||
},
|
||||
WithLogger(discardLog),
|
||||
)
|
||||
server, peer := net.Pipe()
|
||||
t.Cleanup(func() { server.Close(); peer.Close() })
|
||||
t.Cleanup(func() {
|
||||
server.Close()
|
||||
peer.Close()
|
||||
})
|
||||
done := make(chan struct{})
|
||||
go func() { h.ServeTCP(context.Background(), server); close(done) }()
|
||||
go func() {
|
||||
h.ServeTCP(context.Background(), server)
|
||||
close(done)
|
||||
}()
|
||||
waitCount(t, h, 1)
|
||||
|
||||
for range clientSendBuffer + 1 {
|
||||
@@ -283,7 +302,9 @@ func TestTCPNoPongTimeout(t *testing.T) {
|
||||
|
||||
// A max_message_size of 0 uses the default limit.
|
||||
func TestTCPDefaultMessageSize(t *testing.T) {
|
||||
_, addr := tcpHub(t, func(c *config.JSCServerWebSocket) { c.MaxMessageSize = 0 })
|
||||
_, addr := tcpHub(t, func(c *config.JSCServerWebSocket) {
|
||||
c.MaxMessageSize = 0
|
||||
})
|
||||
c := dialTCP(t, addr)
|
||||
c.read() // welcome
|
||||
c.send(`{"type":"echo","id":"1","payload":"` + strings.Repeat("x", 4096) + `"}` + "\n")
|
||||
@@ -299,7 +320,10 @@ func TestTCPDefaultMessageSize(t *testing.T) {
|
||||
func pipeConn(t *testing.T, closeTimeout time.Duration) (*tcpConn, net.Conn) {
|
||||
t.Helper()
|
||||
server, peer := net.Pipe()
|
||||
t.Cleanup(func() { server.Close(); peer.Close() })
|
||||
t.Cleanup(func() {
|
||||
server.Close()
|
||||
peer.Close()
|
||||
})
|
||||
return newTCPConn(server, 0, closeTimeout), peer
|
||||
}
|
||||
|
||||
@@ -370,13 +394,17 @@ func TestTCPConnClose(t *testing.T) {
|
||||
func TestTCPConnCloseNowDuringClose(t *testing.T) {
|
||||
tc, _ := pipeConn(t, 0) // no timeout: Close would wait for ever
|
||||
closed := make(chan error, 1)
|
||||
go func() { closed <- tc.Close(ws.StatusGoingAway, "bye") }()
|
||||
go func() {
|
||||
closed <- tc.Close(ws.StatusGoingAway, "bye")
|
||||
}()
|
||||
for !tc.isClosed() {
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- tc.CloseNow() }()
|
||||
go func() {
|
||||
done <- tc.CloseNow()
|
||||
}()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
@@ -396,7 +424,9 @@ func TestTCPConnCloseNowDuringClose(t *testing.T) {
|
||||
func TestTCPConnCloseTimeout(t *testing.T) {
|
||||
tc, _ := pipeConn(t, 20*time.Millisecond)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- tc.Close(ws.StatusGoingAway, "bye") }()
|
||||
go func() {
|
||||
done <- tc.Close(ws.StatusGoingAway, "bye")
|
||||
}()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
|
||||
@@ -23,10 +23,14 @@ var discardLog = slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
// not nil.
|
||||
func testConf(mutate func(*config.JSCServer)) *config.JSCServer {
|
||||
c := &config.JSCServer{
|
||||
Host: "127.0.0.1",
|
||||
Port: 0,
|
||||
WebSocket: config.JSCServerWebSocket{WriteTimeout: time.Second},
|
||||
Timeouts: config.JSCServerTimeouts{ShutdownTimeout: time.Second},
|
||||
Host: "127.0.0.1",
|
||||
Port: 0,
|
||||
WebSocket: config.JSCServerWebSocket{
|
||||
WriteTimeout: time.Second,
|
||||
},
|
||||
Timeouts: config.JSCServerTimeouts{
|
||||
ShutdownTimeout: time.Second,
|
||||
},
|
||||
}
|
||||
if mutate != nil {
|
||||
mutate(c)
|
||||
@@ -54,7 +58,9 @@ func startServer(
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(cancel)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- s.serve(ctx, ln) }()
|
||||
go func() {
|
||||
done <- s.serve(ctx, ln)
|
||||
}()
|
||||
return h, ln.Addr().String(), cancel, done
|
||||
}
|
||||
|
||||
@@ -71,8 +77,14 @@ func dial(t *testing.T, addr string) *tcpClient {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { conn.Close() })
|
||||
return &tcpClient{t: t, conn: conn, r: bufio.NewReader(conn)}
|
||||
t.Cleanup(func() {
|
||||
conn.Close()
|
||||
})
|
||||
return &tcpClient{
|
||||
t: t,
|
||||
conn: conn,
|
||||
r: bufio.NewReader(conn),
|
||||
}
|
||||
}
|
||||
|
||||
// read returns the next envelope, failing the test after a second.
|
||||
@@ -188,7 +200,9 @@ func TestServerRun(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestServerRunListenError(t *testing.T) {
|
||||
s, _ := newServer(testConf(func(c *config.JSCServer) { c.Port = -1 }))
|
||||
s, _ := newServer(testConf(func(c *config.JSCServer) {
|
||||
c.Port = -1
|
||||
}))
|
||||
if err := s.Run(context.Background()); err == nil || !strings.Contains(err.Error(), "listen") {
|
||||
t.Fatalf("Run = %v; want a listen error", err)
|
||||
}
|
||||
@@ -217,7 +231,10 @@ func TestServerAcceptRetries(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer ln.Close()
|
||||
fl := &flakyListener{Listener: ln, fails: 3}
|
||||
fl := &flakyListener{
|
||||
Listener: ln,
|
||||
fails: 3,
|
||||
}
|
||||
err = s.accept(context.Background(), context.Background(), fl)
|
||||
if !errors.Is(err, net.ErrClosed) || fl.fails != 0 {
|
||||
t.Fatalf("accept = %v with %d failures left; want net.ErrClosed", err, fl.fails)
|
||||
@@ -233,7 +250,10 @@ func TestServerAcceptCancelledInBackOff(t *testing.T) {
|
||||
}
|
||||
defer ln.Close()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
fl := &flakyListener{Listener: ln, fails: 1 << 30}
|
||||
fl := &flakyListener{
|
||||
Listener: ln,
|
||||
fails: 1 << 30,
|
||||
}
|
||||
time.AfterFunc(20*time.Millisecond, cancel)
|
||||
if err := s.accept(ctx, ctx, fl); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("accept = %v; want context.Canceled", err)
|
||||
|
||||
Reference in new issue
Block a user