[FEAT] kv and os compression

 Fix #624

jetstream/kv.ts

@@ -61,6 +61,7 @@ import {
   Republish,
   RetentionPolicy,
   StorageType,
+  StoreCompression,
   StreamConfig,
   StreamInfo,
   StreamSource,
@@ -270,6 +271,9 @@ export class Bucket implements KV, KvRemove {
     if (opts.metadata) {
       sc.metadata = opts.metadata;
     }
+    if (opts.compression) {
+      sc.compression = opts.compression;
+    }
 
     const nci = (this.js as unknown as { nc: NatsConnectionImpl }).nc;
     const have = nci.getServerVersion();
@@ -943,6 +947,10 @@ export class KvStatusImpl implements KvStatus {
   get metadata(): Record<string, string> {
     return this.si.config.metadata ?? {};
   }
+
+  get compression(): StoreCompression {
+    return this.si.config.compression || StoreCompression.None;
+  }
 }
 
 class KvStoredEntryImpl implements KvEntry {

jetstream/objectstore.ts

@@ -48,6 +48,7 @@ import {
   DiscardPolicy,
   PurgeResponse,
   StorageType,
+  StoreCompression,
   StreamConfig,
   StreamInfo,
   StreamInfoRequestOptions,
@@ -105,6 +106,10 @@ export class ObjectStoreStatusImpl implements ObjectStoreStatus {
   get metadata(): Record<string, string> | undefined {
     return this.si.config.metadata;
   }
+
+  get compression(): StoreCompression {
+    return this.si.config.compression || StoreCompression.None;
+  }
 }
 
 export type ServerObjectStoreMeta = {
@@ -789,6 +794,9 @@ export class ObjectStoreImpl implements ObjectStore {
     if (opts.metadata) {
       sc.metadata = opts.metadata;
     }
+    if (opts.compression) {
+      sc.compression = opts.compression;
+    }
 
     try {
       await this.jsm.streams.info(sc.name);

jetstream/tests/kv_test.ts

@@ -36,6 +36,7 @@ import {
   KvOptions,
   nanos,
   StorageType,
+  StoreCompression,
 } from "../mod.ts";
 
 import {
@@ -1584,13 +1585,11 @@ Deno.test("kv - mirror cross domain", async () => {
     const buf: KvEntry[] = [];
     m.set(bucket, buf);
 
-    const done = (async () => {
+    return (async () => {
       for await (const e of iter) {
         buf.push(e);
       }
     })().then();
-
-    return done;
   }
 
   watch(kv, "test", "name").then();
@@ -1932,3 +1931,18 @@ Deno.test("kv - republish header handling", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("kv - compression", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const js = nc.jetstream();
+  const s2 = await js.views.kv("compressed", {
+    compression: StoreCompression.S2,
+  });
+  let status = await s2.status();
+  assertEquals(status.compression, StoreCompression.S2);
+
+  const none = await js.views.kv("none");
+  status = await none.status();
+  assertEquals(status.compression, StoreCompression.None);
+  await cleanup(ns, nc);
+});

jetstream/tests/objectstore_test.ts

@@ -28,7 +28,12 @@ import {
 } from "https://deno.land/[email protected]/assert/mod.ts";
 import { DataBuffer } from "../../nats-base-client/databuffer.ts";
 import { crypto } from "https://deno.land/[email protected]/crypto/mod.ts";
-import { ObjectInfo, ObjectStoreMeta, StorageType } from "../mod.ts";
+import {
+  ObjectInfo,
+  ObjectStoreMeta,
+  StorageType,
+  StoreCompression,
+} from "../mod.ts";
 import { Empty, headers, nanos, StringCodec } from "../../src/mod.ts";
 import { equals } from "https://deno.land/[email protected]/bytes/mod.ts";
 import { SHA256 } from "../../nats-base-client/sha256.js";
@@ -1099,3 +1104,18 @@ Deno.test("objectstore - stream metadata and entry metadata", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("os - compression", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const js = nc.jetstream();
+  const s2 = await js.views.os("compressed", {
+    compression: StoreCompression.S2,
+  });
+  let status = await s2.status();
+  assertEquals(status.compression, StoreCompression.S2);
+
+  const none = await js.views.os("none");
+  status = await none.status();
+  assertEquals(status.compression, StoreCompression.None);
+  await cleanup(ns, nc);
+});

jetstream/types.ts

@@ -47,6 +47,7 @@ import {
   ReplayPolicy,
   Republish,
   StorageType,
+  StoreCompression,
   StreamAlternate,
   StreamConfig,
   StreamInfo,
@@ -1017,6 +1018,12 @@ export interface KvLimits {
    * FIXME: remove this on 1.8
    */
   backingStore: StorageType;
+
+  /**
+   * Sets the compression level of the KV. This feature is only supported in
+   * servers 2.10.x and better.
+   */
+  compression?: StoreCompression;
 }
 
 export interface KvStatus extends KvLimits {
@@ -1365,6 +1372,11 @@ export type ObjectStoreStatus = {
    * 2.10.x and better.
    */
   metadata?: Record<string, string> | undefined;
+  /**
+   * Compression level of the stream. This feature is only supported in
+   * servers 2.10.x and better.
+   */
+  compression: StoreCompression;
 };
 /**
  * @deprecated {@link ObjectStoreStatus}
@@ -1402,6 +1414,11 @@ export type ObjectStoreOptions = {
    * 2.10.x and better.
    */
   metadata?: Record<string, string>;
+  /**
+   * Sets the compression level of the stream. This feature is only supported in
+   * servers 2.10.x and better.
+   */
+  compression?: StoreCompression;
 };
 /**
  * An object that allows reading the object stored under a specified name.