Added useSyncStreams hook that allows you to manage a variable number of sync streams.

Author: Chriztiaan

.changeset/flat-jars-do.md

@@ -0,0 +1,5 @@
+---
+'@powersync/react': minor
+---
+
+Added `useSyncStreams` hook that allows you to manage a variable number of sync streams, compared to the existing `useSyncStream` hook intended for the one and only one use case.

packages/react/src/hooks/streams.ts

@@ -65,6 +65,56 @@ export function useSyncStream(options: UseSyncStreamOptions): SyncStreamStatus |
   return status.forStream(stream) ?? null;
 }
 
+/**
+ * Creates multiple PowerSync stream subscriptions. Subscriptions are kept alive as long as the
+ * React component calling this function. When it unmounts, or when the streams array contents
+ * change, all previous subscriptions are unsubscribed before new ones are created.
+ */
+export function useSyncStreams(streamOptions: UseSyncStreamOptions[]): SyncStreamStatus[] {
+  const db = usePowerSync();
+  const status = useStatus();
+
+  const stringifiedOptions = useMemo(() => JSON.stringify(streamOptions), [streamOptions]);
+  const syncStreams = useMemo(
+    () =>
+      streamOptions.map((options) => {
+        return {
+          stream: db.syncStream(options.name, options.parameters ?? undefined),
+          options
+        };
+      }),
+    [stringifiedOptions]
+  );
+
+  useEffect(() => {
+    let active = true;
+    const resolvedSubs: SyncStreamSubscription[] = [];
+
+    for (const entry of syncStreams) {
+      entry.stream.subscribe(entry.options).then((sub) => {
+        if (active) {
+          resolvedSubs.push(sub);
+        } else {
+          // The cleanup function already ran, unsubscribe immediately.
+          sub.unsubscribe();
+        }
+      });
+    }
+
+    return () => {
+      active = false;
+      for (const sub of resolvedSubs) {
+        sub.unsubscribe();
+      }
+    };
+  }, [stringifiedOptions]);
+
+  return useMemo(
+    () => syncStreams.map((entry) => status.forStream(entry.stream) ?? null),
+    [status, stringifiedOptions]
+  );
+}
+
 /**
  * Returns `true` once all streams in the array have synced at least once.
  */

packages/react/tests/streams.test.tsx

@@ -4,7 +4,7 @@ import React, { act, useSyncExternalStore } from 'react';
 import { AbstractPowerSyncDatabase, ConnectionManager, SyncStatus } from '@powersync/common';
 import { openPowerSync } from './utils';
 import { PowerSyncContext } from '../src/hooks/PowerSyncContext';
-import { useSyncStream, UseSyncStreamOptions } from '../src/hooks/streams';
+import { useSyncStream, useSyncStreams, UseSyncStreamOptions } from '../src/hooks/streams';
 import { useQuery } from '../src/hooks/watched/useQuery';
 import { QuerySyncStreamOptions } from '../src/hooks/watched/watch-types';
 
@@ -142,6 +142,102 @@ describe('stream hooks', () => {
         await waitFor(() => expect(currentStreams()).toHaveLength(0), { timeout: 1000, interval: 100 });
       });
 
+      it('useSyncStreams subscribes and unsubscribes', async () => {
+        expect(currentStreams()).toStrictEqual([]);
+
+        const { result, unmount } = renderHook(() => useSyncStreams([{ name: 'a' }, { name: 'b' }]), {
+          wrapper: testWrapper
+        });
+        expect(result.current).toStrictEqual([null, null]);
+        await waitFor(() => expect(currentStreams()).toHaveLength(2), { timeout: 1000, interval: 100 });
+        await waitFor(() => expect(result.current.every((s) => s !== null)).toBe(true), {
+          timeout: 1000,
+          interval: 100
+        });
+
+        unmount();
+        expect(currentStreams()).toStrictEqual([]);
+      });
+
+      it('useSyncStreams with cached instance', async () => {
+        const existingSubscription = await db.syncStream('a').subscribe();
+        await existingSubscription.unsubscribe();
+
+        const { result } = renderHook(() => useSyncStreams([{ name: 'a' }]), {
+          wrapper: testWrapper
+        });
+        expect(result.current[0]).not.toBeNull();
+      });
+
+      it('useSyncStreams handles array growing from 1 to 2 entries', async () => {
+        let streamOptions: UseSyncStreamOptions[] = [{ name: 'a' }];
+        let streamUpdateListeners: (() => void)[] = [];
+
+        const { result } = renderHook(
+          () => {
+            const options = useSyncExternalStore(
+              (cb) => {
+                streamUpdateListeners.push(cb);
+                return () => {
+                  const index = streamUpdateListeners.indexOf(cb);
+                  if (index != -1) {
+                    streamUpdateListeners.splice(index, 1);
+                  }
+                };
+              },
+              () => streamOptions
+            );
+            return useSyncStreams(options);
+          },
+          { wrapper: testWrapper }
+        );
+
+        await waitFor(() => expect(currentStreams()).toHaveLength(1), { timeout: 1000, interval: 100 });
+
+        // Grow to 2 entries
+        streamOptions = [{ name: 'a' }, { name: 'b' }];
+        act(() => streamUpdateListeners.forEach((cb) => cb()));
+
+        await waitFor(() => expect(currentStreams()).toHaveLength(2), { timeout: 1000, interval: 100 });
+        expect(result.current).toHaveLength(2);
+      });
+
+      it('useSyncStreams handles array shrinking from 2 to 1 entries', async () => {
+        let streamOptions: UseSyncStreamOptions[] = [{ name: 'a' }, { name: 'b' }];
+        let streamUpdateListeners: (() => void)[] = [];
+
+        const { result, unmount } = renderHook(
+          () => {
+            const options = useSyncExternalStore(
+              (cb) => {
+                streamUpdateListeners.push(cb);
+                return () => {
+                  const index = streamUpdateListeners.indexOf(cb);
+                  if (index != -1) {
+                    streamUpdateListeners.splice(index, 1);
+                  }
+                };
+              },
+              () => streamOptions
+            );
+            return useSyncStreams(options);
+          },
+          { wrapper: testWrapper }
+        );
+
+        await waitFor(() => expect(currentStreams()).toHaveLength(2), { timeout: 1000, interval: 100 });
+
+        // Shrink to 1 entry
+        streamOptions = [{ name: 'a' }];
+        act(() => streamUpdateListeners.forEach((cb) => cb()));
+
+        await waitFor(() => expect(currentStreams()).toHaveLength(1), { timeout: 1000, interval: 100 });
+        expect(result.current).toHaveLength(1);
+
+        unmount();
+        expect(currentStreams()).toStrictEqual([]);
+      });
+
       it('handles stream parameter changes', async () => {
         // Start without streams
         let streams: QuerySyncStreamOptions[] = [];