Supporting a stream-based flow (EventSource, SSE)

[thomasbertet]

Hi!

Just curious if you would be interested in supporting a new use case for streamed data.
For instance, it could support EventSource.

From what I can foresee, this would "look-like" the infinite query flow - with the difference being that it's not the front-end pulling new data but the server sending it intermittently.

Also, just wanted to say I find the doc really nice 👍 . Great job!

[tannerlinsley]

Let me start with this... if you can write a custom hook that uses the useInfiniteQuery hook and does this with an event source, then we can have a better discussion on whether something like this should belong in React Query, or simply as a new npm module like react-query-use-event-source

[thomasbertet]

have a better discussion on whether something like this should belong in React Query, or simply as a new npm module

This is actually my question - sorry for not being clear.

if you can write a custom hook that uses the useInfiniteQuery hook and does this with an event source

I'll try to experiment with that and post here the result. As I understand your answer, it might help guide the answer to my question =)

[nimo23]

Let me start with this... if you can write a custom hook that uses the useInfiniteQuery hook and does this with an event source, then we can have a better discussion on whether something like this should belong in React Query or simply as a new npm module like react-query-use-event-source

@tannerlinsley And what if Fetch+Readable Stream are used for the queryFn? I wonder why SSE (and WebSockets) aren't yet an integral part of tanstack-query, since it can also benefit from its asynchronous state management (e.g., structural release, tracked properties, etc.).

For example, with SSE, why would the client consider data stale if the server knows it isn't? Currently, tanstack-query's refetching approach is limited to a more or less heuristic approach to near-real-time delivery. If the user wants to use SSE, with SSE support (e.g., through useSSEQuery), tanstack-query can simplify the setup of SSE endpoints (e.g., by using an internally Fetch+Readable stream, by implicitly setting gcTime: Infinity, staleTime: Infinity, custom refetching strategies, and all the other optimizations that enable seamless use of SSE in tanstack-query— it's complex to know, which settings need to be tweaked in relation to SSE, you know better :). I think, especially for SSE, the "don't call us (=server), we call you (=client)" approach has many advantages (e.g., no unnecessary refetching from the server, no need to transmit unmodified data, etc.).

Tanstack-Query is all about asynchronous state management, and SSE (and WebSockets) are also established web standards that should also benefit from the advantages of asynchronous state management provided by Tanstack-Query. Tanstack Query provides https://tanstack.com/query/latest/docs/reference/streamedQuery - maybe it can also provide SSE support. Even though it's been 5 years, you should consider adding SSE support to the Tanstack query.

For inspiration:

• https://github.com/joshmossas/event-source-plus
• https://github.com/StyleShit/react-use-sse (but uses only EventSource)

Btw: The many posts below and various attempts to integrate SSE into Tanstack-Query prove that this is not trivial. Therefore, it would be useful to simplify this use case by having something likeuseSSEQuery in Tanstack-Query.

[TkDodo]

for ReadableStream, see:

• https://tanstack.com/query/v5/docs/reference/streamedQuery

[tannerlinsley]

Moving this to discussions as this is a feature request.

[thomasbertet]

Thanks, not used to have this discussion thing - sorry =)

[thomasbertet]

So I tried this that works OK, but I feel there is a better way:

The API would be something like:

const { status, data }  = useEventSourceQuery(myQueryKey, myEventSourceURL, myEventName);

as for the implementation I thought of something like:

import * as React from 'react';
import { useInfiniteQuery} from 'react-query';

interface CustomQueryParams {
    eventData: string[];
}

type Status = "success" | "loading" | "error";

