Streamlining multiple streaming update endpoints via WebSocket

[simonvizzini]

I've implemented streaming updates in our application via the onCacheEntryAdded callback as described in the documentation.

This requires quite a bit of boilerplate code, especially if you have multiple topics you want to subscribe to via WebSockets this results in a lot of code duplication.

I studied the documentation and code to see how this could be more streamlined. So far the only option I have found is to create a factory function for onCacheEntryAdded. What I don't like about this solution is the difficulty of typing this factory function properly. I had to import a few types from RTKs /dist folder, which means they are not part of the publicly exported API of the module and may change anytime in the future without notice.

Here is the implementation I came up with so far:

import { BaseQueryFn } from "@reduxjs/toolkit/query";
import { getConnection } from "./connection";
// Importing types form "dist" is a smell
import { QueryCacheLifecycleApi } from "@reduxjs/toolkit/dist/query/endpointDefinitions";
import { MaybeDrafted } from "@reduxjs/toolkit/dist/query/core/buildThunks";

function onCacheEntryAddedFactory<ResultType, QueryArg>(opts: {
 method: string;
 hubUrl: string;
 updateCachedData?(
   // TODO: figure out a way to properly type this
   // eslint-disable-next-line
   data: any,
   draft: MaybeDrafted<ResultType>
 ): void | MaybeDrafted<ResultType>;
}) {
 return {
   async onCacheEntryAdded(
     _arg: QueryArg,
     {
       updateCachedData,
       cacheDataLoaded,
       cacheEntryRemoved,
     }: QueryCacheLifecycleApi<QueryArg, BaseQueryFn, ResultType, string>
   ) {
     const connection = await getConnection(opts.hubUrl);

     const listener = (data: any) => {
       updateCachedData((draft) => {
         if (opts?.updateCachedData) {
           return opts.updateCachedData(data, draft);
         } else {
           return data;
         }
       });
     };

     try {
       await cacheDataLoaded;
     } catch {
       // Do nothing
     }

     connection.on(opts.method, listener);
     await cacheEntryRemoved;
     connection.off(opts.method, listener);
   },
 };
}

export { onCacheEntryAddedFactory };

Which can then be used as follows:

subscribeStatusUpdates: build.query<StatusUpdateMessage[], void>({
  queryFn() {
    return { data: [] };
  },
  ...onCacheEntryAddedFactory({
    method: "GetStatusUpdate",
    hubUrl: SIGNALR_HUB_URL,
    updateCacheData: (data: StatusUpdateMessage, draft) => {
      draft.push(data);
    },
  }),
}),

This works but providing a custom updateCacheData function is a bit awkward, as you need two arguments: the data from the event as well as the draft, and the types might not be the same.

How does everyone else solve multiple streaming update endpoints, without having to repeat too much boilerplate code?