Seeking Feedback on useQueryStream Hook for Observable Integration with React Query

[marcog83]

Hi TanStack Team,

First off, a huge thank you for the fantastic work you’re doing with React Query! It’s been a total game-changer for us, making server state management a breeze.

I’m reaching out for some guidance to validate an idea we’ve been tinkering with—creating a custom hook called useQueryStream to handle server-sent events (SSE) with React Query. We know you’ve mentioned before that streaming data flows aren’t officially supported, but we think our situation might be a bit different, and we’d love to get your take on it.

Here’s Our Situation

Our backend team has endpoints that serve up large data lists and they’ve enabled SSE to push updates in real-time. We’ve got over 10 products, each managed by different frontend teams made up of external developers, junior devs, and a mix of others. If we leave SSE integration up to each team, it’s likely to get pretty messy and inconsistent.

To keep things clean and simple, we’re planning to build an internal library that handles SSE, published the npm registry. We’re committed to using React Query because we love its interface, documentation, and overall approach. Our goal is to create a layer on top of React Query that hides the complexity of SSE but still feels familiar to our developers.

Usage of useQueryStream

 const { data, isLoading } = useQueryStream({
    queryKey: ["grid-stream", queryKey],
    queryFn: mySseImplementation,
  });

The streamQueryBehavior Behavior

Here’s a simplified version of what we’re working on:

export function streamQueryBehavior(client) {
let unsubscribe;
  return {
    onFetch: (context, query: Query) => {
      let cancelled = false;

      const addSignalProperty = (object: unknown) => {
        Object.defineProperty(object, 'signal', {
          enumerable: true,
          get: () => {
            if (context.signal.aborted) {
              cancelled = true;
            } else {
              context.signal.addEventListener('abort', () => {
                cancelled = true;
              });
            }

            return context.signal;
          },
        });
      };
      const fetchFn = async () => {
        const queryFn = ensureQueryFn(context.options, context.fetchOptions);
        
        unsubscribe?.();
        
        const fetchPage = () => {
          if (cancelled) {
            return Promise.reject();
          }
          const { promise, resolve, reject } = Promise.withResolvers();

          const params = {
            next: (newData) => {
              query.setData(newData);
              query.setState({
                ...query.state,
                isComplete: false,
              });
              const cache = client.getQueryCache();

              // Notify cache callback
              cache.config.onSuccess?.(data, query);
              cache.config.onSettled?.(
                data,
                query.state.error,
                query,
              );
            },
            error: (error) => {
              reject(error);
            },
            complete: () => {
              query.setState({
                ...query.state,
                isComplete: !cancelled,
              });
              resolve(query.state.data);
            },
          };

          addSignalProperty(params);
          unsubscribe = queryFn(params);

          return promise;
        };

        return fetchPage();
      };

      context.fetchFn = fetchFn;
    },
  };
}

Why We’re Doing This

We want to make SSE integration super easy for our devs, especially those who aren’t as familiar with it. Our main goals are to:

1. Simplify SSE integration for all our teams.
2. Ensure consistency and reliability across our products.
3. Leverage React Query’s awesome features and keep a consistent mental model.

This is the status

• simple POC.
• re-fetching when the queryKey changes.
• select and enabled
• validate idea and the direction.
• Do invalidate / reset / remove / cancel queries work as expected? (in codesandbox there is a demo about them)
• suspense and ErrorBoundary... ???

How You Can Help

We’d really appreciate any feedback you can give us on this approach. Specifically:

1. Given that this is a simple POC 😄 ... Do you see any potential issues or improvements for our useQueryStream hook?
2. Are there any better ways to integrate SSE with React Query while keeping a similar API?

Thanks a ton for your time and for any insights you can share!

Best,
Marco

---

Below you can find the sandbox with the implementation
https://codesandbox.io/p/sandbox/use-subscription-query-83d72d?file=%2Fsrc%2Fgrid-stream.tsx%3A5%2C2-8%2C6

[TkDodo]

gonna tag @KATT and @juliusmarminge here because I think they have an SSE implementation in trpc 👀

[marcog83]

Thanks a lot for answering 😃

