[Feedback wanted]: streamedQuery

[TkDodo]

We’ve recently released a new API to support streaming through AsyncIterables. You can read about that in the docs here, and we also have an example set-up here.

Since the API is currently marked as experimental, we would value your feedback if you can build the things you want to build with that API. Note that it’s not experimental because it “doesn’t work”, it’s merely marked as experimental so that we can adapt the API to your needs without releasing a new major version.

So please let us know if you’ve worked with the streamedQuery helper, if you like it, if there’s something missing etc. This will help us mark it as stable sooner. Thank you 🙏

[justmoon]

I'm working on a project where I want to sync server and client state. Here is my feedback on streamedQuery:

• Thank you so much for working on this feature. I think this is a common use case and currently whenever I want to do any kind of real-time data, I end up re-implementing parts of Tanstack Query. So I would very much love this to be supported.
• I'm using a reducer (which takes the streamed chunks and updates an immutable state object) so the current interface of streamedQuery doesn't quite work for me. You could add a reducer parameter which defaults to a reducer which concatenates the chunks into an array.
• I'm using Tanstack Router and because streamedQuery never finishes fetching, I couldn't use ensureQueryData in the loader like I normally would. I would be nice to change it somehow such that ensureQueryData returns once the first chunk has been received.
• Connections sometimes drop - re-connecting should probably be handled by the user but it would be nice to have an example for a streamedQuery which handled re-connecting automatically.

[TkDodo]

Thanks for the feedback 👍

I'm using a reducer (which takes the streamed chunks and updates an immutable state object) so the current interface of streamedQuery doesn't quite work for me. You could add a reducer parameter which defaults to a reducer which concatenates the chunks into an array.

I’m not opposed. It’s partially why I’ve made the input to streamedQuery an object - so that we can easily add more configs. Would you like to try and add that? It might be tricky on type level 🤔

I'm using Tanstack Router and because streamedQuery never finishes fetching, I couldn't use ensureQueryData in the loader like I normally would. I would be nice to change it somehow such that ensureQueryData returns once the first chunk has been received.

Ooh that’s a good point, I’ll have to think about that. The streaming makes more sense for observers because they update automatically, not so much for imperative fetches. However, we currently don’t know “how” a fetch was triggered, so we treat them all the same.

I also think this might be necessary for useSuspenseQuery because we throw a promise, and without that I guess react would just suspend until the stream is finished, which isn’t what we’d want. Are you using suspense by any chance?

Connections sometimes drop - re-connecting should probably be handled by the user but it would be nice to have an example for a streamedQuery which handled re-connecting automatically.

Does the Promise reject if the connection drops? If so, I’d expect our retry to kick in

[Rinaldo]

I wrote something very similar to streamedQuery a while ago to handle an api that returns thousands of rows of data and took a minute to do so. There are two differences between my helper and this new streamedQuery helper.

• We stream 100+ rows per second so we throttle calls to setQueryData
• If we already have data when making a new call, we buffer the data until the response is complete instead of updating it incrementally. If the user navigates away from the page and then returns, we want to show the cached full dataset until we have a new full dataset. We don't want to show we're building up the dataset again, and if the refetch fails partway through we'd rather have a complete stale dataset than an incomplete new one.

My feedback

• Add another refetchMode that only updates queryData incrementally if there's no existing data.
• Add some way to throttle updates to queryData. Either have streamedQuery throttle the updates or add a parameter to useQuery to throttle rerenders.

[TkDodo]

Thanks for the feedback 🎉

We stream 100+ rows per second so we throttle calls to setQueryData

good point, but can’t this by done by controlling when you yield ?

If we already have data when making a new call, we buffer the data until the response is complete instead of updating it incrementally.

So in that sense, it would be something like refetchMode: 'replace', which would only call setQueryData once when the stream is completed if data already exists (aka we have a refetch) ?

[Rinaldo]

can’t this by done by controlling when you yield?

Yes, we could have a helper similar to bufferTime to batch updates. In that case we'd yield arrays of items instead of individual items. That actually works in runtime right now since concat flattens arrays, but the typing is wrong. The fix might be as simple as updating this line to have TQueryFnData | TQueryFnData[]

something like refetchMode: 'replace'

Sounds good

[TkDodo]

I think it’s confusing that we unwrap the top level array. Maybe we should just wrap it in an extra array:

return prev.concat([chunk])

to prevent that? Yes that means you’ll get array of array if you yield an array, but you can always .flat() yourself

[Rinaldo]

That might be a good idea, and with a reducer parameter you could still construct a flat array to begin with and not need to use .flat() inside select

[TkDodo]

wanna take a look here:

• feat: refetchMode replace for streamedQuery #9092

[LovesWorking]

