Add compression

Signed-off-by: Tomasz Pietrek <tomasz@nats.io>

controllers/jetstream/stream.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	jsm "github.com/nats-io/jsm.go"
+	"github.com/nats-io/jsm.go/api"
 	jsmapi "github.com/nats-io/jsm.go/api"
 	apis "github.com/nats-io/nack/pkg/jetstream/apis/jetstream/v1beta2"
 	typed "github.com/nats-io/nack/pkg/jetstream/generated/clientset/versioned/typed/jetstream/v1beta2"
@@ -337,6 +338,13 @@ func createStream(ctx context.Context, c jsmClient, spec apis.StreamSpec) (err e
 		opts = append(opts, jsm.DiscardNew())
 	}
 
+	switch spec.Compression {
+	case "s2":
+		opts = append(opts, jsm.Compression(api.S2Compression))
+	case "none":
+		opts = append(opts, jsm.Compression(api.NoCompression))
+	}
+
 	if spec.NoAck {
 		opts = append(opts, jsm.NoAck())
 	}
@@ -492,6 +500,13 @@ func updateStream(ctx context.Context, c jsmClient, spec apis.StreamSpec) (err e
 		config.Sources[i] = jss
 	}
 
+	switch spec.Compression {
+	case "s2":
+		config.Compression = api.S2Compression
+	case "none":
+		config.Compression = api.NoCompression
+	}
+
 	return js.UpdateConfiguration(config)
 }
 

deploy/crds.yml

@@ -193,6 +193,13 @@ spec:
                 description: Sequence number from which the Stream will start.
                 type: number
                 default: 0
+              compression:
+                description: Stream specific compression.
+                type: string
+                enum:
+                - s2
+                - none
+                default: ''
               preventDelete:
                 description: When true, the managed Stream will not be deleted when the resource is deleted
                 type: boolean

pkg/jetstream/apis/jetstream/v1beta2/streamtypes.go

@@ -47,6 +47,7 @@ type StreamSpec struct {
 	Replicas          int              `json:"replicas"`
 	Republish         *RePublish       `json:"republish"`
 	FirstSequence     uint64           `json:"firstSequence"`
+	Compression       string           `json:"compression"`
 	Retention         string           `json:"retention"`
 	Servers           []string         `json:"servers"`
 	Sources           []*StreamSource  `json:"sources"`