🐛 Reconfigure typing + presence streams upon reconnection.

globekeeper · Jul 17, 2024 · 21a9a50 · 21a9a50
1 parent cd43a34
commit 21a9a50
Showing 1 changed file with 8 additions and 8 deletions.
diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go
@@ -99,15 +99,13 @@ func setupNATS(process *process.ProcessContext, cfg *config.JetStream, nc *natsc
 			}),
 			natsclient.ReconnectHandler(func(_ *natsclient.Conn) {
 				logrus.Info("nats connection: client reconnected")
-				for _, stream := range []*nats.StreamConfig{
-					streams[6],
-					streams[10],
-				} {
-					err = configureStream(stream, cfg, s)
-					if err != nil {
-						logrus.WithError(err).WithField("stream", stream.Name).Error("unable to configure a stream")
+				for _, stream := range streams {
+					if stream.Name == OutputTypingEvent || stream.Name == OutputPresenceEvent {
+						err = configureStream(stream, cfg, s)
+						if err != nil {
+							logrus.WithError(err).WithField("stream", stream.Name).Error("unable to configure a stream")
+						}
 					}
-
 				}
 			}),
 			natsclient.ClosedHandler(func(_ *natsclient.Conn) {
@@ -262,6 +260,7 @@ func setupNATS(process *process.ProcessContext, cfg *config.JetStream, nc *natsc
 func configureStream(stream *nats.StreamConfig, cfg *config.JetStream, s nats.JetStreamContext) error {
 	name := cfg.Prefixed(stream.Name)
 	info, err := s.StreamInfo(name)
+	logrus.Infof("nats: configuring stream %s", stream.Name)
 	if err != nil && err != natsclient.ErrStreamNotFound {
 		return fmt.Errorf("get stream info: %w", err)
 	}
@@ -301,6 +300,7 @@ func configureStream(stream *nats.StreamConfig, cfg *config.JetStream, s nats.Je
 		namespaced.Name = name
 		namespaced.Subjects = subjects
 		if _, err = s.AddStream(&namespaced); err != nil {
+			logrus.WithError(err).Errorf("nats: failed to add stream %s", stream.Name)
 			return fmt.Errorf("add stream: %w", err)
 		}
 		logrus.Infof("stream created: %s", stream.Name)