@TkDodo I'm using this to stream AI messages for mobile, I just wanted to say thank you! It was super easy to use and really straightforward. This simplified the code I was working on a lot and works perfect.

[TkDodo]

Thanks for the feedback. Brings me one step closer to marking it as stable 🙌

[marcog83]

Hi, great to hear react-query can support stream!!!
That's amazing.
We did a custom implementation on top of RQ, for Streaming data.
#7581 (comment)
Out of curiosity... What was your thinking process to choose the AsyncIterable way?
Because when we thought about what API could be more generic, we ended up with the Observer one, with a onNext, onComplete and onError.
And I'd like to hear your point of view 🙏

By the way,
Thanks a lot for bringing this functionality into RQ.

[TkDodo]

ugh, it works if queryFn is extracted to a separate const:

Fixed TypeScript Playground

@Andarist is this a TypeScript limitation?

[Andarist]

Without analyzing this deeply... it looks like lack of this: microsoft/TypeScript#54183

[TkDodo]

Gotta love how you have a PR up for everything 😍 . Are there PR preview builds one can try out in the TypeScript Playground to confirm this?

[Andarist]

Are there PR preview builds one can try out in the TypeScript Playground to confirm this?

No, but you could ask Jake there to prepare one (preparing one requires a TS team member pinging their bot)

[Andarist]

As expected, that PR does fix this :p TS playground

[motopods]

@TkDodo Is it possible to add a parameter configuration to limit the size of the array, and when it exceeds the limit, the earliest data will be removed to avoid accumulating too much data for a long time?
For example, if the array size is set to 1, I only need the latest data and don't care about historical data.
Typically used for event-driven.

Or maybe I just want to display the latest 5 data.

Currently, you must manually setQueryData to avoid accumulating too much data and occupying memory.

[marcog83]

I think you can control the flow with rxjs. From version 7.8.1 they implement Symbol.asyncIterator, so you can create an observable and specify the number of items you are interested in, with take (or takeLast).

Below you can find an example ( in real world you would replace interval(500) with fromFetch)

import { interval, take, takeUntil, fromEvent, EMPTY } from 'rxjs'
import { experimental_streamedQuery as streamedQuery } from '@tanstack/react-query'

const makeTicks$ = (signal?: AbortSignal) => {
  // notifier$ emits once when AbortController.abort() fires
  const notifier$ = signal ? fromEvent(signal, 'abort').pipe(take(1)) : EMPTY

  return interval(500).pipe(
    takeLast(5),          // last 5 numbers …
    takeUntil(notifier$)   // …or sooner if the query is aborted
  )
}

export const ticksQuery = streamedQuery({
  queryKey: ['ticks'],
  queryFn: async ({ signal }) => makeTicks$(signal)   // Observable is already AsyncIterable
})

[TkDodo]

I also hope that the async-iterator-helper proposal will “fix” this: https://github.com/tc39/proposal-async-iterator-helpers

[NeilTheFisher]

A use case/mode that I would like to have is one that replaces the query data rather than appending to it. I want a single object instead of a list of them. I believe it should be built into streamedQuery as a mode named "replace", but not to be confused with the replace from the refetchMode option.
I've solved the issue using this helper with the same signature as streamedQuery:

import type { QueryFunction, QueryFunctionContext, QueryKey } from "@tanstack/react-query";

export function replaceQuery<TData = unknown, TKey extends QueryKey = QueryKey>({
  queryFn,
}: {
  queryFn: (
    ctx: QueryFunctionContext<TKey>
  ) => AsyncIterable<TData> | Promise<AsyncIterable<TData>>;
}): QueryFunction<TData, TKey> {
  return async (ctx) => {
    const stream = await queryFn(ctx);
    let last: TData | undefined;

    for await (const chunk of stream) {
      if (ctx.signal.aborted) {
        break;
      }
      last = chunk;
      ctx.client.setQueryData<TData>(ctx.queryKey, chunk);
    }

    if (last !== undefined) return last;

    const cached = ctx.client.getQueryData<TData>(ctx.queryKey);
    if (cached !== undefined) return cached;

    throw new Error("No data emitted and no cached data available");
  };
}

my use case is with trpc-live and orpc. I could (and probably will) do query.data.at(-1), but I didn't want to store the entire history in the cache so that's why I thought of this.

[motopods]

That‘s what i am doing too. Have to set query data everytime.

[motopods]

I believe that keeping only the latest data should be a common operation and should be configurable through a simple parameter.

[NeilTheFisher]

@TkDodo
In this case since we both have this use case it seems like this would make sense to add as an option; a limit to the size of the array and/or simply an option that would replace the query data entirely

[TkDodo]

yeah we could implement something similar to maxPages on infinite queries that just limits how many entries are in the cache, which you could set to 1 ?