Close pubsub topic when broadcaster closes

Closes #227
ipfs · Nov 20, 2024 · 4bb6115 · 4bb6115
1 parent c705d01
commit 4bb6115
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/pubsub_broadcaster.go b/pubsub_broadcaster.go
@@ -39,6 +39,13 @@ func NewPubSubBroadcaster(ctx context.Context, psub *pubsub.PubSub, topic string
 	go func(ctx context.Context, subs *pubsub.Subscription) {
 		<-ctx.Done()
 		subs.Cancel()
+		// subs.Next returns error when subscription closed. Subscription must
+		// be closed before psubTopic can be closed.
+		var err error
+		for err == nil {
+			_, err = subs.Next(ctx)
+		}
+		psubTopic.Close()
 	}(ctx, subs)
 
 	return &PubSubBroadcaster{