nd-json and react-query

[alex-jesper]

Hi,
I have discovered nd-json, which basically streams json objects in a reactive fashion.

Each sent element is valid json, but collectively it is not. A toy example:

{"seqNo":0,"timestamp":"2022-08-25T12:54:54.199366799Z"}
{"seqNo":1,"timestamp":"2022-08-25T12:54:55.199331163Z"}
{"seqNo":2,"timestamp":"2022-08-25T12:54:56.199361384Z"}
{"seqNo":3,"timestamp":"2022-08-25T12:54:57.199364156Z"}
{"seqNo":4,"timestamp":"2022-08-25T12:54:58.199366290Z"}

Is it possible to process each of these objects as they arrive using react-query?

Thanks,
Jesper

[TkDodo]

At the moment, not natively, no. Your queryFn needs to return a promise. You can create a promise that sets up a subscription via nd-json, concatenate the inputs and resolve the promise once you have everything.

[alex-jesper]

Hi,
I made a solution that inserts elements in the cache in the queryFn as the elements arrive. This gives pretty much the behaviour I was looking for.

The code is in the repo here:
https://github.com/alex-jesper/react-query-ndjson

These are the main components as they look atm:

import Row from "../model/Row";
import { useQuery, useQueryClient, UseQueryResult } from "react-query";
const ndjsonStream = require("can-ndjson-stream");

export const useElements = (): UseQueryResult<Row[], unknown> => {
  const queryClient = useQueryClient();
  return useQuery<Row[]>(["elements"], async () => {
    const response = await fetch("http://localhost:8080", {
      headers: {
        accept: "application/x-ndjson",
      },
    });
    const ndjson = ndjsonStream(response.body)
    const reader = ndjson.getReader()

    while (true) {
      const { done, value } = await reader.read()
      if (done) {
        break;
      }

      // Update state
      queryClient.setQueryData(["elements"], (oldData: Row[] | undefined) => {
        return Array.isArray(oldData) ? [...oldData, value] : [value];
      });
    }
    reader.releaseLock();
    const data = queryClient.getQueryData<Row[]>(["elements"])

    return Array.isArray(data) ? data : [];
  });
};

Using the query is done as:

import { useElements } from '../api/Api';

export const NdJsonElements: FC = () => {
    const { data } = useElements()

    return (
        <Box>
            <Typography variant='h4'>Streamed ndjson Elements:</Typography>
            <TableContainer>
                <Table>
                    ...
                    <TableBody>
                        {data?.map((value) => {
                            return (
                                <TableRow key={value.timestamp}>
                                    <TableCell>
                                        {value.seqNo}
                                    </TableCell>
                                    <TableCell>
                                        {new Date(value.timestamp).toISOString()}
                                    </TableCell>
                                </TableRow>
                            )
                        })}
                    </TableBody>
                </Table>
            </TableContainer>
        </Box>
    )
}

Hope this is useful to someone.

@TkDodo Any thoughts about this approach?

Thanks,
Jesper

[alex-jesper]

One problem with this is with the abort/signal mechanism.

When a component is unmounted have the option of:

• Not consuming the abort signal, causing the request to continue
• Consuming the signal causing the request to stop and the cache to be cleared

For this streaming approach I would like to cancel the fetch, but not clear the cache. In other words: stop the transfer of new elements, but keep the ones I already got.

That need is not meaning full in a typical fetch call, where its either done or not.

Is there a way to get a abort mechanism that keeps the cache?

Thanks,
Jesper

[alex-jesper]

This is turning into a monologue, sorry about that. For the sake of someone being in the same situation in the future I will update a bit.

I solved the abort problem by using the abort signal from react-query to cancel my fetch. Which causes a reset of data. The reset of data is unwanted as elements I have already fetched are valid. To avoid that reset, I keep a separate cache for temporary fetched elements, these elements are used as basis for the next fetch.

The effect is that cancel works, and the elements fetched before the cancel is cached, which seems most logical.

The code linked previously is updated, to reflect.

Thanks,
Jesper

[TkDodo]

regarding the aborting: react-query provides the abort signal, which you can forward to axios. This will abort the fetch. You can further add your own event listener and listen to the abort event of that signal: signal.addEventListener('abort', abortListener)

when this event occurs, you probably should stop the while loop I guess.

[alex-jesper]

Thanks for the input about the event listener. It does not seem to be necessary to exit the loop as an the exception kills it. At least no new elements are received after unmount.