From 9e341501b22fe4d8e3ab6b5e2b6da7ee46e94fec Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Fri, 7 Jan 2022 16:46:49 +0000 Subject: [PATCH] Fix namespacing --- setup/jetstream/nats.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go index eec68d820..5d8906601 100644 --- a/setup/jetstream/nats.go +++ b/setup/jetstream/nats.go @@ -74,14 +74,22 @@ func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (nats.JetStreamContex logrus.WithError(err).Fatal("Unable to get stream info") } if info == nil { - stream.Subjects = []string{name} + // Define a default subject if not already specified. + if len(stream.Subjects) == 0 { + stream.Subjects = []string{name} + } + // If we're trying to keep everything in memory (e.g. unit tests) // then overwrite the storage policy. if cfg.InMemory { stream.Storage = nats.MemoryStorage } - if _, err = s.AddStream(stream); err != nil { + // Namespace the streams without modifying the original streams + // array, otherwise we end up with namespaces on namespaces. + namespaced := *stream + namespaced.Name = name + if _, err = s.AddStream(&namespaced); err != nil { logrus.WithError(err).WithField("stream", name).Fatal("Unable to add stream") } }