PR Feedback: use Nats original config

Signed-off-by: Taleb Zeghmi

api/jsonschema/schema.json

@@ -497,7 +497,7 @@
           "type": "array"
         },
         "streamConfig": {
-          "description": "Optional configuration for the streams to be created in this JetStream service, if specified, it will be merged with the default configuration in controller-config. It accepts a YAML format configuration, available fields include, \"maxBytes\", \"maxMsgs\", \"maxAge\" (e.g. 72h), \"replicas\" (1, 3, 5), \"duplicates\" (e.g. 5m), \"retention\" (e.g. RetentionPolicy (default) or InterestPolicy), \"Discard\" (e.g. DiscardOld (default) or DiscardNew).",
+          "description": "Optional configuration for the streams to be created in this JetStream service, if specified, it will be merged with the default configuration in controller-config. It accepts a YAML format configuration, available fields include, \"maxBytes\", \"maxMsgs\", \"maxAge\" (e.g. 72h), \"replicas\" (1, 3, 5), \"duplicates\" (e.g. 5m), \"retention\" (e.g. 0: Limits (default), 1: Interest, 2: WorkQueue), \"Discard\" (e.g. 0: DiscardOld (default), 1: DiscardNew).",
           "type": "string"
         },
         "tolerations": {

eventbus/jetstream/base/jetstream.go

@@ -127,16 +127,16 @@ func (stream *Jetstream) CreateStream(conn *JetstreamConnection) error {
 		return err
 	}
 
-	v.SetDefault("retention", "LimitsPolicy")
-	v.SetDefault("discard", "DiscardOld")
+	v.SetDefault("retention", 0) // Limits
+	v.SetDefault("discard", 0)   // DiscardOld
 
-	retentionPolicy, err := getRetentionPolicy(v.GetString("retention"))
+	retentionPolicy, err := intToRetentionPolicy(v.GetInt("retention"))
 	if err != nil {
 		stream.Logger.Errorf("invalid retention policy: %s, error: %v", retentionPolicy, err)
 		return err
 	}
 
-	discardPolicy, err := getDiscardPolicy(v.GetString("discard"))
+	discardPolicy, err := intToDiscardPolicy(v.GetInt("discard"))
 	if err != nil {
 		stream.Logger.Errorf("invalid discard policy: %s, error: %v", discardPolicy, err)
 		return err
@@ -173,24 +173,17 @@ func (stream *Jetstream) CreateStream(conn *JetstreamConnection) error {
 	return nil
 }
 
-func getRetentionPolicy(policy string) (nats.RetentionPolicy, error) {
-	switch policy {
-	case "LimitsPolicy":
-		return nats.LimitsPolicy, nil
-	case "InterestPolicy":
-		return nats.InterestPolicy, nil
-	default:
-		return -1, fmt.Errorf("valid values are 'LimitsPolicy' or 'InterestPolicy'")
+func intToRetentionPolicy(i int) (nats.RetentionPolicy, error) {
+	if i < 0 || i > int(nats.WorkQueuePolicy) {
+		// Handle invalid value, return a default value or panic
+		return -1, fmt.Errorf("invalid int for RetentionPolicy: %d", i)
 	}
+	return nats.RetentionPolicy(i), nil
 }
 
-func getDiscardPolicy(policy string) (nats.DiscardPolicy, error) {
-	switch policy {
-	case "DiscardOld":
-		return nats.DiscardOld, nil
-	case "DiscardNew":
-		return nats.DiscardNew, nil
-	default:
-		return -1, fmt.Errorf("valid values are 'DiscardOld' or 'DiscardNew'")
+func intToDiscardPolicy(i int) (nats.DiscardPolicy, error) {
+	if i < 0 || i > int(nats.DiscardNew) {
+		return -1, fmt.Errorf("invalid int for DiscardPolicy: %d", i)
 	}
+	return nats.DiscardPolicy(i), nil
 }

manifests/base/controller-manager/controller-config.yaml

@@ -25,9 +25,10 @@ data:
           maxAge: 168h
           maxBytes: -1
           replicas: 3
-          duplicates: 300s
-          retention: LimitsPolicy
-          discard: DiscardOld
+          duplicates:           # 0: Limits, 1: Interest, 2: WorkQueue
+          retention: 0
+          # 0: DiscardOld, 1: DiscardNew
+          discard: 0
         versions:
         - version: latest
           natsImage: nats:2.10.10

manifests/install.yaml

@@ -330,8 +330,10 @@ data:
           maxBytes: -1
           replicas: 3
           duplicates: 300s
-          retention: LimitsPolicy
-          discard: DiscardOld
+          # 0: Limits, 1: Interest, 2: WorkQueue
+          retention: 0
+          # 0: DiscardOld, 1: DiscardNew
+          discard: 0
         versions:
         - version: latest
           natsImage: nats:2.10.10

manifests/namespace-install.yaml

@@ -250,8 +250,10 @@ data:
           maxBytes: -1
           replicas: 3
           duplicates: 300s
-          retention: LimitsPolicy
-          discard: DiscardOld
+          # 0: Limits, 1: Interest, 2: WorkQueue
+          retention: 0
+          # 0: DiscardOld, 1: DiscardNew
+          discard: 0
         versions:
         - version: latest
           natsImage: nats:2.10.10

pkg/apis/eventbus/v1alpha1/jetstream_eventbus.go

@@ -80,7 +80,7 @@ type JetStreamBus struct {
 	StartArgs []string `json:"startArgs,omitempty" protobuf:"bytes,17,rep,name=startArgs"`
 	// Optional configuration for the streams to be created in this JetStream service, if specified, it will be merged with the default configuration in controller-config.
 	// It accepts a YAML format configuration, available fields include, "maxBytes", "maxMsgs", "maxAge" (e.g. 72h), "replicas" (1, 3, 5), "duplicates" (e.g. 5m),
-	// "retention" (e.g. RetentionPolicy (default) or InterestPolicy), "Discard" (e.g. DiscardOld (default) or DiscardNew).
+	// "retention" (e.g. 0: Limits (default), 1: Interest, 2: WorkQueue), "Discard" (e.g. 0: DiscardOld (default), 1: DiscardNew).
 	// +optional
 	StreamConfig *string `json:"streamConfig,omitempty" protobuf:"bytes,18,opt,name=streamConfig"`
 	// Maximum number of bytes in a message payload, 0 means unlimited. Defaults to 1MB