export const useEventSourceQuery = (key: string, url: string, eventName: string) => {

    const eventSource = React.useRef<EventSource>(new EventSource(url));

    const queryFn = React.useCallback((_: any, params: CustomQueryParams) => {
        if (!params) { return Promise.resolve([])}
        return Promise.resolve(params.eventData);
    }, [])
    
    const  { data, fetchMore } = useInfiniteQuery<string[], string, CustomQueryParams>(key, queryFn as any, { getFetchMore: () => ({ eventData: []})})

    const customStatus = React.useRef<Status>('success'); 

    React.useEffect(() => {
        const evtSource = eventSource.current;

        const onEvent = function (ev: MessageEvent | Event) {
            if (!e.data) {
                return;
            }
            // Let's assume here we receive multiple data, ie. e.data is an array.
            fetchMore({ eventData: e.data });
        }
        const onError = () => { customStatus.current = 'error' };

        evtSource.addEventListener(eventName, onEvent);
        evtSource.addEventListener('error', onError);

        return () => {
            evtSource.removeEventListener(eventName,onEvent);
            evtSource.removeEventListener('error', onError);
        }

    }, [url, eventName, fetchMore])

    return {status: customStatus.current, data };
}

Some critical thoughts regarding this: I'm still unsure if this is the way to go. Does the URL and event name should live in this hook, like is it something this hook should handle OR should we pass some kind of async generator function so that the hook does not even know about EventSource - ie. it could work for anything async other than simple promise.

WDYT ?

[tannerlinsley]

Hmmm. Reworking the lib to handle generators too would be a massive breaking change and probably cause a lot of bloat for such a slim use case. The question I'm wondering about is if you think it would be possible to first, make custom hook that uses generators and wraps useInfiniteQuery. If that is possible, then you could make another custom hook off of that that could handle an event source.

Whether or not these hooks would make it into the core of the library... I'm not sure. But that doesn't mean we can't build them.

[thomasbertet]

Soo based on your latest suggestion, I made two rough things:

One using async generator to pull data:

export const useAyncGeneratorQuery = (key: string, asyncGeneratorFn: () => Generator<string[], string[], unknown>) => {

    const queryFn = React.useCallback((_: any, params: CustomQueryParams) => {
        if (!params) { return Promise.resolve([])}
        return Promise.resolve(params.eventData);
    }, [])
    
    const  { data, fetchMore } = useInfiniteQuery<string[], string, CustomQueryParams>(key, queryFn as any, { getFetchMore: () => ({ eventData: []})})

    const customStatus = React.useRef<Status>('success'); 

    React.useEffect(() => {
        (async function doReceive() {
            try { 
                for await (let data of asyncGeneratorFn()) {
                    fetchMore({ eventData: data });
                }
            } catch (e) {
                customStatus.current = 'error';
            }
        })();

    }, [asyncGeneratorFn, fetchMore])

    return {status: customStatus.current, data };
}

and one using simple callback paradigm:

type CallbackType = (data: string[]) => void;

type InitCallbackType = (cb: CallbackType) => void;

export const useCallbackQuery = (key: string, initCallbackQuery: InitCallbackType) => {

    const queryFn = React.useCallback((_: any, params: CustomQueryParams) => {
        if (!params) { return Promise.resolve([])}
        return Promise.resolve(params.eventData);
    }, [])
    
    const  { data, fetchMore } = useInfiniteQuery<string[], string, CustomQueryParams>(key, queryFn as any, { getFetchMore: () => ({ eventData: []})})

    const callback = React.useCallback((data) => { fetchMore({ eventData: data })}, [fetchMore]);

    React.useEffect(() => {  initCallbackQuery(callback); }, [callback, initCallbackQuery] )
    
    const customStatus = React.useRef<Status>('success'); 

    return {status: customStatus.current, data };
}

I said rough because both are not handling errors and/or maybe edge-cases. They also have an initial empty array.

You can check the live code there https://codesandbox.io/s/interesting-bhaskara-zw2dd?file=/src/App.tsx . Can you think of anything that would break things / edge cases I should handle ?

Also, when clicking on the sandbox browser pane, everything seems to be cleared somehow - not sure if related to react-query but seems weird nonetheless.

[JuanLuisGarciaBorrego]

