mirror of
https://github.com/go-gitea/gitea.git
synced 2026-10-30 08:51:51 +00:00
silverwind
7308c5fb97
Keep notification counts and stopwatches in sync when tabs open or reconnect, closing the gap behind [this flake](https://github.com/go-gitea/gitea/actions/runs/31312525323/job/93242429394). 1. Send current notification counts and stopwatch state on each WebSocket connection, replacing the separate state fetches tabs made on every connect. 2. Tabs joining an already open connection fetch current state once, since its snapshot went out before they attached. 3. Send events queued during the handshake before the snapshot, so older events can't roll it back. 4. Wait for the Redis SUBSCRIBE acknowledgement, so the snapshot read after subscribing can't miss an event.
73 lines
2.2 KiB
Go
73 lines
2.2 KiB
Go
// Copyright 2026 The Gitea Authors. All rights reserved.
|
|
// SPDX-License-Identifier: MIT
|
|
|
|
package pubsub
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.dev/modules/test"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
// TestRedisBroker runs the shared broker scenarios against a real Redis-backed
|
|
// RedisBroker, plus the backend-specific ones. Delivery crosses a Redis
|
|
// round-trip (hence the wider timeout) and HasTopicSubscribers answers
|
|
// conservatively true, so subscriber tracking is not exact.
|
|
// One redis-server is shared by all of them; starting one per test dominated
|
|
// the package runtime.
|
|
func TestRedisBroker(t *testing.T) {
|
|
redisConn := test.PrepareTestRedis(t)
|
|
|
|
newBroker := func(t *testing.T) Broker {
|
|
broker, err := NewRedisBroker(redisConn)
|
|
require.NoError(t, err)
|
|
return broker
|
|
}
|
|
testBrokerBasic(t, newBroker, 2*time.Second)
|
|
|
|
// RedisBroker cannot see other processes' subscribers, so it answers true even
|
|
// with none locally rather than risk dropping a push.
|
|
t.Run("HasTopicSubscribersWithoutAny", func(t *testing.T) {
|
|
assert.True(t, newBroker(t).HasTopicSubscribers("topic"))
|
|
})
|
|
|
|
// RedisBroker tears down its per-topic Redis subscription and internal
|
|
// state once the last local subscriber cancels.
|
|
t.Run("CancelCleansTopicState", func(t *testing.T) {
|
|
b, isRedisBroker := newBroker(t).(*RedisBroker)
|
|
require.True(t, isRedisBroker)
|
|
ch, cancel := b.Subscribe(t.Name())
|
|
cancel()
|
|
|
|
_, ok := <-ch
|
|
assert.False(t, ok, "channel must be closed after cancel")
|
|
|
|
b.mu.RLock()
|
|
_, present := b.topics[t.Name()]
|
|
b.mu.RUnlock()
|
|
assert.False(t, present, "topic state must be removed after last subscriber cancels")
|
|
})
|
|
|
|
// Two RedisBroker instances sharing one Redis simulate two Gitea processes -
|
|
// a publish on one must reach a subscriber on the other.
|
|
t.Run("CrossBroker", func(t *testing.T) {
|
|
publisher, subscriber := newBroker(t), newBroker(t)
|
|
|
|
ch, cancel := subscriber.Subscribe(t.Name())
|
|
defer cancel()
|
|
|
|
publisher.Publish(t.Name(), []byte("cross-process"))
|
|
|
|
select {
|
|
case msg := <-ch:
|
|
assert.Equal(t, []byte("cross-process"), msg)
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("subscriber on second broker did not receive message")
|
|
}
|
|
})
|
|
}
|