| 1234567891011121314151617181920212223242526272829303132333435363738394041424344 |
- package repository
- import (
- "context"
- "testing"
- "time"
- "github.com/alicebob/miniredis/v2"
- "github.com/redis/go-redis/v9"
- "github.com/stretchr/testify/require"
- )
- func TestChannelCachePubSub(t *testing.T) {
- redisServer := miniredis.RunT(t)
- publisherRedis := redis.NewClient(&redis.Options{Addr: redisServer.Addr()})
- subscriberRedis := redis.NewClient(&redis.Options{Addr: redisServer.Addr()})
- t.Cleanup(func() {
- _ = publisherRedis.Close()
- _ = subscriberRedis.Close()
- })
- publisher := NewChannelCache(publisherRedis)
- subscriber := NewChannelCache(subscriberRedis)
- ctx, cancel := context.WithCancel(context.Background())
- t.Cleanup(cancel)
- received := make(chan struct{}, 1)
- subscriber.SubscribeUpdates(ctx, func() {
- select {
- case received <- struct{}{}:
- default:
- }
- })
- require.Eventually(t, func() bool {
- return redisServer.PubSubNumSub(channelCachePubSubKey)[channelCachePubSubKey] == 1
- }, time.Second, 10*time.Millisecond)
- require.NoError(t, publisher.NotifyUpdate(context.Background()))
- select {
- case <-received:
- case <-time.After(time.Second):
- t.Fatal("channel cache invalidation notification was not received")
- }
- }
|