Any new about this ?
Thanks!

[madelineurl]

Also curious!

[rmcsharry]

Did you go ahead with this @thomasbertet ? If so, some update on how it faired would be very welcome :)

[thomasbertet]

Hi! Sorry for the late update to you all. Unfortunately I don't have update on this. This was an experiment that we did not use in production. So I guess you can try to take this as an inspiration but really no guarantee! Sorry!

[madelineurl]

No worries, thanks for the update and original inspiration!

[edbrett]

I have found a simpler solution using the queryClient API and updating the data from the event source events. Maybe this is an anti pattern for react-query but it works nicely for me. Something along the lines of:

import { useQuery, useQueryClient } from 'react-query'

export const useEventSourceQuery = (queryKey, url, eventName) => {
  const queryClient = useQueryClient()

  const fetchData = () => {
    const evtSource = new EventSource(url)

    evtSource.addEventListener(eventName, (event) => {
      const eventData = event.data && JSON.parse(event.data)

      if (eventData) {
        queryClient.setQueryData(queryKey, eventData)
      }
    })
  }

  return useQuery(queryKey, fetchData)
}

[TkDodo]

why would you do that with useQuery instead of with a normal useEffect ? The problem I see with this approach is that the queryFn can run multiple times due to background refetches, and each time it does, you're leaking an event source and a listener, because there is no way to clean it up inside the queryFn. If you have a query that has a normal queryFn that points to an endpoint that gives you data for your queryKey, you can just use that, and additionally, setup a useEffect that updates via setQueryData. It's pretty much the same concept I have outlined for websockets here: https://tkdodo.eu/blog/using-web-sockets-with-react-query

[edbrett]

Initially I did try this with a useEffect but ran in to some problems, the main one being reusability. In my use case I had I need to call the same hooks that depended on this event source for quite a few components across many views of my application, so in order to do so I needed then to implement some form of global state/context for tracking the event source, and then caching the responses. useState wasn't going to work and the rest of our application depended on react query as our data cache and a universal way to invalidate. react-query gives this for free with the above method through its own context. As for the event source leaking, my example above was a simplification of the actual implementation. I have an additional event listener to close the stream. I'll take a look at the blog!

[reisblucas]

@edbrett Could you share how do you implements the additional event listener to close the stream?

[chalscc]

@edbrett Could you bring more insight on the full implementation?

[dan-cooke]

Just for anyone landing here today, most browsers now support fetch streams

So there is no need to use EventSource anymore, you can simply use fetch in your react query

Here's a package from Microsoft that does exactly this

https://github.com/Azure/fetch-event-source

So you could just swap out your fetch function for this one, and you get SSE support inside react query

[TimLinsenmayer]

Going to chime in here: fetch-event-source was last released on Apr 25, 2021 with quite a number of open issues and PRs. While it's nice as a test ground, I wouldn't implement this in a production environment.
Probably a similar version can be achieved with SSE.js (which is more maintained) - sadly due to time constraints I can't whip up a poc of it...

[reisblucas]

I'm so sorry that I didn't bring my solution in time.

I've questioned @edbrett above to share some example of how do we close the stream using the React Query and started to build my solution whitout using it as @TkDodo suggested.

Below, we can see that I achieve how to control and open/close some Event Source using the eventSources object as Singleton to control my SSE, also didn't refactored it to be a Class, to be used with React Query and didn't abstract things that shouldn't be in this file.

So, I hope it could help you guys to achieve your solutions and reaffirming: this was a work solution as a method to guarantee that ONE and ONLY ONE instance must be opened at a time from some Event as SSE in 2023.

import { alertService } from '@/alerts'

interface EventSources {
  [key: string]: EventSource
}

type AllowedEventServices = 'alerts' | 'chat' | 'scheduling'

// should work and behave as singleton
const eventSources: EventSources = {}

