mirror of
https://github.com/router-for-me/CLIProxyAPI.git
synced 2026-06-22 06:57:37 +08:00
f353979e0a
- Introduced `NotifyUsageRefresh` in `redisqueue` to notify subscribers of usage refresh events. - Enhanced `Watcher` logic to trigger usage refresh notifications on client changes (add/update/remove). - Updated tests to validate proper broadcast of usage refresh messages to subscribers. - Added support for initial `support_refresh` payload upon subscription initialization.
91 lines
2.4 KiB
Go
91 lines
2.4 KiB
Go
package 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()
|
|
|
|
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")
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestNotifyUsageRefreshBroadcastsOnlyToUsageSubscribers(t *testing.T) {
|
|
withEnabledQueue(t, func() {
|
|
subscriber, unsubscribe := SubscribeUsage()
|
|
defer unsubscribe()
|
|
|
|
requireUsageSubscriberPayload(t, subscriber, usageSupportRefreshPayload)
|
|
|
|
NotifyUsageRefresh()
|
|
requireUsageSubscriberPayload(t, subscriber, usageRefreshPayload)
|
|
|
|
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)
|
|
}
|
|
}
|