diff --git a/backend_redis.go b/backend_redis.go index c5e2d8b..864ed3e 100644 --- a/backend_redis.go +++ b/backend_redis.go @@ -6,6 +6,7 @@ import ( "fmt" "log" "sync" + "time" "github.com/redis/go-redis/v9" ) @@ -53,6 +54,7 @@ func (r *RedisBackend) Publish(event *Event) error { pipe := r.client.Pipeline() pipe.LPush(ctx, redisBufferKey, data) pipe.LTrim(ctx, redisBufferKey, 0, int64(r.bufferSize-1)) + pipe.Expire(ctx, redisBufferKey, 24*time.Hour) pipe.Publish(ctx, redisChannel, data) _, err = pipe.Exec(ctx) return err diff --git a/backend_redis_test.go b/backend_redis_test.go index 5852a09..e73a644 100644 --- a/backend_redis_test.go +++ b/backend_redis_test.go @@ -169,6 +169,21 @@ func TestRedisBackend_bufferTrims(t *testing.T) { } } +func TestRedisBackend_bufferHasTTL(t *testing.T) { + backend, mr := newTestRedisBackend(t, 100) + defer backend.Close() + + backend.Publish(&Event{ID: "e1", Path: "test", Payload: map[string]any{}}) + + ttl := mr.TTL(redisBufferKey) + if ttl <= 0 { + t.Fatalf("expected positive TTL on buffer key, got %v", ttl) + } + if ttl > 24*time.Hour { + t.Fatalf("expected TTL <= 24h, got %v", ttl) + } +} + func TestNewRedisBackend_success(t *testing.T) { mr := miniredis.RunT(t) backend, err := NewRedisBackend("redis://"+mr.Addr(), 100) diff --git a/broker.go b/broker.go index 60b6bad..91e9cf9 100644 --- a/broker.go +++ b/broker.go @@ -85,6 +85,9 @@ func (b *Broker) Subscribe(path string, lastEventID string) (<-chan *Event, func break } } + if len(b.subscribers[path]) == 0 { + delete(b.subscribers, path) + } func() { defer func() { recover() }() close(ch) diff --git a/broker_test.go b/broker_test.go index 871f5f8..9af2da9 100644 --- a/broker_test.go +++ b/broker_test.go @@ -132,6 +132,29 @@ func TestBroker_unsubscribeStopsDelivery(t *testing.T) { } } +func TestBroker_unsubscribeCleansUpEmptyPath(t *testing.T) { + b, cancel := newTestBroker(100) + defer cancel() + + _, unsub := b.Subscribe("ephemeral/path", "") + + b.mu.RLock() + _, exists := b.subscribers["ephemeral/path"] + b.mu.RUnlock() + if !exists { + t.Fatal("expected subscriber map entry to exist") + } + + unsub() + + b.mu.RLock() + _, exists = b.subscribers["ephemeral/path"] + b.mu.RUnlock() + if exists { + t.Fatal("expected subscriber map entry to be deleted after last unsub") + } +} + func TestBroker_ringBufferWraps(t *testing.T) { b, cancel := newTestBroker(5) defer cancel()