[#128] Make market data provider generic

apps/trading/pages/markets/index.page.tsx

@@ -8,7 +8,6 @@ import { MarketListTable, getRowNodeId } from '@vegaprotocol/market-list';
 import {
   Markets_markets,
   Markets_markets_data,
-  MarketsDataProviderCallbackArg,
   marketsDataProvider,
 } from '@vegaprotocol/graphql';
 
@@ -24,47 +23,43 @@ const Markets = () => {
   const initialized = useRef<boolean>(false);
 
   useEffect(() => {
-    return marketsDataProvider(
+    return marketsDataProvider(client, ({ data, error, loading, delta }) => {
-      client,
+      setError(error);
-      ({ data, error, loading, delta }: MarketsDataProviderCallbackArg) => {
+      setLoading(loading);
-        setError(error);
+      if (!error && !loading) {
-        setLoading(loading);
+        if (!initialized.current || !gridRef.current) {
-        if (!error && !loading) {
+          initialized.current = true;
-          if (!initialized.current || !gridRef.current) {
+          setMarkets(data);
-            initialized.current = true;
+        } else {
-            setMarkets(data);
+          const update: Markets_markets[] = [];
-          } else {
+          const add: Markets_markets[] = [];
-            const update: Markets_markets[] = [];
-            const add: Markets_markets[] = [];
 
-            // split into updates and adds
+          // split into updates and adds
-            if (!gridRef.current || !delta) return;
+          if (!gridRef.current) return;
+          const rowNode = gridRef.current.api.getRowNode(
+            getRowNodeId(delta.market)
+          );
 
-            const rowNode = gridRef.current.api.getRowNode(
+          if (rowNode) {
-              getRowNodeId(delta.market)
+            const updatedData = produce(
+              rowNode.data.data,
+              (draft: Markets_markets_data) => merge(draft, delta)
             );
-
+            if (updatedData !== rowNode.data.data) {
-            if (rowNode) {
+              update.push({ ...rowNode.data, data: delta });
-              const updatedData = produce(
-                rowNode.data.data,
-                (draft: Markets_markets_data) => merge(draft, delta)
-              );
-              if (updatedData !== rowNode.data.data) {
-                update.push({ ...rowNode.data, data: delta });
-              }
-            }
-            // @TODO - else add new market
-            if (update.length || add.length) {
-              gridRef.current.api.applyTransactionAsync({
-                update,
-                add,
-                addIndex: 0,
-              });
             }
           }
+          // @TODO - else add new market
+          if (update.length || add.length) {
+            gridRef.current.api.applyTransactionAsync({
+              update,
+              add,
+              addIndex: 0,
+            });
+          }
         }
       }
-    );
+    });
   }, [client, initialized]);
 
   return (
@@ -72,7 +67,7 @@ const Markets = () => {
       {(data) => (
         <MarketListTable
           ref={gridRef}
-          markets={data}
+          data={data}
           onRowClicked={(id) =>
             push(`${pathname}/${id}?portfolio=orders&trade=orderbook`)
           }

libs/graphql/src/data-providers/generic-data-provider.ts

@@ -0,0 +1,125 @@
+import { produce } from 'immer';
+import type { Draft } from 'immer';
+import type {
+  ApolloClient,
+  DocumentNode,
+  TypedDocumentNode,
+} from '@apollo/client';
+import type { Subscription } from 'zen-observable-ts';
+
+export function makeDataProvider<QueryData, Data, SubscriptionData, Delta>(
+  query: DocumentNode | TypedDocumentNode<QueryData, any>, // eslint-disable-line @typescript-eslint/no-explicit-any
+  subscriptionQuery: DocumentNode | TypedDocumentNode<SubscriptionData, any>, // eslint-disable-line @typescript-eslint/no-explicit-any
+  update: (draft: Draft<Data>[] | null, delta: Delta) => void,
+  getData: (subscriptionData: QueryData) => Data[] | null,
+  getDelta: (subscriptionData: SubscriptionData) => Delta
+) {
+  type C = (arg: {
+    data: Data[] | null;
+    error?: Error;
+    loading: boolean;
+    delta?: Delta;
+  }) => void;
+  const callbacks: C[] = [];
+  const updateQueue: Delta[] = [];
+
+  let data: Data[] | null = null;
+  let error: Error | undefined = undefined;
+  let loading = false;
+  let client: ApolloClient<object> | undefined = undefined;
+  let subscription: Subscription | undefined = undefined;
+
+  const notify = (callback: C, delta?: Delta) => {
+    callback({
+      data,
+      error,
+      loading,
+      delta,
+    });
+  };
+
+  const notifyAll = (delta?: Delta) => {
+    callbacks.forEach((callback) => notify(callback, delta));
+  };
+
+  const initialize = async () => {
+    if (subscription) {
+      return;
+    }
+    loading = true;
+    error = undefined;
+    notifyAll();
+    if (!client) {
+      return;
+    }
+    subscription = client
+      .subscribe<SubscriptionData>({
+        query: subscriptionQuery,
+      })
+      .subscribe(({ data: subscriptionData }) => {
+        if (!subscriptionData) {
+          return;
+        }
+        const delta = getDelta(subscriptionData);
+        if (loading) {
+          updateQueue.push(delta);
+        } else {
+          const newData = produce(data, (draft) => {
+            update(draft, delta);
+          });
+          if (newData === data) {
+            return;
+          }
+          data = newData;
+          notifyAll(delta);
+        }
+      });
+    try {
+      const res = await client.query<QueryData>({ query });
+      data = getData(res.data);
+      if (updateQueue && updateQueue.length > 0) {
+        data = produce(data, (draft) => {
+          while (updateQueue.length) {
+            const delta = updateQueue.shift();
+            if (delta) {
+              update(draft, delta);
+            }
+          }
+        });
+      }
+    } catch (e) {
+      error = e as Error;
+      subscription.unsubscribe();
+      subscription = undefined;
+    } finally {
+      loading = false;
+      notifyAll();
+    }
+  };
+
+  const unsubscribe = (callback: C) => {
+    callbacks.splice(callbacks.indexOf(callback), 1);
+    if (callbacks.length === 0) {
+      if (subscription) {
+        subscription.unsubscribe();
+        subscription = undefined;
+      }
+      data = null;
+      error = undefined;
+      loading = false;
+    }
+  };
+
+  return (c: ApolloClient<object>, callback: C) => {
+    if (!client) {
+      client = c;
+    }
+    callbacks.push(callback);
+    if (callbacks.length === 1) {
+      initialize();
+    } else {
+      notify(callback);
+    }
+    return () => unsubscribe(callback);
+  };
+}

libs/graphql/src/data-providers/markets-data-provider.ts

@@ -1,8 +1,6 @@
 import { gql } from '@apollo/client';
-import { produce } from 'immer';
-import type { ApolloClient } from '@apollo/client';
-import type { Subscription } from 'zen-observable-ts';
 import { Markets, Markets_markets } from '../__generated__/Markets';
+import { makeDataProvider } from './generic-data-provider';
 
 import {
   MarketDataSub,
@@ -57,137 +55,25 @@ const MARKET_DATA_SUB = gql`
   }
 `;
 
-export interface MarketsDataProviderCallbackArg {
+export const marketsDataProvider = makeDataProvider<
-  data: Markets_markets[] | null;
+  Markets,
-  error?: Error;
+  Markets_markets,
-  loading: boolean;
+  MarketDataSub,
-  delta?: MarketDataSub_marketData;
+  MarketDataSub_marketData
-}
+>(
-
+  MARKETS_QUERY,
-export interface MarketsDataProviderCallback {
+  MARKET_DATA_SUB,
-  (arg: MarketsDataProviderCallbackArg): void;
+  (draft: Markets_markets[] | null, delta: MarketDataSub_marketData) => {
-}
+    if (!draft) {
-
+      return;
-const callbacks: MarketsDataProviderCallback[] = [];
-const updateQueue: MarketDataSub_marketData[] = [];
-
-let data: Markets_markets[] | null = null;
-let error: Error | undefined = undefined;
-let loading = false;
-let client: ApolloClient<object> | undefined = undefined;
-let subscription: Subscription | undefined = undefined;
-
-const notify = (
-  callback: MarketsDataProviderCallback,
-  delta?: MarketDataSub_marketData
-) => {
-  callback({
-    data,
-    error,
-    loading,
-    delta,
-  });
-};
-
-const notifyAll = (delta?: MarketDataSub_marketData) => {
-  callbacks.forEach((callback) => notify(callback, delta));
-};
-
-const update = (
-  draft: Markets_markets[] | null,
-  delta: MarketDataSub_marketData
-) => {
-  if (!draft) {
-    return;
-  }
-  const index = draft.findIndex((m) => m.id === delta.market.id);
-  if (index !== -1) {
-    draft[index].data = delta;
-  }
-  // @TODO - else push new market to draft
-};
-
-const initialize = async () => {
-  if (subscription) {
-    return;
-  }
-  loading = true;
-  error = undefined;
-  notifyAll();
-  if (!client) {
-    return;
-  }
-  subscription = client
-    .subscribe<MarketDataSub>({
-      query: MARKET_DATA_SUB,
-    })
-    .subscribe(({ data: delta }) => {
-      if (!delta) {
-        return;
-      }
-      if (loading) {
-        updateQueue.push(delta.marketData);
-      } else {
-        const newData = produce(data, (draft) => {
-          update(draft, delta.marketData);
-        });
-        if (newData === data) {
-          return;
-        }
-        data = newData;
-        notifyAll(delta.marketData);
-      }
-    });
-  try {
-    const res = await client.query<Markets>({
-      query: MARKETS_QUERY,
-    });
-    data = res.data.markets;
-    if (updateQueue && updateQueue.length > 0) {
-      data = produce(data, (draft) => {
-        while (updateQueue.length) {
-          const delta = updateQueue.shift();
-          if (delta) {
-            update(draft, delta);
-          }
-        }
-      });
     }
-  } catch (e) {
+    const index = draft.findIndex((m) => m.id === delta.market.id);
-    error = e as Error;
+    if (index !== -1) {
-    subscription.unsubscribe();
+      draft[index].data = delta;
-    subscription = undefined;
-  } finally {
-    loading = false;
-    notifyAll();
-  }
-};
-
-const unsubscribe = (callback: MarketsDataProviderCallback) => {
-  callbacks.splice(callbacks.indexOf(callback), 1);
-  if (callbacks.length === 0) {
-    if (subscription) {
-      subscription.unsubscribe();
-      subscription = undefined;
     }
-    data = null;
+    // @TODO - else push new market to draft
-    error = undefined;
+  },
-    loading = false;
+  (responseData: Markets): Markets_markets[] | null => responseData.markets,
-  }
+  (subscriptionData: MarketDataSub): MarketDataSub_marketData =>
-};
+    subscriptionData.marketData
-
+);
-export const marketsDataProvider = (
-  c: ApolloClient<object>,
-  callback: MarketsDataProviderCallback
-) => {
-  if (!client) {
-    client = c;
-  }
-  callbacks.push(callback);
-  if (callbacks.length === 1) {
-    initialize();
-  } else {
-    notify(callback);
-  }
-  return () => unsubscribe(callback);
-};