More napi readme links. Fix npm openKv return for native. Fix unraw w…

…atch stream.
skymethod · Dec 2, 2023 · 3a2c9f9 · 3a2c9f9
1 parent 5f49b7c
commit 3a2c9f9
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 20 deletions.
diff --git a/napi/README.md b/napi/README.md
@@ -3,8 +3,8 @@
 ![https://github.com/skymethod/kv-connect-kit/actions](https://github.com/skymethod/kv-connect-kit/workflows/npm/badge.svg)
 
 A [Deno KV](https://deno.com/kv) client library optimized for Node.js.
- - Access Deno Deploy remote databases (or any endpoint implementing the open [KV Connect](https://github.com/denoland/denokv/blob/main/proto/kv-connect.md) protocol) on Node, Bun, the browser, or any JavaScript environment.
- - Create local KV databases backed by [SQLite](https://www.sqlite.org/index.html), using optimized native NAPI packages for Node - compatible with DBs created by Deno itself.
+ - Access [Deno Deploy](https://deno.com/deploy) remote databases (or any endpoint implementing the open [KV Connect](https://github.com/denoland/denokv/blob/main/proto/kv-connect.md) protocol) on Node, Bun, the browser, or any JavaScript environment.
+ - Create local KV databases backed by [SQLite](https://www.sqlite.org/index.html), using optimized native [NAPI](https://nodejs.org/docs/latest-v18.x/api/n-api.html) packages for Node - compatible with DBs created by Deno itself.
  - Create ephemeral in-memory KV instances backed by SQLite memory files or by a lightweight JS-only implementation for testing.
  - Zero JS dependencies, architecture-specific native code for SQLite backend (see below)
  - Simply call the exported `openKv` function (equiv to [`Deno.openKv`](https://deno.land/api?s=Deno.openKv&unstable)) with a url or local path to get started!

diff --git a/src/npm.ts b/src/npm.ts
@@ -1,5 +1,6 @@
 import { check, checkOptionalBoolean, checkOptionalString, checkRecord } from './check.ts';
 import { makeInMemoryService } from './in_memory.ts';
+import { Kv } from './kv_types.ts';
 import { DecodeV8, EncodeV8 } from './kv_util.ts';
 import { isNapiInterface, makeNapiBasedService } from './napi_based.ts';
 import { makeNativeService } from './native.ts';
@@ -23,7 +24,7 @@ export type KvImplementation = 'in-memory' | 'sqlite' | 'remote';
  *
  * When no path is provided, this will use an ephemeral in-memory implementation.
  */
-export async function openKv(path?: string, opts: Record<string, unknown> & { debug?: boolean, implementation?: KvImplementation } = {}) {
+export async function openKv(path?: string, opts: Record<string, unknown> & { debug?: boolean, implementation?: KvImplementation } = {}): Promise<Kv> {
     checkOptionalString('path', path);
     checkRecord('opts', opts);
     checkOptionalBoolean('opts.debug', opts.debug);
@@ -35,7 +36,7 @@ export async function openKv(path?: string, opts: Record<string, unknown> & { de
     if ('Deno' in globalThis && !implementation) {
         // deno-lint-ignore no-explicit-any
         const { openKv } = (globalThis as any).Deno;
-        if (typeof openKv === 'function') return makeNativeService();
+        if (typeof openKv === 'function') return makeNativeService().openKv(path);
     }
 
     // use in-memory implementation if no path provided

diff --git a/src/unraw_watch_stream.ts b/src/unraw_watch_stream.ts
@@ -7,7 +7,6 @@ export function makeUnrawWatchStream(rawWatchStream: ReadableStream<KvEntryMaybe
     let latest: KvEntryMaybe<unknown>[] | undefined;
     let signal = defer<void>();
     let cancelled = false;
-    let n = 1;
     return new ReadableStream({
         start(controller) {
             (async () => {
@@ -26,28 +25,34 @@ export function makeUnrawWatchStream(rawWatchStream: ReadableStream<KvEntryMaybe
             })();
         },
         async pull(controller) {
-            if (n++ === 1) return; // assume the first pull is not user-initiated
             if (!latest) await signal;
             if (!latest) return;
-            if (!pulled) {
-                pulled = latest;
-                controller.enqueue(pulled);
-                return;
-            }
-            let changed = false;
-            for (let i = 0; i < latest.length; i++) {
-                if (latest[i].versionstamp === pulled[i].versionstamp) continue;
-                changed = true;
-                break;
-            }
-            if (changed) {
-                pulled = latest;
-                controller.enqueue(pulled);
+            while (true) {
+                let changed = false;
+                if (pulled) {
+                    for (let i = 0; i < latest.length; i++) {
+                        if (latest[i].versionstamp === pulled[i].versionstamp) continue;
+                        changed = true;
+                        break;
+                    }
+                } else {
+                    pulled = latest;
+                    changed = pulled.some(v => v.versionstamp !== null);
+                }
+                if (changed) {
+                    pulled = latest;
+                    controller.enqueue(pulled);
+                    return;
+                } else {
+                    await signal;
+                }
             }
         },
         async cancel() {
             cancelled = true;
             await onCancel();
         },
+    }, { 
+        highWaterMark: 0 // ensure all pulls are user-initiated
     });
 }