@tanstack/query-core - npm Package Compare versions

Comparing version 5.85.9 to 5.86.0

build/legacy/streamedQuery.cjs

@@ -28,5 +28,6 @@ "use strict";
 function streamedQuery({
-  queryFn,
+  streamFn,
   refetchMode = "reset",
-  maxChunks
+  reducer = (items, chunk) => (0, import_utils.addToEnd)(items, chunk),
+  initialValue = []
 }) {
@@ -44,4 +45,4 @@ return async (context) => {
     }
-    let result = [];
-    const stream = await queryFn(context);
+    let result = initialValue;
+    const stream = await streamFn(context);
     for await (const chunk of stream) {
@@ -54,8 +55,6 @@ if (context.signal.aborted) {
           context.queryKey,
-          (prev = []) => {
-            return (0, import_utils.addToEnd)(prev, chunk, maxChunks);
-          }
+          (prev) => reducer(prev === void 0 ? initialValue : prev, chunk)
         );
       }
-      result = (0, import_utils.addToEnd)(result, chunk, maxChunks);
+      result = reducer(result, chunk);
     }
@@ -62,0 +61,0 @@ if (isRefetch && refetchMode === "replace" && !context.signal.aborted) {

build/legacy/streamedQuery.cjs.map

@@ -1,1 +0,1 @@
-{"version":3,"sources":["../../src/streamedQuery.ts"],"sourcesContent":["import { addToEnd } from './utils'\nimport type { QueryFunction, QueryFunctionContext, QueryKey } from './types'\n\n/**\n * This is a helper function to create a query function that streams data from an AsyncIterable.\n * Data will be an Array of all the chunks received.\n * The query will be in a 'pending' state until the first chunk of data is received, but will go to 'success' after that.\n * The query will stay in fetchStatus 'fetching' until the stream ends.\n * @param queryFn - The function that returns an AsyncIterable to stream data from.\n * @param refetchMode - Defines how re-fetches are handled.\n * Defaults to `'reset'`, erases all data and puts the query back into `pending` state.\n * Set to `'append'` to append new data to the existing data.\n * Set to `'replace'` to write all data to the cache once the stream ends.\n * @param maxChunks - The maximum number of chunks to keep in the cache.\n * Defaults to `undefined`, meaning all chunks will be kept.\n * If `undefined` or `0`, the number of chunks is unlimited.\n * If the number of chunks exceeds this number, the oldest chunk will be removed.\n */\nexport function streamedQuery<\n  TQueryFnData = unknown,\n  TQueryKey extends QueryKey = QueryKey,\n>({\n  queryFn,\n  refetchMode = 'reset',\n  maxChunks,\n}: {\n  queryFn: (\n    context: QueryFunctionContext<TQueryKey>,\n  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>\n  refetchMode?: 'append' | 'reset' | 'replace'\n  maxChunks?: number\n}): QueryFunction<Array<TQueryFnData>, TQueryKey> {\n  return async (context) => {\n    const query = context.client\n      .getQueryCache()\n      .find({ queryKey: context.queryKey, exact: true })\n    const isRefetch = !!query && query.state.data !== undefined\n\n    if (isRefetch && refetchMode === 'reset') {\n      query.setState({\n        status: 'pending',\n        data: undefined,\n        error: null,\n        fetchStatus: 'fetching',\n      })\n    }\n\n    let result: Array<TQueryFnData> = []\n    const stream = await queryFn(context)\n\n    for await (const chunk of stream) {\n      if (context.signal.aborted) {\n        break\n      }\n\n      // don't append to the cache directly when replace-refetching\n      if (!isRefetch || refetchMode !== 'replace') {\n        context.client.setQueryData<Array<TQueryFnData>>(\n          context.queryKey,\n          (prev = []) => {\n            return addToEnd(prev, chunk, maxChunks)\n          },\n        )\n      }\n      result = addToEnd(result, chunk, maxChunks)\n    }\n\n    // finalize result: replace-refetching needs to write to the cache\n    if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {\n      context.client.setQueryData<Array<TQueryFnData>>(context.queryKey, result)\n    }\n\n    return context.client.getQueryData(context.queryKey)!\n  }\n}\n"],"mappings":";;;;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AAAA,mBAAyB;AAkBlB,SAAS,cAGd;AAAA,EACA;AAAA,EACA,cAAc;AAAA,EACd;AACF,GAMkD;AAChD,SAAO,OAAO,YAAY;AACxB,UAAM,QAAQ,QAAQ,OACnB,cAAc,EACd,KAAK,EAAE,UAAU,QAAQ,UAAU,OAAO,KAAK,CAAC;AACnD,UAAM,YAAY,CAAC,CAAC,SAAS,MAAM,MAAM,SAAS;AAElD,QAAI,aAAa,gBAAgB,SAAS;AACxC,YAAM,SAAS;AAAA,QACb,QAAQ;AAAA,QACR,MAAM;AAAA,QACN,OAAO;AAAA,QACP,aAAa;AAAA,MACf,CAAC;AAAA,IACH;AAEA,QAAI,SAA8B,CAAC;AACnC,UAAM,SAAS,MAAM,QAAQ,OAAO;AAEpC,qBAAiB,SAAS,QAAQ;AAChC,UAAI,QAAQ,OAAO,SAAS;AAC1B;AAAA,MACF;AAGA,UAAI,CAAC,aAAa,gBAAgB,WAAW;AAC3C,gBAAQ,OAAO;AAAA,UACb,QAAQ;AAAA,UACR,CAAC,OAAO,CAAC,MAAM;AACb,uBAAO,uBAAS,MAAM,OAAO,SAAS;AAAA,UACxC;AAAA,QACF;AAAA,MACF;AACA,mBAAS,uBAAS,QAAQ,OAAO,SAAS;AAAA,IAC5C;AAGA,QAAI,aAAa,gBAAgB,aAAa,CAAC,QAAQ,OAAO,SAAS;AACrE,cAAQ,OAAO,aAAkC,QAAQ,UAAU,MAAM;AAAA,IAC3E;AAEA,WAAO,QAAQ,OAAO,aAAa,QAAQ,QAAQ;AAAA,EACrD;AACF;","names":[]}
+{"version":3,"sources":["../../src/streamedQuery.ts"],"sourcesContent":["import { addToEnd } from './utils'\nimport type { QueryFunction, QueryFunctionContext, QueryKey } from './types'\n\ntype BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {\n  streamFn: (\n    context: QueryFunctionContext<TQueryKey>,\n  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>\n  refetchMode?: 'append' | 'reset' | 'replace'\n}\n\ntype SimpleStreamedQueryParams<\n  TQueryFnData,\n  TQueryKey extends QueryKey,\n> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {\n  reducer?: never\n  initialValue?: never\n}\n\ntype ReducibleStreamedQueryParams<\n  TQueryFnData,\n  TData,\n  TQueryKey extends QueryKey,\n> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {\n  reducer: (acc: TData, chunk: TQueryFnData) => TData\n  initialValue: TData\n}\n\ntype StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> =\n  | SimpleStreamedQueryParams<TQueryFnData, TQueryKey>\n  | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>\n\n/**\n * This is a helper function to create a query function that streams data from an AsyncIterable.\n * Data will be an Array of all the chunks received.\n * The query will be in a 'pending' state until the first chunk of data is received, but will go to 'success' after that.\n * The query will stay in fetchStatus 'fetching' until the stream ends.\n * @param queryFn - The function that returns an AsyncIterable to stream data from.\n * @param refetchMode - Defines how re-fetches are handled.\n * Defaults to `'reset'`, erases all data and puts the query back into `pending` state.\n * Set to `'append'` to append new data to the existing data.\n * Set to `'replace'` to write all data to the cache once the stream ends.\n * @param reducer - A function to reduce the streamed chunks into the final data.\n * Defaults to a function that appends chunks to the end of the array.\n * @param initialValue - Initial value to be used while the first chunk is being fetched.\n */\nexport function streamedQuery<\n  TQueryFnData = unknown,\n  TData = Array<TQueryFnData>,\n  TQueryKey extends QueryKey = QueryKey,\n>({\n  streamFn,\n  refetchMode = 'reset',\n  reducer = (items, chunk) =>\n    addToEnd(items as Array<TQueryFnData>, chunk) as TData,\n  initialValue = [] as TData,\n}: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<\n  TData,\n  TQueryKey\n> {\n  return async (context) => {\n    const query = context.client\n      .getQueryCache()\n      .find({ queryKey: context.queryKey, exact: true })\n    const isRefetch = !!query && query.state.data !== undefined\n    if (isRefetch && refetchMode === 'reset') {\n      query.setState({\n        status: 'pending',\n        data: undefined,\n        error: null,\n        fetchStatus: 'fetching',\n      })\n    }\n\n    let result = initialValue\n\n    const stream = await streamFn(context)\n\n    for await (const chunk of stream) {\n      if (context.signal.aborted) {\n        break\n      }\n\n      // don't append to the cache directly when replace-refetching\n      if (!isRefetch || refetchMode !== 'replace') {\n        context.client.setQueryData<TData>(context.queryKey, (prev) =>\n          reducer(prev === undefined ? initialValue : prev, chunk),\n        )\n      }\n      result = reducer(result, chunk)\n    }\n\n    // finalize result: replace-refetching needs to write to the cache\n    if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {\n      context.client.setQueryData<TData>(context.queryKey, result)\n    }\n\n    return context.client.getQueryData(context.queryKey)!\n  }\n}\n"],"mappings":";;;;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AAAA,mBAAyB;AA6ClB,SAAS,cAId;AAAA,EACA;AAAA,EACA,cAAc;AAAA,EACd,UAAU,CAAC,OAAO,cAChB,uBAAS,OAA8B,KAAK;AAAA,EAC9C,eAAe,CAAC;AAClB,GAGE;AACA,SAAO,OAAO,YAAY;AACxB,UAAM,QAAQ,QAAQ,OACnB,cAAc,EACd,KAAK,EAAE,UAAU,QAAQ,UAAU,OAAO,KAAK,CAAC;AACnD,UAAM,YAAY,CAAC,CAAC,SAAS,MAAM,MAAM,SAAS;AAClD,QAAI,aAAa,gBAAgB,SAAS;AACxC,YAAM,SAAS;AAAA,QACb,QAAQ;AAAA,QACR,MAAM;AAAA,QACN,OAAO;AAAA,QACP,aAAa;AAAA,MACf,CAAC;AAAA,IACH;AAEA,QAAI,SAAS;AAEb,UAAM,SAAS,MAAM,SAAS,OAAO;AAErC,qBAAiB,SAAS,QAAQ;AAChC,UAAI,QAAQ,OAAO,SAAS;AAC1B;AAAA,MACF;AAGA,UAAI,CAAC,aAAa,gBAAgB,WAAW;AAC3C,gBAAQ,OAAO;AAAA,UAAoB,QAAQ;AAAA,UAAU,CAAC,SACpD,QAAQ,SAAS,SAAY,eAAe,MAAM,KAAK;AAAA,QACzD;AAAA,MACF;AACA,eAAS,QAAQ,QAAQ,KAAK;AAAA,IAChC;AAGA,QAAI,aAAa,gBAAgB,aAAa,CAAC,QAAQ,OAAO,SAAS;AACrE,cAAQ,OAAO,aAAoB,QAAQ,UAAU,MAAM;AAAA,IAC7D;AAEA,WAAO,QAAQ,OAAO,aAAa,QAAQ,QAAQ;AAAA,EACrD;AACF;","names":[]}

build/legacy/streamedQuery.d.cts

@@ -5,2 +5,15 @@ import { I as QueryKey, a1 as QueryFunctionContext, Y as QueryFunction } from './hydration-BC7iBQD-.cjs';
 
+type BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {
+    streamFn: (context: QueryFunctionContext<TQueryKey>) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>;
+    refetchMode?: 'append' | 'reset' | 'replace';
+};
+type SimpleStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+    reducer?: never;
+    initialValue?: never;
+};
+type ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+    reducer: (acc: TData, chunk: TQueryFnData) => TData;
+    initialValue: TData;
+};
+type StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> = SimpleStreamedQueryParams<TQueryFnData, TQueryKey> | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>;
 /**
@@ -16,13 +29,8 @@ * This is a helper function to create a query function that streams data from an AsyncIterable.
  * Set to `'replace'` to write all data to the cache once the stream ends.
- * @param maxChunks - The maximum number of chunks to keep in the cache.
- * Defaults to `undefined`, meaning all chunks will be kept.
- * If `undefined` or `0`, the number of chunks is unlimited.
- * If the number of chunks exceeds this number, the oldest chunk will be removed.
+ * @param reducer - A function to reduce the streamed chunks into the final data.
+ * Defaults to a function that appends chunks to the end of the array.
+ * @param initialValue - Initial value to be used while the first chunk is being fetched.
  */