[KATT]

The short answer: you could use tRPC if you use TS/JS on the backend

The longer answer - I'll mainly link code

This is a finite stream (not SSE) / more for LLM responses etc:

• tl;dr we have a useQuery() that has a .data property which has an array which is all the results of the
• We update the query result in the queryFn like this: https://github.com/trpc/trpc/blob/de90f34436465eb9b883593ba840acd18fb70900/packages/react-query/src/shared/hooks/createHooksInternal.tsx#L189-L209
• You can see how the status/fetchStatus updates while it's running here: https://github.com/trpc/trpc/blob/de90f34436465eb9b883593ba840acd18fb70900/packages/tests/server/react/useQuery.test.tsx#L265-L274

With subscriptions through SSE, we have a custom hook - we don't use React Query for it as it doesn't really make sense as we don't want to cache any results etc

• https://github.com/trpc/trpc/blob/next/packages/client/src/links/httpSubscriptionLink.ts
• https://github.com/trpc/trpc/blob/de90f34436465eb9b883593ba840acd18fb70900/packages/client/src/internals/TRPCUntypedClient.ts#L134-L167
• https://github.com/trpc/trpc/blob/de90f34436465eb9b883593ba840acd18fb70900/packages/react-query/src/shared/hooks/createHooksInternal.tsx#L305-L349

There's still some to be desired in it - we don't return the state and you just get an onError-callback when it errors etc

I don't think you want to use React Query for SSE - caching data for an infinite list of subscriptions seems wasteful. You rather just want a callback function when new data comes in.

Make sure to tell your backend team to implement the last-event-id-part so your clients don't miss any event and have a way for the client to set it at the first request.

You can see the usage of our SSE-implementation on https://github.com/trpc/examples-next-sse-chat

I don't have time to guide you more than this, hopefully it's helpful.

[marcog83]

Wow, thanks for the quick answer.
Unfortunately our BE is Java :)
For sure those insights will be helpful. i'll dig into your code to figure out similarities and differences, and where i can take "inspiration" 😄 .

I think our POC moved the focus from SSE to Observables in general, and we envision SSE like one implementation for Observables.

We are interested in React Query cache capability because our use case is to fetch a big amount of data in chunks.
BE emits a complete event when all the data are sent.
So, given a set of filters, the endpoint returns for example 20.000 records. FE then renders/paginate them in a ag-grid table. instead of waiting for the whole dataset to be processed, the BE can start emit events whenever a chunk is ready to be consumed.
If the the entire dataset is downloaded by the FE, then we can re-use the cached values. (until invalidate/remove/cancel )

Well... what to say... i really appreciate your answers, i'll do my homeworks and take all the learnings i can from your code.
And please. if you see that this approach is too naive or it can be a source of headhache feel free to comment here. ( better to fail fast )

[KATT]

Sounds like you are doing a finite result and not an infinite one.

SSE isn't the right solution, as it's designed for infinite results, but rather something similar to my first reference.

We use jsonl for this. Our implementation is a bit hacky (b/c we combine it with request batching too) but you can have a peek here https://github.com/trpc/trpc/blob/next/packages/server/src/unstable-core-do-not-import/stream/jsonl.ts

[marcog83]

I studied your implementation. thanks for sharing the exact lines.
About SSE I'll have a round with BE team, maybe we can work with jsonl. I didn't know about it, but it seems it can fit 100%.
I think that both implementation are doing almost the same thing, in 2 different way. ( this is why i love coding ).

The main point is to reach this point at retryer.ts, where we need a promise to be resolved somehow.

Where config.fn is the fetchFn of the context from the query.

In your implementation you
- kept the QueryObserver and
- played inside the queryFn to resolve the promise after the async for loop

In our implementation we
- created a new StreamQueryObserver with behavior to overwrite the fetchFn ( similar to InfiniteObserver )
- Played inside the fetchFn to resolve the promise when the complete is called. We pass the callbacks to queryFn that is responsible to do the subscription.

So just to recap my findings...
I think i learnt more about my code, by understanding your 😄 (and i simplified it)
and as usual any feedback or comments are welcome