fix: Replicas can now be set for streams

Author: aholstenson

internal/events/streams.go

@@ -166,7 +166,11 @@ func (m *Manager) EnsureStream(ctx context.Context, config *StreamConfig) (*Stre
 		streamConfig.Sources = sources
 	}
 
-	if config.Replicas != nil && *config.Replicas == 0 {
+	if config.Replicas != nil {
+		if *config.Replicas <= 0 {
+			return nil, errors.New("replicas must be greater than 0")
+		}
+
 		streamConfig.Replicas = int(*config.Replicas)
 	} else {
 		// TODO: We may want to have a config value in the Windshift server that sets the default value

internal/events/streams_test.go

@@ -1799,5 +1799,47 @@ var _ = Describe("Streams", func() {
 			})
 			Expect(err).To(HaveOccurred())
 		})
+
+		It("replicas defaults to 1", func(ctx context.Context) {
+			_, err := js.Stream(ctx, "test")
+			Expect(err).To(MatchError(jetstream.ErrStreamNotFound))
+
+			_, err = manager.EnsureStream(ctx, &events.StreamConfig{
+				Name: "test",
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			stream, err := js.Stream(ctx, "test")
+			Expect(err).ToNot(HaveOccurred())
+			Expect(stream.CachedInfo().Config.Replicas).To(Equal(1))
+		})
+
+		It("specifying zero replicas errors", func(ctx context.Context) {
+			_, err := js.Stream(ctx, "test")
+			Expect(err).To(MatchError(jetstream.ErrStreamNotFound))
+
+			r := uint(0)
+			_, err = manager.EnsureStream(ctx, &events.StreamConfig{
+				Name:     "test",
+				Replicas: &r,
+			})
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("can create stream with 1 replica", func(ctx context.Context) {
+			_, err := js.Stream(ctx, "test")
+			Expect(err).To(MatchError(jetstream.ErrStreamNotFound))
+
+			r := uint(1)
+			_, err = manager.EnsureStream(ctx, &events.StreamConfig{
+				Name:     "test",
+				Replicas: &r,
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			stream, err := js.Stream(ctx, "test")
+			Expect(err).ToNot(HaveOccurred())
+			Expect(stream.CachedInfo().Config.Replicas).To(Equal(1))
+		})
 	})
 })