feat: return all keys

cluster/cluster_test_tool/pubsub_cluster_fixture.go

@@ -208,7 +208,7 @@ type Delivery struct {
 	Data     int
 }
 
-func NewInMemorySubscriberStore() *InMemorySubscribersStore[*cluster.Subscribers] {
+func NewInMemorySubscriberStore() cluster.KeyValueStore[*cluster.Subscribers] {
 	return &InMemorySubscribersStore[*cluster.Subscribers]{
 		store: &sync.Map{},
 	}
@@ -236,3 +236,12 @@ func (i *InMemorySubscribersStore[T]) Clear(_ context.Context, key string) error
 	i.store.Delete(key)
 	return nil
 }
+
+func (i *InMemorySubscribersStore[T]) Keys(_ context.Context) ([]string, error) {
+	keys := make([]string, 0)
+	i.store.Range(func(key, value interface{}) bool {
+		keys = append(keys, key.(string))
+		return true
+	})
+	return keys, nil
+}

cluster/key_value_store.go

@@ -10,11 +10,17 @@ type KeyValueStore[T any] interface {
 	Get(ctx context.Context, key string) (T, error)
 	// Clear the value for the given key.
 	Clear(ctx context.Context, key string) error
+	// Keys returns all the keys in the store.
+	Keys(ctx context.Context) ([]string, error)
 }
 
 // EmptyKeyValueStore is a key value store that does nothing.
 type EmptyKeyValueStore[T any] struct{}
 
+func NewEmptyKeyValueStore[T any]() KeyValueStore[T] {
+	return &EmptyKeyValueStore[T]{}
+}
+
 func (e *EmptyKeyValueStore[T]) Set(_ context.Context, _ string, _ T) error { return nil }
 
 func (e *EmptyKeyValueStore[T]) Get(_ context.Context, _ string) (T, error) {
@@ -23,3 +29,7 @@ func (e *EmptyKeyValueStore[T]) Get(_ context.Context, _ string) (T, error) {
 }
 
 func (e *EmptyKeyValueStore[T]) Clear(_ context.Context, _ string) error { return nil }
+
+func (e *EmptyKeyValueStore[T]) Keys(_ context.Context) ([]string, error) {
+	return make([]string, 0), nil
+}