-declare function streamedQuery<TQueryFnData = unknown, TQueryKey extends QueryKey = QueryKey>({ queryFn, refetchMode, maxChunks, }: {
-    queryFn: (context: QueryFunctionContext<TQueryKey>) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>;
-    refetchMode?: 'append' | 'reset' | 'replace';
-    maxChunks?: number;
-}): QueryFunction<Array<TQueryFnData>, TQueryKey>;
+declare function streamedQuery<TQueryFnData = unknown, TData = Array<TQueryFnData>, TQueryKey extends QueryKey = QueryKey>({ streamFn, refetchMode, reducer, initialValue, }: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<TData, TQueryKey>;
 
 export { streamedQuery };

build/legacy/streamedQuery.d.ts

@@ -5,2 +5,15 @@ import { I as QueryKey, a1 as QueryFunctionContext, Y as QueryFunction } from './hydration-BpLOP9Dw.js';
 
+type BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {
+    streamFn: (context: QueryFunctionContext<TQueryKey>) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>;
+    refetchMode?: 'append' | 'reset' | 'replace';
+};
+type SimpleStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+    reducer?: never;
+    initialValue?: never;
+};
+type ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+    reducer: (acc: TData, chunk: TQueryFnData) => TData;
+    initialValue: TData;
+};
+type StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> = SimpleStreamedQueryParams<TQueryFnData, TQueryKey> | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>;
 /**
@@ -16,13 +29,8 @@ * This is a helper function to create a query function that streams data from an AsyncIterable.
  * Set to `'replace'` to write all data to the cache once the stream ends.
- * @param maxChunks - The maximum number of chunks to keep in the cache.
- * Defaults to `undefined`, meaning all chunks will be kept.
- * If `undefined` or `0`, the number of chunks is unlimited.
- * If the number of chunks exceeds this number, the oldest chunk will be removed.
+ * @param reducer - A function to reduce the streamed chunks into the final data.
+ * Defaults to a function that appends chunks to the end of the array.
+ * @param initialValue - Initial value to be used while the first chunk is being fetched.
  */
