Replies: 14 comments 52 replies
-
|
I'm working on a project where I want to sync server and client state. Here is my feedback on streamedQuery:
|
Beta Was this translation helpful? Give feedback.
-
|
Thanks for the feedback 👍
I’m not opposed. It’s partially why I’ve made the input to
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
Does the Promise reject if the connection drops? If so, I’d expect our |
Beta Was this translation helpful? Give feedback.
-
|
I wrote something very similar to
My feedback
|
Beta Was this translation helpful? Give feedback.
-
|
Thanks for the feedback 🎉
good point, but can’t this by done by controlling when you
So in that sense, it would be something like |
Beta Was this translation helpful? Give feedback.
-
Yes, we could have a helper similar to bufferTime to batch updates. In that case we'd
Sounds good |
Beta Was this translation helpful? Give feedback.
-
|
I think it’s confusing that we unwrap the top level array. Maybe we should just wrap it in an extra array: to prevent that? Yes that means you’ll get array of array if you yield an array, but you can always |
Beta Was this translation helpful? Give feedback.
-
|
That might be a good idea, and with a |
Beta Was this translation helpful? Give feedback.
-
|
wanna take a look here: |
Beta Was this translation helpful? Give feedback.
-
|
@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. |
Beta Was this translation helpful? Give feedback.
-
|
Thanks for the feedback. Brings me one step closer to marking it as stable 🙌 |
Beta Was this translation helpful? Give feedback.
-
|
Hi, great to hear react-query can support stream!!! By the way, |
Beta Was this translation helpful? Give feedback.
-
|
ugh, it works if @Andarist is this a TypeScript limitation? |
Beta Was this translation helpful? Give feedback.
-
|
Without analyzing this deeply... it looks like lack of this: microsoft/TypeScript#54183 |
Beta Was this translation helpful? Give feedback.
-
|
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? |
Beta Was this translation helpful? Give feedback.
-
No, but you could ask Jake there to prepare one (preparing one requires a TS team member pinging their bot) |
Beta Was this translation helpful? Give feedback.
-
|
As expected, that PR does fix this :p TS playground |
Beta Was this translation helpful? Give feedback.
-
|
@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? 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. |
Beta Was this translation helpful? Give feedback.
-
|
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
}) |
Beta Was this translation helpful? Give feedback.
-
|
I also hope that the |
Beta Was this translation helpful? Give feedback.
-
|
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 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 |
Beta Was this translation helpful? Give feedback.
-
|
yeah we could implement something similar to |
Beta Was this translation helpful? Give feedback.
-
|
Have a similar usecase and yeah a simple solution of limiting the size of the "pages" would rectify this |
Beta Was this translation helpful? Give feedback.
-
|
alright, gonna ship this: |
Beta Was this translation helpful? Give feedback.
-
|
shipped in: |
Beta Was this translation helpful? Give feedback.
-
|
Thanks your job! |
Beta Was this translation helpful? Give feedback.
-
|
I need the ability to surgically replace specific chunks in the cache. I'm getting back a stream of events, and typically I just want to append them. However some of the events are updates to older entries, and when those come in I want to dump the old entry. It's similar to Docker build output, where there are multiple progress bars going at once: Subsequent events for the same ID look like this: [
{
"id": "sha256:1593650c75729f64218ae272e8ffff9da7bbba9599bd1815877da99a2651fd9b",
"data": "22.40MB / 23.59MB"
},
{
"id": "sha256:1593650c75729f64218ae272e8ffff9da7bbba9599bd1815877da99a2651fd9b",
"data": "23.00MB / 23.59MB"
},
{
"id": "sha256:1593650c75729f64218ae272e8ffff9da7bbba9599bd1815877da99a2651fd9b",
"data": "23.59MB / 23.59MB"
}
]There can be a lot of events with the same ID, so I don't want to leave them in the cache. ATM I am using a fork of |
Beta Was this translation helpful? Give feedback.
-
|
hm, I’m not sure if we should build this into usually I would just say keep all chunks in cache and derive the final result with |
Beta Was this translation helpful? Give feedback.
-
|
Yeah, it's just a lot of extra data to keep in the cache. I don't necessarily need that low level of an API. If there was a way to specify a key to use for each chunk that the data was de-duped by, that would work. For example: const query = queryOptions({
queryKey: ['data'],
queryFn: streamedQuery({
queryFn: fetchDataInChunks,
keyFn: (chunk) => chunk.key,
}),
})It's not too hard to maintain my own version so no worries if that doesn't fit with the rest of the API. |
Beta Was this translation helpful? Give feedback.
-
|
I found myself in a similar position. I'm periodically fetching state of local network devices, and I only want to keep the latest for each device. Sometimes, I need to append a new device, but mostly it's replacing that I have to do. I have also implemented my own fork of |
Beta Was this translation helpful? Give feedback.
-
|
yeah, we’re likely gonna go with this: |
Beta Was this translation helpful? Give feedback.
-
|
I'm having some issues cancelling a running streaming query. Here's some example code. Hitting the stop button does not cancel the query (I still see the API call running in Network). Am I misunderstanding how to use Tanstack here? Sorry if I'm missing something super obvious. import {
QueryClient,
QueryClientProvider,
queryOptions,
useQuery,
} from "@tanstack/react-query";
import { experimental_streamedQuery as streamedQuery } from "@tanstack/react-query";
import { ReactQueryDevtools } from "@tanstack/react-query-devtools";
import { useState } from "react";
const queryClient = new QueryClient();
function StreamingPageInner() {
const [isStreaming, setIsStreaming] = useState(false);
const query = queryOptions({
queryKey: ["blah"],
queryFn: streamedQuery({
queryFn: () => {
return {
async *[Symbol.asyncIterator]() {
const response = await fetch(
"http://localhost:8000/v1/simulation/streaming-example",
{
headers: {
Accept: "text/event-stream",
},
},
);
const reader = response.body?.getReader();
if (!reader) throw new Error("No reader available");
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
// Convert the Uint8Array to text
const text = new TextDecoder().decode(value);
yield text;
}
} finally {
reader.releaseLock();
}
},
};
},
}),
staleTime: Infinity,
enabled: isStreaming,
});
const handleStreamingToggle = () => {
if (isStreaming) {
// Cancel the ongoing query
console.log("Cancelling query");
queryClient.cancelQueries({ queryKey: ["blah"] });
}
setIsStreaming(!isStreaming);
};
const { data } = useQuery(query);
return (
<QueryClientProvider client={queryClient}>
<div className="p-8">
<h1 className="text-2xl font-bold mb-4">Streaming Text</h1>
<button
onClick={handleStreamingToggle}
className="mb-4 px-4 py-2 bg-blue-500 text-white rounded hover:bg-blue-600 transition-colors"
>
{isStreaming ? "Stop Stream" : "Start Stream"}
</button>
<div className="text-lg text-white">{data}</div>
</div>
</QueryClientProvider>
);
}
function StreamingPagev2() {
return (
<QueryClientProvider client={queryClient}>
<StreamingPageInner />
<ReactQueryDevtools initialIsOpen={false} />
</QueryClientProvider>
);
}
export default StreamingPagev2;Here's a working version that doesn't leverage Tanstack import { useEffect, useState } from "react";
function StreamingPageInner() {
const [isStreaming, setIsStreaming] = useState(false);
const [streamedData, setStreamedData] = useState("");
const [reader, setReader] =
useState<ReadableStreamDefaultReader<Uint8Array> | null>(null);
useEffect(() => {
if (!isStreaming) {
// Clean up the reader when streaming is stopped
if (reader) {
reader.cancel();
setReader(null);
}
return;
}
const startStreaming = async () => {
try {
const response = await fetch(
"http://localhost:8000/v1/simulation/streaming-example",
{
headers: {
Accept: "text/event-stream",
},
},
);
const newReader = response.body?.getReader();
if (!newReader) throw new Error("No reader available");
setReader(newReader);
while (true) {
const { done, value } = await newReader.read();
if (done) break;
const text = new TextDecoder().decode(value);
setStreamedData((prev) => prev + text);
}
} catch (error) {
console.error("Streaming error:", error);
setIsStreaming(false);
}
};
startStreaming();
return () => {
if (reader) {
reader.cancel();
setReader(null);
}
};
}, [isStreaming]);
const handleStreamingToggle = () => {
setIsStreaming(!isStreaming);
if (!isStreaming) {
setStreamedData(""); // Clear previous data when starting new stream
}
};
return (
<div className="p-8">
<h1 className="text-2xl font-bold mb-4">Streaming Text</h1>
<button
onClick={handleStreamingToggle}
className="mb-4 px-4 py-2 bg-blue-500 text-white rounded hover:bg-blue-600 transition-colors"
>
{isStreaming ? "Stop Stream" : "Start Stream"}
</button>
<div className="text-lg text-white">{streamedData}</div>
</div>
);
}
export default function StreamingPage() {
return <StreamingPageInner />;
} |
Beta Was this translation helpful? Give feedback.
-
|
You need to opt in to the abortSignal passed to the queryFn: https://tanstack.com/query/latest/docs/framework/react/guides/query-cancellation#using-fetch |
Beta Was this translation helpful? Give feedback.
-
|
oh nice. That worked! Thank you!! Posting the working version incase it's useful for someone in the future import {
QueryClient,
QueryClientProvider,
queryOptions,
useQuery,
} from "@tanstack/react-query";
import { experimental_streamedQuery as streamedQuery } from "@tanstack/react-query";
import { ReactQueryDevtools } from "@tanstack/react-query-devtools";
import { useState } from "react";
const queryClient = new QueryClient();
function StreamingPageInner() {
const [isStreaming, setIsStreaming] = useState(false);
const query = queryOptions({
queryKey: ["blah"],
queryFn: streamedQuery({
queryFn: ({ signal }) => {
return {
async *[Symbol.asyncIterator]() {
const response = await fetch(
"http://localhost:8000/v1/simulation/streaming-example",
{
headers: {
Accept: "text/event-stream",
},
signal,
},
);
const reader = response.body?.getReader();
if (!reader) throw new Error("No reader available");
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
// Convert the Uint8Array to text
const text = new TextDecoder().decode(value);
yield text;
}
} finally {
reader.releaseLock();
}
},
};
},
}),
staleTime: Infinity,
enabled: isStreaming,
});
const handleStreamingToggle = () => {
if (isStreaming) {
// Cancel the ongoing query
console.log("Cancelling query");
queryClient.cancelQueries({ queryKey: ["blah"] });
}
setIsStreaming(!isStreaming);
};
const { data } = useQuery(query);
return (
<QueryClientProvider client={queryClient}>
<div className="p-8">
<h1 className="text-2xl font-bold mb-4">Streaming Text</h1>
<button
onClick={handleStreamingToggle}
className="mb-4 px-4 py-2 bg-blue-500 text-white rounded hover:bg-blue-600 transition-colors"
>
{isStreaming ? "Stop Stream" : "Start Stream"}
</button>
<div className="text-lg text-white">{data}</div>
</div>
</QueryClientProvider>
);
}
function StreamingPagev2() {
return (
<QueryClientProvider client={queryClient}>
<StreamingPageInner />
<ReactQueryDevtools initialIsOpen={false} />
</QueryClientProvider>
);
}
export default StreamingPagev2; |
Beta Was this translation helpful? Give feedback.
-
|
Feature request Description Example 1 - Video views live const subscribeToVideoViews = ({videoId}) => {
return {
async *[Symbol.asyncIterator]() {
let { resolve, promise } = Promise.withResolvers();
const initialStateMessage = subscribe({
route: `video.${videoId}.views`,
onNext(videoViews) {
resolve(videoViews);
},
});
while (true) {
const videoViews = await promise;
const promiseWithResolves = Promise.withResolvers();
resolve = promiseWithResolves.resolve;
promise = promiseWithResolves.promise;
yield videoViews;
}
},
};
}
queryOptions({
queryKey: ['video', videoId, 'views'],
queryFn: streamedQuery({
queryFn: ({ signal }) =>
subscribeToVideoViews({ videoId, }),
}),
});Example 2 - Live user updates const subscribeToVideoViews = ({userId, getData}) => {
return {
async *[Symbol.asyncIterator]() {
let { resolve, promise } = Promise.withResolvers();
const initialStateMessage = subscribe({
route: `users.${userId}`,
onNext(userMessage) {
if (userMessage.type === 'initialState') {
resolve(userMessage.payload) // User
} else if (userMessage.type === 'patch') {
resolve({
...getData(),
...userMessage.payload // Partial<User>
})
}
},
});
while (true) {
const user = await promise;
const promiseWithResolves = Promise.withResolvers();
resolve = promiseWithResolves.resolve;
promise = promiseWithResolves.promise;
yield user;
}
},
};
}
queryOptions({
queryKey: ['users', userId],
queryFn: streamedQuery({
queryFn: ({getData}) => // Note getData that helps with updates
subscribeToUser({ userId, getData }),
}),
}); |
Beta Was this translation helpful? Give feedback.
-
|
It would, but it adds an unnecessary structure on top of the existing model. This seems like a useful feature for "time travel", but in practice only the last computed state is needed. In addition, it corrupts |
Beta Was this translation helpful? Give feedback.
-
|
One more thing that would be handy for streams is a separate Imagine a WS temporarily loses a connection after it has been stablished. Typically, you don't want the UI to fallback to skeletons right away, there's a period in which a low-key reconnection is happening. In such scenarios, it would be ideally to either:
|
Beta Was this translation helpful? Give feedback.
-
|
Isn't that a bit out of scope. streamedQuery is focused on establishing caching and returning queries, something else should be worried about if the network state of the browser and act accordingly. |
Beta Was this translation helpful? Give feedback.
-
|
@dweng0 I agree, it's not directly a function of const deferredSocketId = useDeferredValue(socketId)
useSuspenseQuery({
queryKey: ['socket', deferredSocketId, 'video', videoId, 'views']
queryFn: streamedQuery({
queryFn: () =>
subscribeToVideoViews({ videoId }),
}),
select: (chunks) => chunks?.[chunks.length - 1]
}) |
Beta Was this translation helpful? Give feedback.
-
I think the |
Beta Was this translation helpful? Give feedback.
-
|
I'd like to note that I'm not entirely convinced a query—especially one keyed identically across subsequent calls—is the best fit for a chat-AI use case. Given the non-deterministic nature of AI-generated responses in conversational contexts, even identical prompts can yield substantially different replies, which undermines the typical assumptions of cacheability. Additionally, implementing reliable cache invalidation right after a streaming response completes is non-trivial, and risks either stale message retention or premature eviction depending on timing. |
Beta Was this translation helpful? Give feedback.
-
|
sure, but AI is not the only use-case for streaming... |
Beta Was this translation helpful? Give feedback.
-
|
Thanks—totally fair that streaming has broader use cases beyond AI. I just want to clarify where I’m coming from: I landed here directly from the TanStack website’s “example” section, which specifically highlights an AI chat use case featuring streaming. That context is what prompted my feedback. I still stand by the opinion that, in the specific case of AI chat—especially with non-deterministic outputs and the complexity around cache invalidation—it might not be the best fit for a query with identical keys across calls. |
Beta Was this translation helpful? Give feedback.
-
|
Hi, what if you delegate the "reduction" of the data to the consumer of the function? i mean an API like this streamedQuery({
queryFn: myStream,
reducer: (prev = '', chunk) => prev + chunk,
})and internally you decide to use the optional reducer or the addToEnd util function... something like below export function streamedQuery<
TChunk = unknown,
TAccum = TChunk,
TQueryKey extends QueryKey = QueryKey,
>({
queryFn,
reducer,
refetchMode = 'reset',
maxChunks,
}: {
queryFn: (
context: QueryFunctionContext<TQueryKey>,
) => AsyncIterable<TChunk> | Promise<AsyncIterable<TChunk>>
reducer?: (prev: TAccum, chunk: TChunk) => TAccum
refetchMode?: 'append' | 'reset' | 'replace'
maxChunks?: number
}): QueryFunction<TAccum, TQueryKey> {
return async (context) => {
const query = context.client
.getQueryCache()
.find({ queryKey: context.queryKey, exact: true })
const isRefetch = !!query && query.state.data !== undefined
if (isRefetch && refetchMode === 'reset') {
query.setState({
status: 'pending',
data: undefined,
error: null,
fetchStatus: 'fetching',
})
}
let result: TAccum = reducer
? undefined as any
: ([] as unknown as TAccum) // fallback to array default
const stream = await queryFn(context)
for await (const chunk of stream) {
if (context.signal.aborted) break
result = reducer
? reducer(result, chunk)
: addToEnd(result as TChunk[], chunk, maxChunks) as TAccum
if (!isRefetch || refetchMode !== 'replace') {
context.client.setQueryData<TAccum>(context.queryKey, result)
}
}
if (isRefetch && refetchMode === 'replace' && !context.signal.aborted) {
context.client.setQueryData<TAccum>(context.queryKey, result)
}
return context.client.getQueryData<TAccum>(context.queryKey)!
}
}mine is puedo code, i didn't test it, so maybe i'm missing something |
Beta Was this translation helpful? Give feedback.
-
|
I agree, a reducer makes more sense as the base interface than any particular type of reducer. Some particular reducers: adding to a list, adding to a list up to some limit, keeping the latest item only, map, sorted map, sorted set, regular set with max size, etc. Some can be provided for convenience or defaults, but the base interface should be more flexible. Many use-cases will likely have different reductions over the infinite stream - since it's infinite, it's important to manage the memory. |
Beta Was this translation helpful? Give feedback.
-
|
that’s a good point, but then
I think we can try replacing |
Beta Was this translation helpful? Give feedback.
-
|
@TkDodo |
Beta Was this translation helpful? Give feedback.
-
|
Can we add an onData callback function for event triggering scenarios? For example, every time a user clicks the play audio button, the data {action: 'play_audio'} is pushed, but the data has not changed, and useEffect cannot detect the change, so the event can only be triggered for the first time. After receiving the event, the data must be cleared before the next trigger can be performed. |
Beta Was this translation helpful? Give feedback.
-
|
Suppose a game like this, when a user clicks jump once, other users will receive {action: 'jump'}. So the jump animation is played once, but when the jump is performed again the second time, there is no change from the data perspective. But the character should jump twice. For this case, add an additional event unique id to each event. Or record all historical data, monitor the change in the length of the array, and only take the last data. However, for event-driven situations, most of the time only the latest data is concerned, that is, maxChunks is set to 1. Then there will be no simple way to detect when the same data arrives. |
Beta Was this translation helpful? Give feedback.
-
|
Do you think react-query and streamed query fit well with your use case? I mean i can imagine scenarios with data streaming to fetch a lot of data in chunks, but I never thought about events. My first thought goes to pub/sub pattern or observables or Server event/web socket |
Beta Was this translation helpful? Give feedback.
-
|
This is indeed a pub/sub model. But tanstack-query can be extended to easily achieve this effect, just add an onData callback when a message is received. |
Beta Was this translation helpful? Give feedback.
-
|
For this scenario not enough data is being passed to the UI. For example if you're streaming just the name of the action then it will work in a rudimentary way. However, you hit temporal problems pretty quickly, the most basic of them being if someone jumps multiple times. You could fix this on the UI but it doesn't resolve other more complex temporal problems you will face with a networked game. I personally would add a timestamp of when the server sent the action out. This would solve the issue of the data not triggering effects on the UI (for example when a user jumps twice in quick succession). But it would also help later on if you want to handle things like latency issues. |
Beta Was this translation helpful? Give feedback.
-
I don’t think I wanna add any new callbacks given how bad the last callbacks on useQuery turned out 😅 |
Beta Was this translation helpful? Give feedback.
-
|
Hi @TkDodo , thanks for the great work. I am using |
Beta Was this translation helpful? Give feedback.
-
|
@TkDodo it might be that the events that are available are exposed in some way so users can hook into them if required. |
Beta Was this translation helpful? Give feedback.
-
|
Thanks for sharing. BTW I do not think that it is a "EventSource" specific scenario. The "streaming" should be refer to sth more generic than EventSource. |
Beta Was this translation helpful? Give feedback.
-
|
It seems like the |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
-
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 notexperimentalbecause it “doesn’t work”, it’s merely marked asexperimentalso 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
streamedQueryhelper, if you like it, if there’s something missing etc. This will help us mark it asstablesooner. Thank you 🙏Beta Was this translation helpful? Give feedback.
All reactions