channel_cache_test.go 1.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344
  1. package repository
  2. import (
  3. "context"
  4. "testing"
  5. "time"
  6. "github.com/alicebob/miniredis/v2"
  7. "github.com/redis/go-redis/v9"
  8. "github.com/stretchr/testify/require"
  9. )
  10. func TestChannelCachePubSub(t *testing.T) {
  11. redisServer := miniredis.RunT(t)
  12. publisherRedis := redis.NewClient(&redis.Options{Addr: redisServer.Addr()})
  13. subscriberRedis := redis.NewClient(&redis.Options{Addr: redisServer.Addr()})
  14. t.Cleanup(func() {
  15. _ = publisherRedis.Close()
  16. _ = subscriberRedis.Close()
  17. })
  18. publisher := NewChannelCache(publisherRedis)
  19. subscriber := NewChannelCache(subscriberRedis)
  20. ctx, cancel := context.WithCancel(context.Background())
  21. t.Cleanup(cancel)
  22. received := make(chan struct{}, 1)
  23. subscriber.SubscribeUpdates(ctx, func() {
  24. select {
  25. case received <- struct{}{}:
  26. default:
  27. }
  28. })
  29. require.Eventually(t, func() bool {
  30. return redisServer.PubSubNumSub(channelCachePubSubKey)[channelCachePubSubKey] == 1
  31. }, time.Second, 10*time.Millisecond)
  32. require.NoError(t, publisher.NotifyUpdate(context.Background()))
  33. select {
  34. case <-received:
  35. case <-time.After(time.Second):
  36. t.Fatal("channel cache invalidation notification was not received")
  37. }
  38. }