-declare function streamedQuery<TQueryFnData = unknown, TQueryKey extends QueryKey = QueryKey>({ queryFn, refetchMode, maxChunks, }: {
-    queryFn: (context: QueryFunctionContext<TQueryKey>) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>;
-    refetchMode?: 'append' | 'reset' | 'replace';
-    maxChunks?: number;
-}): QueryFunction<Array<TQueryFnData>, TQueryKey>;
+declare function streamedQuery<TQueryFnData = unknown, TData = Array<TQueryFnData>, TQueryKey extends QueryKey = QueryKey>({ streamFn, refetchMode, reducer, initialValue, }: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<TData, TQueryKey>;
 
 export { streamedQuery };

build/legacy/streamedQuery.js

@@ -6,5 +6,6 @@ import "./chunk-PXG64RU4.js";
 function streamedQuery({
-  queryFn,
+  streamFn,
   refetchMode = "reset",
-  maxChunks
+  reducer = (items, chunk) => addToEnd(items, chunk),
+  initialValue = []
 }) {
@@ -22,4 +23,4 @@ return async (context) => {
     }
-    let result = [];
-    const stream = await queryFn(context);
+    let result = initialValue;
+    const stream = await streamFn(context);
     for await (const chunk of stream) {
@@ -32,8 +33,6 @@ if (context.signal.aborted) {
           context.queryKey,
-          (prev = []) => {
-            return addToEnd(prev, chunk, maxChunks);
-          }
+          (prev) => reducer(prev === void 0 ? initialValue : prev, chunk)
         );
       }
-      result = addToEnd(result, chunk, maxChunks);
+      result = reducer(result, chunk);
     }
@@ -40,0 +39,0 @@ if (isRefetch && refetchMode === "replace" && !context.signal.aborted) {

build/legacy/streamedQuery.js.map

@@ -1,1 +0,1 @@
-{"version":3,"sources":["../../src/streamedQuery.ts"],"sourcesContent":["import { addToEnd } from './utils'\nimport type { QueryFunction, QueryFunctionContext, QueryKey } from './types'\n\n/**\n * This is a helper function to create a query function that streams data from an AsyncIterable.\n * Data will be an Array of all the chunks received.\n * The query will be in a 'pending' state until the first chunk of data is received, but will go to 'success' after that.\n * The query will stay in fetchStatus 'fetching' until the stream ends.\n * @param queryFn - The function that returns an AsyncIterable to stream data from.\n * @param refetchMode - Defines how re-fetches are handled.\n * Defaults to `'reset'`, erases all data and puts the query back into `pending` state.\n * Set to `'append'` to append new data to the existing data.\n * Set to `'replace'` to write all data to the cache once the stream ends.\n * @param maxChunks - The maximum number of chunks to keep in the cache.\n * Defaults to `undefined`, meaning all chunks will be kept.\n * If `undefined` or `0`, the number of chunks is unlimited.\n * If the number of chunks exceeds this number, the oldest chunk will be removed.\n */\nexport function streamedQuery<\n  TQueryFnData = unknown,\n  TQueryKey extends QueryKey = QueryKey,\n>({\n  queryFn,\n  refetchMode = 'reset',\n  maxChunks,\n}: {\n  queryFn: (\n    context: QueryFunctionContext<TQueryKey>,\n  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>\n  refetchMode?: 'append' | 'reset' | 'replace'\n  maxChunks?: number\n}): QueryFunction<Array<TQueryFnData>, TQueryKey> {\n  return async (context) => {\n    const query = context.client\n      .getQueryCache()\n      .find({ queryKey: context.queryKey, exact: true })\n    const isRefetch = !!query && query.state.data !== undefined\n\n    if (isRefetch && refetchMode === 'reset') {\n      query.setState({\n        status: 'pending',\n        data: undefined,\n        error: null,\n        fetchStatus: 'fetching',\n      })\n    }\n\n    let result: Array<TQueryFnData> = []\n    const stream = await queryFn(context)\n\n    for await (const chunk of stream) {\n      if (context.signal.aborted) {\n        break\n      }\n\n      // don't append to the cache directly when replace-refetching\n      if (!isRefetch || refetchMode !== 'replace') {\n        context.client.setQueryData<Array<TQueryFnData>>(\n          context.queryKey,\n          (prev = []) => {\n            return addToEnd(prev, chunk, maxChunks)\n          },\n        )\n      }\n      result = addToEnd(result, chunk, maxChunks)\n    }\n\n    // finalize result: replace-refetching needs to write to the cache\n    if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {\n      context.client.setQueryData<Array<TQueryFnData>>(context.queryKey, result)\n    }\n\n    return context.client.getQueryData(context.queryKey)!\n  }\n}\n"],"mappings":";;;AAAA,SAAS,gBAAgB;AAkBlB,SAAS,cAGd;AAAA,EACA;AAAA,EACA,cAAc;AAAA,EACd;AACF,GAMkD;AAChD,SAAO,OAAO,YAAY;AACxB,UAAM,QAAQ,QAAQ,OACnB,cAAc,EACd,KAAK,EAAE,UAAU,QAAQ,UAAU,OAAO,KAAK,CAAC;AACnD,UAAM,YAAY,CAAC,CAAC,SAAS,MAAM,MAAM,SAAS;AAElD,QAAI,aAAa,gBAAgB,SAAS;AACxC,YAAM,SAAS;AAAA,QACb,QAAQ;AAAA,QACR,MAAM;AAAA,QACN,OAAO;AAAA,QACP,aAAa;AAAA,MACf,CAAC;AAAA,IACH;AAEA,QAAI,SAA8B,CAAC;AACnC,UAAM,SAAS,MAAM,QAAQ,OAAO;AAEpC,qBAAiB,SAAS,QAAQ;AAChC,UAAI,QAAQ,OAAO,SAAS;AAC1B;AAAA,MACF;AAGA,UAAI,CAAC,aAAa,gBAAgB,WAAW;AAC3C,gBAAQ,OAAO;AAAA,UACb,QAAQ;AAAA,UACR,CAAC,OAAO,CAAC,MAAM;AACb,mBAAO,SAAS,MAAM,OAAO,SAAS;AAAA,UACxC;AAAA,QACF;AAAA,MACF;AACA,eAAS,SAAS,QAAQ,OAAO,SAAS;AAAA,IAC5C;AAGA,QAAI,aAAa,gBAAgB,aAAa,CAAC,QAAQ,OAAO,SAAS;AACrE,cAAQ,OAAO,aAAkC,QAAQ,UAAU,MAAM;AAAA,IAC3E;AAEA,WAAO,QAAQ,OAAO,aAAa,QAAQ,QAAQ;AAAA,EACrD;AACF;","names":[]}
+{"version":3,"sources":["../../src/streamedQuery.ts"],"sourcesContent":["import { addToEnd } from './utils'\nimport type { QueryFunction, QueryFunctionContext, QueryKey } from './types'\n\ntype BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {\n  streamFn: (\n    context: QueryFunctionContext<TQueryKey>,\n  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>\n  refetchMode?: 'append' | 'reset' | 'replace'\n}\n\ntype SimpleStreamedQueryParams<\n  TQueryFnData,\n  TQueryKey extends QueryKey,\n> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {\n  reducer?: never\n  initialValue?: never\n}\n\ntype ReducibleStreamedQueryParams<\n  TQueryFnData,\n  TData,\n  TQueryKey extends QueryKey,\n> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {\n  reducer: (acc: TData, chunk: TQueryFnData) => TData\n  initialValue: TData\n}\n\ntype StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> =\n  | SimpleStreamedQueryParams<TQueryFnData, TQueryKey>\n  | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>\n\n/**\n * This is a helper function to create a query function that streams data from an AsyncIterable.\n * Data will be an Array of all the chunks received.\n * The query will be in a 'pending' state until the first chunk of data is received, but will go to 'success' after that.\n * The query will stay in fetchStatus 'fetching' until the stream ends.\n * @param queryFn - The function that returns an AsyncIterable to stream data from.\n * @param refetchMode - Defines how re-fetches are handled.\n * Defaults to `'reset'`, erases all data and puts the query back into `pending` state.\n * Set to `'append'` to append new data to the existing data.\n * Set to `'replace'` to write all data to the cache once the stream ends.\n * @param reducer - A function to reduce the streamed chunks into the final data.\n * Defaults to a function that appends chunks to the end of the array.\n * @param initialValue - Initial value to be used while the first chunk is being fetched.\n */\nexport function streamedQuery<\n  TQueryFnData = unknown,\n  TData = Array<TQueryFnData>,\n  TQueryKey extends QueryKey = QueryKey,\n>({\n  streamFn,\n  refetchMode = 'reset',\n  reducer = (items, chunk) =>\n    addToEnd(items as Array<TQueryFnData>, chunk) as TData,\n  initialValue = [] as TData,\n}: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<\n  TData,\n  TQueryKey\n> {\n  return async (context) => {\n    const query = context.client\n      .getQueryCache()\n      .find({ queryKey: context.queryKey, exact: true })\n    const isRefetch = !!query && query.state.data !== undefined\n    if (isRefetch && refetchMode === 'reset') {\n      query.setState({\n        status: 'pending',\n        data: undefined,\n        error: null,\n        fetchStatus: 'fetching',\n      })\n    }\n\n    let result = initialValue\n\n    const stream = await streamFn(context)\n\n    for await (const chunk of stream) {\n      if (context.signal.aborted) {\n        break\n      }\n\n      // don't append to the cache directly when replace-refetching\n      if (!isRefetch || refetchMode !== 'replace') {\n        context.client.setQueryData<TData>(context.queryKey, (prev) =>\n          reducer(prev === undefined ? initialValue : prev, chunk),\n        )\n      }\n      result = reducer(result, chunk)\n    }\n\n    // finalize result: replace-refetching needs to write to the cache\n    if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {\n      context.client.setQueryData<TData>(context.queryKey, result)\n    }\n\n    return context.client.getQueryData(context.queryKey)!\n  }\n}\n"],"mappings":";;;AAAA,SAAS,gBAAgB;AA6ClB,SAAS,cAId;AAAA,EACA;AAAA,EACA,cAAc;AAAA,EACd,UAAU,CAAC,OAAO,UAChB,SAAS,OAA8B,KAAK;AAAA,EAC9C,eAAe,CAAC;AAClB,GAGE;AACA,SAAO,OAAO,YAAY;AACxB,UAAM,QAAQ,QAAQ,OACnB,cAAc,EACd,KAAK,EAAE,UAAU,QAAQ,UAAU,OAAO,KAAK,CAAC;AACnD,UAAM,YAAY,CAAC,CAAC,SAAS,MAAM,MAAM,SAAS;AAClD,QAAI,aAAa,gBAAgB,SAAS;AACxC,YAAM,SAAS;AAAA,QACb,QAAQ;AAAA,QACR,MAAM;AAAA,QACN,OAAO;AAAA,QACP,aAAa;AAAA,MACf,CAAC;AAAA,IACH;AAEA,QAAI,SAAS;AAEb,UAAM,SAAS,MAAM,SAAS,OAAO;AAErC,qBAAiB,SAAS,QAAQ;AAChC,UAAI,QAAQ,OAAO,SAAS;AAC1B;AAAA,MACF;AAGA,UAAI,CAAC,aAAa,gBAAgB,WAAW;AAC3C,gBAAQ,OAAO;AAAA,UAAoB,QAAQ;AAAA,UAAU,CAAC,SACpD,QAAQ,SAAS,SAAY,eAAe,MAAM,KAAK;AAAA,QACzD;AAAA,MACF;AACA,eAAS,QAAQ,QAAQ,KAAK;AAAA,IAChC;AAGA,QAAI,aAAa,gBAAgB,aAAa,CAAC,QAAQ,OAAO,SAAS;AACrE,cAAQ,OAAO,aAAoB,QAAQ,UAAU,MAAM;AAAA,IAC7D;AAEA,WAAO,QAAQ,OAAO,aAAa,QAAQ,QAAQ;AAAA,EACrD;AACF;","names":[]}

build/modern/streamedQuery.cjs

@@ -28,5 +28,6 @@ "use strict";
 function streamedQuery({
-  queryFn,
+  streamFn,
   refetchMode = "reset",
-  maxChunks
+  reducer = (items, chunk) => (0, import_utils.addToEnd)(items, chunk),
+  initialValue = []
 }) {
@@ -44,4 +45,4 @@ return async (context) => {
     }
-    let result = [];
-    const stream = await queryFn(context);
+    let result = initialValue;
+    const stream = await streamFn(context);
     for await (const chunk of stream) {
@@ -54,8 +55,6 @@ if (context.signal.aborted) {
           context.queryKey,
-          (prev = []) => {
-            return (0, import_utils.addToEnd)(prev, chunk, maxChunks);
-          }
+          (prev) => reducer(prev === void 0 ? initialValue : prev, chunk)
         );
       }
-      result = (0, import_utils.addToEnd)(result, chunk, maxChunks);
+      result = reducer(result, chunk);
     }
@@ -62,0 +61,0 @@ if (isRefetch && refetchMode === "replace" && !context.signal.aborted) {

build/modern/streamedQuery.cjs.map

@@ -1,1 +0,1 @@
-{"version":3,"sources":["../../src/streamedQuery.ts"],"sourcesContent":["import { addToEnd } from './utils'\nimport type { QueryFunction, QueryFunctionContext, QueryKey } from './types'\n\n/**\n * This is a helper function to create a query function that streams data from an AsyncIterable.\n * Data will be an Array of all the chunks received.\n * The query will be in a 'pending' state until the first chunk of data is received, but will go to 'success' after that.\n * The query will stay in fetchStatus 'fetching' until the stream ends.\n * @param queryFn - The function that returns an AsyncIterable to stream data from.\n * @param refetchMode - Defines how re-fetches are handled.\n * Defaults to `'reset'`, erases all data and puts the query back into `pending` state.\n * Set to `'append'` to append new data to the existing data.\n * Set to `'replace'` to write all data to the cache once the stream ends.\n * @param maxChunks - The maximum number of chunks to keep in the cache.\n * Defaults to `undefined`, meaning all chunks will be kept.\n * If `undefined` or `0`, the number of chunks is unlimited.\n * If the number of chunks exceeds this number, the oldest chunk will be removed.\n */\nexport function streamedQuery<\n  TQueryFnData = unknown,\n  TQueryKey extends QueryKey = QueryKey,\n>({\n  queryFn,\n  refetchMode = 'reset',\n  maxChunks,\n}: {\n  queryFn: (\n    context: QueryFunctionContext<TQueryKey>,\n  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>\n  refetchMode?: 'append' | 'reset' | 'replace'\n  maxChunks?: number\n}): QueryFunction<Array<TQueryFnData>, TQueryKey> {\n  return async (context) => {\n    const query = context.client\n      .getQueryCache()\n      .find({ queryKey: context.queryKey, exact: true })\n    const isRefetch = !!query && query.state.data !== undefined\n\n    if (isRefetch && refetchMode === 'reset') {\n      query.setState({\n        status: 'pending',\n        data: undefined,\n        error: null,\n        fetchStatus: 'fetching',\n      })\n    }\n\n    let result: Array<TQueryFnData> = []\n    const stream = await queryFn(context)\n\n    for await (const chunk of stream) {\n      if (context.signal.aborted) {\n        break\n      }\n\n      // don't append to the cache directly when replace-refetching\n      if (!isRefetch || refetchMode !== 'replace') {\n        context.client.setQueryData<Array<TQueryFnData>>(\n          context.queryKey,\n          (prev = []) => {\n            return addToEnd(prev, chunk, maxChunks)\n          },\n        )\n      }\n      result = addToEnd(result, chunk, maxChunks)\n    }\n\n    // finalize result: replace-refetching needs to write to the cache\n    if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {\n      context.client.setQueryData<Array<TQueryFnData>>(context.queryKey, result)\n    }\n\n    return context.client.getQueryData(context.queryKey)!\n  }\n}\n"],"mappings":";;;;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AAAA,mBAAyB;AAkBlB,SAAS,cAGd;AAAA,EACA;AAAA,EACA,cAAc;AAAA,EACd;AACF,GAMkD;AAChD,SAAO,OAAO,YAAY;AACxB,UAAM,QAAQ,QAAQ,OACnB,cAAc,EACd,KAAK,EAAE,UAAU,QAAQ,UAAU,OAAO,KAAK,CAAC;AACnD,UAAM,YAAY,CAAC,CAAC,SAAS,MAAM,MAAM,SAAS;AAElD,QAAI,aAAa,gBAAgB,SAAS;AACxC,YAAM,SAAS;AAAA,QACb,QAAQ;AAAA,QACR,MAAM;AAAA,QACN,OAAO;AAAA,QACP,aAAa;AAAA,MACf,CAAC;AAAA,IACH;AAEA,QAAI,SAA8B,CAAC;AACnC,UAAM,SAAS,MAAM,QAAQ,OAAO;AAEpC,qBAAiB,SAAS,QAAQ;AAChC,UAAI,QAAQ,OAAO,SAAS;AAC1B;AAAA,MACF;AAGA,UAAI,CAAC,aAAa,gBAAgB,WAAW;AAC3C,gBAAQ,OAAO;AAAA,UACb,QAAQ;AAAA,UACR,CAAC,OAAO,CAAC,MAAM;AACb,uBAAO,uBAAS,MAAM,OAAO,SAAS;AAAA,UACxC;AAAA,QACF;AAAA,MACF;AACA,mBAAS,uBAAS,QAAQ,OAAO,SAAS;AAAA,IAC5C;AAGA,QAAI,aAAa,gBAAgB,aAAa,CAAC,QAAQ,OAAO,SAAS;AACrE,cAAQ,OAAO,aAAkC,QAAQ,UAAU,MAAM;AAAA,IAC3E;AAEA,WAAO,QAAQ,OAAO,aAAa,QAAQ,QAAQ;AAAA,EACrD;AACF;","names":[]}
+{"version":3,"sources":["../../src/streamedQuery.ts"],"sourcesContent":["import { addToEnd } from './utils'\nimport type { QueryFunction, QueryFunctionContext, QueryKey } from './types'\n\ntype BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {\n  streamFn: (\n    context: QueryFunctionContext<TQueryKey>,\n  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>\n  refetchMode?: 'append' | 'reset' | 'replace'\n}\n\ntype SimpleStreamedQueryParams<\n  TQueryFnData,\n  TQueryKey extends QueryKey,\n> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {\n  reducer?: never\n  initialValue?: never\n}\n\ntype ReducibleStreamedQueryParams<\n  TQueryFnData,\n  TData,\n  TQueryKey extends QueryKey,\n> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {\n  reducer: (acc: TData, chunk: TQueryFnData) => TData\n  initialValue: TData\n}\n\ntype StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> =\n  | SimpleStreamedQueryParams<TQueryFnData, TQueryKey>\n  | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>\n\n/**\n * This is a helper function to create a query function that streams data from an AsyncIterable.\n * Data will be an Array of all the chunks received.\n * The query will be in a 'pending' state until the first chunk of data is received, but will go to 'success' after that.\n * The query will stay in fetchStatus 'fetching' until the stream ends.\n * @param queryFn - The function that returns an AsyncIterable to stream data from.\n * @param refetchMode - Defines how re-fetches are handled.\n * Defaults to `'reset'`, erases all data and puts the query back into `pending` state.\n * Set to `'append'` to append new data to the existing data.\n * Set to `'replace'` to write all data to the cache once the stream ends.\n * @param reducer - A function to reduce the streamed chunks into the final data.\n * Defaults to a function that appends chunks to the end of the array.\n * @param initialValue - Initial value to be used while the first chunk is being fetched.\n */\nexport function streamedQuery<\n  TQueryFnData = unknown,\n  TData = Array<TQueryFnData>,\n  TQueryKey extends QueryKey = QueryKey,\n>({\n  streamFn,\n  refetchMode = 'reset',\n  reducer = (items, chunk) =>\n    addToEnd(items as Array<TQueryFnData>, chunk) as TData,\n  initialValue = [] as TData,\n}: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<\n  TData,\n  TQueryKey\n> {\n  return async (context) => {\n    const query = context.client\n      .getQueryCache()\n      .find({ queryKey: context.queryKey, exact: true })\n    const isRefetch = !!query && query.state.data !== undefined\n    if (isRefetch && refetchMode === 'reset') {\n      query.setState({\n        status: 'pending',\n        data: undefined,\n        error: null,\n        fetchStatus: 'fetching',\n      })\n    }\n\n    let result = initialValue\n\n    const stream = await streamFn(context)\n\n    for await (const chunk of stream) {\n      if (context.signal.aborted) {\n        break\n      }\n\n      // don't append to the cache directly when replace-refetching\n      if (!isRefetch || refetchMode !== 'replace') {\n        context.client.setQueryData<TData>(context.queryKey, (prev) =>\n          reducer(prev === undefined ? initialValue : prev, chunk),\n        )\n      }\n      result = reducer(result, chunk)\n    }\n\n    // finalize result: replace-refetching needs to write to the cache\n    if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {\n      context.client.setQueryData<TData>(context.queryKey, result)\n    }\n\n    return context.client.getQueryData(context.queryKey)!\n  }\n}\n"],"mappings":";;;;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AAAA,mBAAyB;AA6ClB,SAAS,cAId;AAAA,EACA;AAAA,EACA,cAAc;AAAA,EACd,UAAU,CAAC,OAAO,cAChB,uBAAS,OAA8B,KAAK;AAAA,EAC9C,eAAe,CAAC;AAClB,GAGE;AACA,SAAO,OAAO,YAAY;AACxB,UAAM,QAAQ,QAAQ,OACnB,cAAc,EACd,KAAK,EAAE,UAAU,QAAQ,UAAU,OAAO,KAAK,CAAC;AACnD,UAAM,YAAY,CAAC,CAAC,SAAS,MAAM,MAAM,SAAS;AAClD,QAAI,aAAa,gBAAgB,SAAS;AACxC,YAAM,SAAS;AAAA,QACb,QAAQ;AAAA,QACR,MAAM;AAAA,QACN,OAAO;AAAA,QACP,aAAa;AAAA,MACf,CAAC;AAAA,IACH;AAEA,QAAI,SAAS;AAEb,UAAM,SAAS,MAAM,SAAS,OAAO;AAErC,qBAAiB,SAAS,QAAQ;AAChC,UAAI,QAAQ,OAAO,SAAS;AAC1B;AAAA,MACF;AAGA,UAAI,CAAC,aAAa,gBAAgB,WAAW;AAC3C,gBAAQ,OAAO;AAAA,UAAoB,QAAQ;AAAA,UAAU,CAAC,SACpD,QAAQ,SAAS,SAAY,eAAe,MAAM,KAAK;AAAA,QACzD;AAAA,MACF;AACA,eAAS,QAAQ,QAAQ,KAAK;AAAA,IAChC;AAGA,QAAI,aAAa,gBAAgB,aAAa,CAAC,QAAQ,OAAO,SAAS;AACrE,cAAQ,OAAO,aAAoB,QAAQ,UAAU,MAAM;AAAA,IAC7D;AAEA,WAAO,QAAQ,OAAO,aAAa,QAAQ,QAAQ;AAAA,EACrD;AACF;","names":[]}

build/modern/streamedQuery.d.cts

@@ -5,2 +5,15 @@ import { I as QueryKey, a1 as QueryFunctionContext, Y as QueryFunction } from './hydration-BC7iBQD-.cjs';
 
+type BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {
+    streamFn: (context: QueryFunctionContext<TQueryKey>) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>;
+    refetchMode?: 'append' | 'reset' | 'replace';
+};
+type SimpleStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+    reducer?: never;
+    initialValue?: never;
+};
+type ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+    reducer: (acc: TData, chunk: TQueryFnData) => TData;
+    initialValue: TData;
+};
+type StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> = SimpleStreamedQueryParams<TQueryFnData, TQueryKey> | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>;
 /**
@@ -16,13 +29,8 @@ * This is a helper function to create a query function that streams data from an AsyncIterable.
  * Set to `'replace'` to write all data to the cache once the stream ends.
- * @param maxChunks - The maximum number of chunks to keep in the cache.
- * Defaults to `undefined`, meaning all chunks will be kept.
- * If `undefined` or `0`, the number of chunks is unlimited.
- * If the number of chunks exceeds this number, the oldest chunk will be removed.
+ * @param reducer - A function to reduce the streamed chunks into the final data.
+ * Defaults to a function that appends chunks to the end of the array.
+ * @param initialValue - Initial value to be used while the first chunk is being fetched.
  */
-declare function streamedQuery<TQueryFnData = unknown, TQueryKey extends QueryKey = QueryKey>({ queryFn, refetchMode, maxChunks, }: {
-    queryFn: (context: QueryFunctionContext<TQueryKey>) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>;
-    refetchMode?: 'append' | 'reset' | 'replace';
-    maxChunks?: number;
-}): QueryFunction<Array<TQueryFnData>, TQueryKey>;
+declare function streamedQuery<TQueryFnData = unknown, TData = Array<TQueryFnData>, TQueryKey extends QueryKey = QueryKey>({ streamFn, refetchMode, reducer, initialValue, }: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<TData, TQueryKey>;
 
 export { streamedQuery };

build/modern/streamedQuery.d.ts

@@ -5,2 +5,15 @@ import { I as QueryKey, a1 as QueryFunctionContext, Y as QueryFunction } from './hydration-BpLOP9Dw.js';
 
+type BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {
+    streamFn: (context: QueryFunctionContext<TQueryKey>) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>;
+    refetchMode?: 'append' | 'reset' | 'replace';
+};
+type SimpleStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+    reducer?: never;
+    initialValue?: never;
+};
+type ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+    reducer: (acc: TData, chunk: TQueryFnData) => TData;
+    initialValue: TData;
+};
+type StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> = SimpleStreamedQueryParams<TQueryFnData, TQueryKey> | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>;
 /**
@@ -16,13 +29,8 @@ * This is a helper function to create a query function that streams data from an AsyncIterable.
  * Set to `'replace'` to write all data to the cache once the stream ends.
- * @param maxChunks - The maximum number of chunks to keep in the cache.
- * Defaults to `undefined`, meaning all chunks will be kept.
- * If `undefined` or `0`, the number of chunks is unlimited.
- * If the number of chunks exceeds this number, the oldest chunk will be removed.
+ * @param reducer - A function to reduce the streamed chunks into the final data.
+ * Defaults to a function that appends chunks to the end of the array.
+ * @param initialValue - Initial value to be used while the first chunk is being fetched.
  */
