Replies: 9 comments 37 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.
-
|
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.
-
wouldn’t it be possible to just keep them in the cache as an array of chunks, and then use |
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.
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