const defaultSSEBaseRoute = 'https://www.myurlthatshouldbesubscribed.com'
const SSEBaseUrl: Record<string, string> = {
  alerts: `${defaultSSEBaseRoute}/alert/events/subscribe`,
} as const

function isExistentConnection(baseUrl: string): EventSource | undefined {
  return eventSources[baseUrl]
}

async function getToken(allowedEventService: AllowedEventServices) {
  switch (allowedEventService) {
    case 'alerts':
      return alertService.eventToken()

    default:
      return {
        token: 'default',
        expiresAt: 0,
      }
  }
}

async function open(allowedEventService: AllowedEventServices) {
  const token = (await getToken(allowedEventService)).token
  const baseUrl = SSEBaseUrl[allowedEventService]
  const existentConnection = isExistentConnection(baseUrl)

  const sourceUrl = `${baseUrl}?sse_token=${token}`

  // for some reason our first render opne the connection exist and the readyState is 2
  if (existentConnection) {
    existentConnection.close()
  }

  // open connection the first time
  const source = new EventSource(sourceUrl)
  eventSources[baseUrl] = source
  return source
}

function getAllSources() {
  return { ...eventSources }
}

function close(url: string) {
  const source = eventSources[url]

  if (source) {
    source.close()
    delete eventSources[url]
  }
}

function closeAll() {
  for (const url in eventSources) {
    if (eventSources.hasOwnProperty(url)) {
      eventSources[url].close()
      delete eventSources[url]
    }
  }
}

export const EventSourceFactory = {
  open,
  getAllSources,
  close,
  closeAll,
}

This runs in my useEffect from NotificationListener component that runs at the top of my application, in my case it's _app.tsx from Next.js, and it fed our custom toast notification real-time in UI, with color, text, alert, etc...

// NotificationListener
// ...
useEffect(() => {
    async function alertAsyncEventSource() {
      const sourceOpenAction = async () => EventSourceFactory.open('alerts')
      const source = await sourceOpenAction()

      AlertEventTypes.forEach((eventType) => {
        source.addEventListener(eventType, toastHandler)

        source.onerror = (e: Event) =>
          handleAlertSourceOnError(e, source, sourceOpenAction, handleReconnectionSource, toastHandler)
      })

      return () => {
        source.close()

        AlertEventTypes.forEach((eventType) => {
          source.removeEventListener(eventType, toastHandler)
          reconnectionSource?.removeEventListener(eventType, toastHandler)
        })

        return
      }
    }

    alertAsyncEventSource()
    // eslint-disable-next-line react-hooks/exhaustive-deps
  }, [])
// ...

[nimo23]

I don't understand the following:

Just for anyone landing here today, most browsers now support fetch streams

So there is no need to use EventSource anymore, you can simply use fetch in your react query

I don't understand how Streams (https://streams.spec.whatwg.org/#readablestream) can replace EventSource. Both are different things, right? How can you claim that using EventSource is no longer necessary and can be replaced by Streams?

[kacetal]

@nimo23 Absolutely, you use your nativefetch and accept content type text/event-stream and server should keep connection open.
You can implement it or use this library, for example: eventsource

[nimo23]

Ok, I think the main benefit of Fetch+Readable Stream is the ability to use all the features provided by the FetchAPI.

[loarsaw]

https://tkdodo.eu/blog/using-web-sockets-with-react-query

export const useQuerySubscription = () => {
    const { data, error } = useQuery({ queryKey: [`todo`] })
    const queryClient = useQueryClient()
    useEffect(() => {
        const eventSource = new EventSource(`url`);
        eventSource.addEventListener("open", () => {
            console.log("opened")
        })
        eventSource.addEventListener("PUT", (event) => {
            const queryData = event.data && JSON.parse(event.data);
            queryClient.setQueriesData({ queryKey: [`todo`] }, queryData)
        })
        return () => {
            eventSource.close()
        }
    }, [queryClient])


    return {
        data: useMemo(() =>{
        
        }, [data]), error
    }
}