-declare function streamedQuery<TQueryFnData = unknown, TQueryKey extends QueryKey = QueryKey>({ queryFn, refetchMode, maxChunks, }: {
-    queryFn: (context: QueryFunctionContext<TQueryKey>) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>;
-    refetchMode?: 'append' | 'reset' | 'replace';
-    maxChunks?: number;
-}): QueryFunction<Array<TQueryFnData>, TQueryKey>;
+declare function streamedQuery<TQueryFnData = unknown, TData = Array<TQueryFnData>, TQueryKey extends QueryKey = QueryKey>({ streamFn, refetchMode, reducer, initialValue, }: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<TData, TQueryKey>;
 
 export { streamedQuery };

build/modern/streamedQuery.js

 // src/streamedQuery.ts
 import { addToEnd } from "./utils.js";
 function streamedQuery({
-  queryFn,
+  streamFn,
   refetchMode = "reset",
-  maxChunks
+  reducer = (items, chunk) => addToEnd(items, chunk),
+  initialValue = []
 }) {
@@ -19,4 +20,4 @@ return async (context) => {
     }
-    let result = [];
-    const stream = await queryFn(context);
+    let result = initialValue;
+    const stream = await streamFn(context);
     for await (const chunk of stream) {
@@ -29,8 +30,6 @@ if (context.signal.aborted) {
           context.queryKey,
-          (prev = []) => {
-            return addToEnd(prev, chunk, maxChunks);
-          }
+          (prev) => reducer(prev === void 0 ? initialValue : prev, chunk)
         );
       }
-      result = addToEnd(result, chunk, maxChunks);
+      result = reducer(result, chunk);
     }
@@ -37,0 +36,0 @@ if (isRefetch && refetchMode === "replace" && !context.signal.aborted) {

build/modern/streamedQuery.js.map

@@ -1,1 +0,1 @@
-{"version":3,"sources":["../../src/streamedQuery.ts"],"sourcesContent":["import { addToEnd } from './utils'\nimport type { QueryFunction, QueryFunctionContext, QueryKey } from './types'\n\n/**\n * This is a helper function to create a query function that streams data from an AsyncIterable.\n * Data will be an Array of all the chunks received.\n * The query will be in a 'pending' state until the first chunk of data is received, but will go to 'success' after that.\n * The query will stay in fetchStatus 'fetching' until the stream ends.\n * @param queryFn - The function that returns an AsyncIterable to stream data from.\n * @param refetchMode - Defines how re-fetches are handled.\n * Defaults to `'reset'`, erases all data and puts the query back into `pending` state.\n * Set to `'append'` to append new data to the existing data.\n * Set to `'replace'` to write all data to the cache once the stream ends.\n * @param maxChunks - The maximum number of chunks to keep in the cache.\n * Defaults to `undefined`, meaning all chunks will be kept.\n * If `undefined` or `0`, the number of chunks is unlimited.\n * If the number of chunks exceeds this number, the oldest chunk will be removed.\n */\nexport function streamedQuery<\n  TQueryFnData = unknown,\n  TQueryKey extends QueryKey = QueryKey,\n>({\n  queryFn,\n  refetchMode = 'reset',\n  maxChunks,\n}: {\n  queryFn: (\n    context: QueryFunctionContext<TQueryKey>,\n  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>\n  refetchMode?: 'append' | 'reset' | 'replace'\n  maxChunks?: number\n}): QueryFunction<Array<TQueryFnData>, TQueryKey> {\n  return async (context) => {\n    const query = context.client\n      .getQueryCache()\n      .find({ queryKey: context.queryKey, exact: true })\n    const isRefetch = !!query && query.state.data !== undefined\n\n    if (isRefetch && refetchMode === 'reset') {\n      query.setState({\n        status: 'pending',\n        data: undefined,\n        error: null,\n        fetchStatus: 'fetching',\n      })\n    }\n\n    let result: Array<TQueryFnData> = []\n    const stream = await queryFn(context)\n\n    for await (const chunk of stream) {\n      if (context.signal.aborted) {\n        break\n      }\n\n      // don't append to the cache directly when replace-refetching\n      if (!isRefetch || refetchMode !== 'replace') {\n        context.client.setQueryData<Array<TQueryFnData>>(\n          context.queryKey,\n          (prev = []) => {\n            return addToEnd(prev, chunk, maxChunks)\n          },\n        )\n      }\n      result = addToEnd(result, chunk, maxChunks)\n    }\n\n    // finalize result: replace-refetching needs to write to the cache\n    if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {\n      context.client.setQueryData<Array<TQueryFnData>>(context.queryKey, result)\n    }\n\n    return context.client.getQueryData(context.queryKey)!\n  }\n}\n"],"mappings":";AAAA,SAAS,gBAAgB;AAkBlB,SAAS,cAGd;AAAA,EACA;AAAA,EACA,cAAc;AAAA,EACd;AACF,GAMkD;AAChD,SAAO,OAAO,YAAY;AACxB,UAAM,QAAQ,QAAQ,OACnB,cAAc,EACd,KAAK,EAAE,UAAU,QAAQ,UAAU,OAAO,KAAK,CAAC;AACnD,UAAM,YAAY,CAAC,CAAC,SAAS,MAAM,MAAM,SAAS;AAElD,QAAI,aAAa,gBAAgB,SAAS;AACxC,YAAM,SAAS;AAAA,QACb,QAAQ;AAAA,QACR,MAAM;AAAA,QACN,OAAO;AAAA,QACP,aAAa;AAAA,MACf,CAAC;AAAA,IACH;AAEA,QAAI,SAA8B,CAAC;AACnC,UAAM,SAAS,MAAM,QAAQ,OAAO;AAEpC,qBAAiB,SAAS,QAAQ;AAChC,UAAI,QAAQ,OAAO,SAAS;AAC1B;AAAA,MACF;AAGA,UAAI,CAAC,aAAa,gBAAgB,WAAW;AAC3C,gBAAQ,OAAO;AAAA,UACb,QAAQ;AAAA,UACR,CAAC,OAAO,CAAC,MAAM;AACb,mBAAO,SAAS,MAAM,OAAO,SAAS;AAAA,UACxC;AAAA,QACF;AAAA,MACF;AACA,eAAS,SAAS,QAAQ,OAAO,SAAS;AAAA,IAC5C;AAGA,QAAI,aAAa,gBAAgB,aAAa,CAAC,QAAQ,OAAO,SAAS;AACrE,cAAQ,OAAO,aAAkC,QAAQ,UAAU,MAAM;AAAA,IAC3E;AAEA,WAAO,QAAQ,OAAO,aAAa,QAAQ,QAAQ;AAAA,EACrD;AACF;","names":[]}
+{"version":3,"sources":["../../src/streamedQuery.ts"],"sourcesContent":["import { addToEnd } from './utils'\nimport type { QueryFunction, QueryFunctionContext, QueryKey } from './types'\n\ntype BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {\n  streamFn: (\n    context: QueryFunctionContext<TQueryKey>,\n  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>\n  refetchMode?: 'append' | 'reset' | 'replace'\n}\n\ntype SimpleStreamedQueryParams<\n  TQueryFnData,\n  TQueryKey extends QueryKey,\n> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {\n  reducer?: never\n  initialValue?: never\n}\n\ntype ReducibleStreamedQueryParams<\n  TQueryFnData,\n  TData,\n  TQueryKey extends QueryKey,\n> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {\n  reducer: (acc: TData, chunk: TQueryFnData) => TData\n  initialValue: TData\n}\n\ntype StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> =\n  | SimpleStreamedQueryParams<TQueryFnData, TQueryKey>\n  | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>\n\n/**\n * This is a helper function to create a query function that streams data from an AsyncIterable.\n * Data will be an Array of all the chunks received.\n * The query will be in a 'pending' state until the first chunk of data is received, but will go to 'success' after that.\n * The query will stay in fetchStatus 'fetching' until the stream ends.\n * @param queryFn - The function that returns an AsyncIterable to stream data from.\n * @param refetchMode - Defines how re-fetches are handled.\n * Defaults to `'reset'`, erases all data and puts the query back into `pending` state.\n * Set to `'append'` to append new data to the existing data.\n * Set to `'replace'` to write all data to the cache once the stream ends.\n * @param reducer - A function to reduce the streamed chunks into the final data.\n * Defaults to a function that appends chunks to the end of the array.\n * @param initialValue - Initial value to be used while the first chunk is being fetched.\n */\nexport function streamedQuery<\n  TQueryFnData = unknown,\n  TData = Array<TQueryFnData>,\n  TQueryKey extends QueryKey = QueryKey,\n>({\n  streamFn,\n  refetchMode = 'reset',\n  reducer = (items, chunk) =>\n    addToEnd(items as Array<TQueryFnData>, chunk) as TData,\n  initialValue = [] as TData,\n}: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<\n  TData,\n  TQueryKey\n> {\n  return async (context) => {\n    const query = context.client\n      .getQueryCache()\n      .find({ queryKey: context.queryKey, exact: true })\n    const isRefetch = !!query && query.state.data !== undefined\n    if (isRefetch && refetchMode === 'reset') {\n      query.setState({\n        status: 'pending',\n        data: undefined,\n        error: null,\n        fetchStatus: 'fetching',\n      })\n    }\n\n    let result = initialValue\n\n    const stream = await streamFn(context)\n\n    for await (const chunk of stream) {\n      if (context.signal.aborted) {\n        break\n      }\n\n      // don't append to the cache directly when replace-refetching\n      if (!isRefetch || refetchMode !== 'replace') {\n        context.client.setQueryData<TData>(context.queryKey, (prev) =>\n          reducer(prev === undefined ? initialValue : prev, chunk),\n        )\n      }\n      result = reducer(result, chunk)\n    }\n\n    // finalize result: replace-refetching needs to write to the cache\n    if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {\n      context.client.setQueryData<TData>(context.queryKey, result)\n    }\n\n    return context.client.getQueryData(context.queryKey)!\n  }\n}\n"],"mappings":";AAAA,SAAS,gBAAgB;AA6ClB,SAAS,cAId;AAAA,EACA;AAAA,EACA,cAAc;AAAA,EACd,UAAU,CAAC,OAAO,UAChB,SAAS,OAA8B,KAAK;AAAA,EAC9C,eAAe,CAAC;AAClB,GAGE;AACA,SAAO,OAAO,YAAY;AACxB,UAAM,QAAQ,QAAQ,OACnB,cAAc,EACd,KAAK,EAAE,UAAU,QAAQ,UAAU,OAAO,KAAK,CAAC;AACnD,UAAM,YAAY,CAAC,CAAC,SAAS,MAAM,MAAM,SAAS;AAClD,QAAI,aAAa,gBAAgB,SAAS;AACxC,YAAM,SAAS;AAAA,QACb,QAAQ;AAAA,QACR,MAAM;AAAA,QACN,OAAO;AAAA,QACP,aAAa;AAAA,MACf,CAAC;AAAA,IACH;AAEA,QAAI,SAAS;AAEb,UAAM,SAAS,MAAM,SAAS,OAAO;AAErC,qBAAiB,SAAS,QAAQ;AAChC,UAAI,QAAQ,OAAO,SAAS;AAC1B;AAAA,MACF;AAGA,UAAI,CAAC,aAAa,gBAAgB,WAAW;AAC3C,gBAAQ,OAAO;AAAA,UAAoB,QAAQ;AAAA,UAAU,CAAC,SACpD,QAAQ,SAAS,SAAY,eAAe,MAAM,KAAK;AAAA,QACzD;AAAA,MACF;AACA,eAAS,QAAQ,QAAQ,KAAK;AAAA,IAChC;AAGA,QAAI,aAAa,gBAAgB,aAAa,CAAC,QAAQ,OAAO,SAAS;AACrE,cAAQ,OAAO,aAAoB,QAAQ,UAAU,MAAM;AAAA,IAC7D;AAEA,WAAO,QAAQ,OAAO,aAAa,QAAQ,QAAQ;AAAA,EACrD;AACF;","names":[]}

package.json

 {
   "name": "@tanstack/query-core",
-  "version": "5.85.9",
+  "version": "5.86.0",
   "description": "The framework agnostic core that powers TanStack Query",
@@ -5,0 +5,0 @@ "author": "tannerlinsley",

src/streamedQuery.ts

 import { addToEnd } from './utils'
 import type { QueryFunction, QueryFunctionContext, QueryKey } from './types'
 
+type BaseStreamedQueryParams<TQueryFnData, TQueryKey extends QueryKey> = {
+  streamFn: (
+    context: QueryFunctionContext<TQueryKey>,
+  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>
+  refetchMode?: 'append' | 'reset' | 'replace'
+}
+
+type SimpleStreamedQueryParams<
+  TQueryFnData,
+  TQueryKey extends QueryKey,
+> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+  reducer?: never
+  initialValue?: never
+}
+
+type ReducibleStreamedQueryParams<
+  TQueryFnData,
+  TData,
+  TQueryKey extends QueryKey,
+> = BaseStreamedQueryParams<TQueryFnData, TQueryKey> & {
+  reducer: (acc: TData, chunk: TQueryFnData) => TData
+  initialValue: TData
+}
+
+type StreamedQueryParams<TQueryFnData, TData, TQueryKey extends QueryKey> =
+  | SimpleStreamedQueryParams<TQueryFnData, TQueryKey>
+  | ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>
+
 /**
@@ -14,21 +42,20 @@ * This is a helper function to create a query function that streams data from an AsyncIterable.
  * Set to `'replace'` to write all data to the cache once the stream ends.
- * @param maxChunks - The maximum number of chunks to keep in the cache.
- * Defaults to `undefined`, meaning all chunks will be kept.
- * If `undefined` or `0`, the number of chunks is unlimited.
- * If the number of chunks exceeds this number, the oldest chunk will be removed.
+ * @param reducer - A function to reduce the streamed chunks into the final data.
+ * Defaults to a function that appends chunks to the end of the array.
+ * @param initialValue - Initial value to be used while the first chunk is being fetched.
  */
 export function streamedQuery<
   TQueryFnData = unknown,
+  TData = Array<TQueryFnData>,
   TQueryKey extends QueryKey = QueryKey,
 >({
-  queryFn,
+  streamFn,
   refetchMode = 'reset',
-  maxChunks,
-}: {
-  queryFn: (
-    context: QueryFunctionContext<TQueryKey>,
-  ) => AsyncIterable<TQueryFnData> | Promise<AsyncIterable<TQueryFnData>>
-  refetchMode?: 'append' | 'reset' | 'replace'
-  maxChunks?: number
-}): QueryFunction<Array<TQueryFnData>, TQueryKey> {
+  reducer = (items, chunk) =>
+    addToEnd(items as Array<TQueryFnData>, chunk) as TData,
+  initialValue = [] as TData,
+}: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<
+  TData,
+  TQueryKey
+> {
   return async (context) => {
@@ -39,3 +66,2 @@ const query = context.client
     const isRefetch = !!query && query.state.data !== undefined
-
     if (isRefetch && refetchMode === 'reset') {
@@ -50,5 +76,6 @@ query.setState({
 
-    let result: Array<TQueryFnData> = []
-    const stream = await queryFn(context)
+    let result = initialValue
 
+    const stream = await streamFn(context)
+
     for await (const chunk of stream) {
@@ -61,10 +88,7 @@ if (context.signal.aborted) {
       if (!isRefetch || refetchMode !== 'replace') {
-        context.client.setQueryData<Array<TQueryFnData>>(
-          context.queryKey,
-          (prev = []) => {
-            return addToEnd(prev, chunk, maxChunks)
-          },
+        context.client.setQueryData<TData>(context.queryKey, (prev) =>
+          reducer(prev === undefined ? initialValue : prev, chunk),
         )
       }
-      result = addToEnd(result, chunk, maxChunks)
+      result = reducer(result, chunk)
     }
@@ -74,3 +98,3 @@
     if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {
-      context.client.setQueryData<Array<TQueryFnData>>(context.queryKey, result)
+      context.client.setQueryData<TData>(context.queryKey, result)
     }
@@ -77,0 +101,0 @@

New alerts

Network access

Supply chain risk

This module accesses the network.

Found 1 instance in 1 package

Environment variable access

Supply chain risk

Package accesses environment variables, which may be a sign of credential stuffing or data theft.

Found 1 instance in 1 package

Long strings

Supply chain risk

Contains long string literals, which may be a sign of obfuscated or packed code.

Found 1 instance in 1 package

Fixed alerts

Network access

Supply chain risk

This module accesses the network.

Found 1 instance in 1 package

Environment variable access

Supply chain risk

Package accesses environment variables, which may be a sign of credential stuffing or data theft.

Found 1 instance in 1 package

Long strings

Supply chain risk

Contains long string literals, which may be a sign of obfuscated or packed code.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

2053799

0.33%

Lines of code

23260

0.14%

No dependency changes