[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]

it’s weird. when I write the selection function, it seems to work:

but then once I’m done, it’s unknown:

TypeScript Playground

seems like a bug

[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)