Something went wrong. Try again.
sloproxy
Something went wrong. Try again.
3.6 kB · 135 lines
Go
at commit e8e39526
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136package redisqueue
import ( "testing" "time")
func TestEnqueueBroadcastsToUsageSubscribersAndSkipsQueue(t *testing.T) { withEnabledQueue(t, func() { first, unsubscribeFirst := SubscribeUsage() defer unsubscribeFirst() second, unsubscribeSecond := SubscribeUsage() defer unsubscribeSecond()
requireUsageSubscriberPayload(t, first, usageSupportRefreshPayload) requireUsageSubscriberPayload(t, second, usageSupportRefreshPayload)
Enqueue([]byte("usage-record"))
requireUsageSubscriberPayload(t, first, "usage-record") requireUsageSubscriberPayload(t, second, "usage-record")
if items := PopOldest(1); len(items) != 0 { t.Fatalf("PopOldest() items = %q, want empty after subscriber broadcast", items) }
unsubscribeFirst() unsubscribeSecond()
Enqueue([]byte("queued-record")) items := PopOldest(1) if len(items) != 1 || string(items[0]) != "queued-record" { t.Fatalf("PopOldest() items = %q, want queued record after unsubscribe", items) } })}
func TestSetEnabledFalseClosesUsageSubscribers(t *testing.T) { withEnabledQueue(t, func() { subscriber, unsubscribe := SubscribeUsage() defer unsubscribe() errorSubscriber, unsubscribeErrors := SubscribeErrors() defer unsubscribeErrors()
requireUsageSubscriberPayload(t, subscriber, usageSupportRefreshPayload)
SetEnabled(false)
select { case _, ok := <-subscriber: if ok { t.Fatalf("subscriber channel remained open after SetEnabled(false)") } case <-time.After(time.Second): t.Fatalf("timeout waiting for subscriber close") }
select { case _, ok := <-errorSubscriber: if ok { t.Fatalf("error subscriber channel remained open after SetEnabled(false)") } case <-time.After(time.Second): t.Fatalf("timeout waiting for error subscriber close") } })}
func TestEnqueueErrorBroadcastsToErrorSubscribersAndDiscardsWithoutSubscribers(t *testing.T) { withEnabledQueue(t, func() { subscriber, unsubscribe := SubscribeErrors() defer unsubscribe()
EnqueueError([]byte("error-record")) requireUsageSubscriberPayload(t, subscriber, "error-record")
unsubscribe()
EnqueueError([]byte("discarded-error")) requireErrorQueueEmpty(t) })}
func TestNotifyUsageRefreshBroadcastsOnlyToUsageSubscribers(t *testing.T) { withEnabledQueue(t, func() { subscriber, unsubscribe := SubscribeUsage() defer unsubscribe() errorSubscriber, unsubscribeErrors := SubscribeErrors() defer unsubscribeErrors()
requireUsageSubscriberPayload(t, subscriber, usageSupportRefreshPayload)
NotifyUsageRefresh() requireUsageSubscriberPayload(t, subscriber, usageRefreshPayload)
select { case got := <-errorSubscriber: t.Fatalf("error subscriber received usage refresh payload %q", string(got)) default: }
unsubscribe() NotifyUsageRefresh() if items := PopOldest(1); len(items) != 0 { t.Fatalf("PopOldest() items = %q, want empty after refresh notification without subscribers", items) } })}
func requireUsageSubscriberPayload(t *testing.T, subscriber <-chan []byte, want string) { t.Helper()
select { case got, ok := <-subscriber: if !ok { t.Fatalf("subscriber closed before receiving %q", want) } if string(got) != want { t.Fatalf("subscriber payload = %q, want %q", string(got), want) } case <-time.After(time.Second): t.Fatalf("timeout waiting for subscriber payload %q", want) }}
func requireErrorQueueEmpty(t *testing.T) { t.Helper()
errorGlobal.mu.Lock() defer errorGlobal.mu.Unlock()
if len(errorGlobal.items)-errorGlobal.head != 0 { t.Fatalf("error queue retained %d item(s), want none", len(errorGlobal.items)-errorGlobal.head) }}