Managers y Middleware

Reactive Data Client usa el patrón de store flux, que se caracteriza por ser fácil de entender y depurar gracias al flujo de datos unidireccional del store. Las actualizaciones de estado las realiza una función reducer.

En las arquitecturas flux, es fundamental que todas las funciones del ciclo flux sean puras. Los Managers proporcionan una orquestación centralizada de los efectos secundarios. En otras palabras, son el medio para interactuar con el mundo exterior a Data Client.

Por ejemplo, NetworkManager orquesta la obtención de datos y SubscriptionManager lleva el registro de qué recursos están suscritos con useLive o useSubscription. Al centralizar el control, NetworkManager deduplica automáticamente los fetches, y SubscriptionManager mantendrá actualizados solo los recursos que se están renderizando activamente.

Esto convierte a los Managers en la mejor forma de integrar efectos secundarios adicionales como registro (logging), reporte de errores, métricas, notificaciones, flujos de datos, actualización al recuperar el foco o la conexión, sincronización entre pestañas y persistencia sin conexión. También se pueden personalizar para cambiar comportamientos centrales.

Managers por defecto
NetworkManager	Convierte los dispatches de fetch en llamadas de red
SubscriptionManager	Maneja las suscripciones de sondeo (polling)
DevToolsManager	Habilita la depuración
Managers adicionales
LogoutManager	Maneja el HTTP 401 (u otras condiciones de cierre de sesión)

Ejemplos

Reactive Data Client mejora la seguridad de tipos y la ergonomía al realizar los dispatches y el acceso al store con su Controller

Registro (logging) con middleware

import type { Manager, Middleware } from '@data-client/react';

export default class LoggingManager implements Manager {
  middleware: Middleware = controller => next => async action => {
    console.log('before', action, controller.getState());
    await next(action);
    console.log('after', action, controller.getState());
  };

  cleanup() {}
}

Reporte de errores

Reporta los fetches fallidos a servicios de monitoreo como Sentry inspeccionando las acciones SET_RESPONSE con error definido.

import {
  type Manager,
  type Middleware,
  actionTypes,
} from '@data-client/react';
import { captureException } from '@sentry/react';

export default class ErrorReportManager implements Manager {
  middleware: Middleware = controller => next => async action => {
    if (action.type === actionTypes.SET_RESPONSE && action.error)
      captureException(action.response, {
        extra: { endpoint: action.endpoint.name, args: action.args },
      });
    return next(action);
  };

  cleanup() {}
}

Métricas

Mide los tiempos de los fetches observando las acciones FETCH. action.meta.promise se resuelve cuando el fetch termina.

import {
  type Manager,
  type Middleware,
  actionTypes,
} from '@data-client/react';
import { trackTiming } from './analytics';

export default class MetricsManager implements Manager {
  middleware: Middleware = controller => next => async action => {
    if (action.type === actionTypes.FETCH) {
      const start = performance.now();
      action.meta.promise
        .finally(() => {
          trackTiming(action.endpoint.name, performance.now() - start);
        })
        // the fetch's caller handles errors; this only observes timing
        .catch(() => {});
    }
    return next(action);
  };

  cleanup() {}
}

Notificaciones (toasts)

Muestra un toast cuando cualquier mutación tiene éxito o falla.

import {
  type Manager,
  type Middleware,
  actionTypes,
} from '@data-client/react';
import { toast } from './toast';

export default class ToastManager implements Manager {
  middleware: Middleware = controller => next => async action => {
    if (
      action.type === actionTypes.SET_RESPONSE &&
      action.endpoint.sideEffect
    ) {
      if (action.error) toast.error(`${action.endpoint.name} failed`);
      else toast.success(`${action.endpoint.name} succeeded`);
    }
    return next(action);
  };

  cleanup() {}
}

Actualizar al recuperar el foco o la conexión

Controller.expireAll() marca los datos como obsoletos, lo que dispara una nueva obtención de cualquier dato que se esté renderizando activamente sin suspender (stale-while-revalidate). init() y cleanup() administran los listeners de eventos.

import type { Manager, Middleware, Controller } from '@data-client/react';

export default class RefreshManager implements Manager {
  declare protected controller: Controller;
  protected handle = () =>
    this.controller.expireAll({ testKey: () => true });

  middleware: Middleware = controller => {
    this.controller = controller;
    return next => async action => next(action);
  };

  init() {
    window.addEventListener('focus', this.handle);
    window.addEventListener('online', this.handle);
  }

  cleanup() {
    window.removeEventListener('focus', this.handle);
    window.removeEventListener('online', this.handle);
  }
}

Sincronización entre pestañas

Cuando una mutación tiene éxito en una pestaña, marca los datos como obsoletos en todas las demás pestañas usando BroadcastChannel.

import {
  type Manager,
  type Middleware,
  actionTypes,
} from '@data-client/react';

export default class TabSyncManager implements Manager {
  protected channel = new BroadcastChannel('data-client');

  middleware: Middleware = controller => {
    this.channel.onmessage = () =>
      controller.expireAll({ testKey: () => true });
    return next => async action => {
      if (
        action.type === actionTypes.SET_RESPONSE &&
        action.endpoint.sideEffect &&
        !action.error
      )
        this.channel.postMessage('mutation');
      return next(action);
    };
  };

  cleanup() {
    this.channel.close();
  }
}

Persistencia sin conexión

Persiste el store con IndexedDB (aquí mediante idb-keyval); restáuralo con el initialState de DataProvider. Las escrituras en IndexedDB son asíncronas y usan structured clone en lugar de bloquear el hilo principal con la serialización JSON, como haría localStorage. Aplicar debounce a las escrituras mantiene económicas las ráfagas rápidas de acciones. Ten en cuenta los tiempos de caducidad al restaurar.

import type { Manager, Middleware } from '@data-client/react';
import { set } from 'idb-keyval';

export default class PersistManager implements Manager {
  declare protected timer?: ReturnType<typeof setTimeout>;

  middleware: Middleware = controller => next => async action => {
    await next(action);
    // debounce: persist at most once per second
    clearTimeout(this.timer);
    this.timer = setTimeout(() => {
      // in-flight optimistic updates reference functions, so are not persistable
      const state = { ...controller.getState(), optimistic: [] };
      set('data-client', state);
    }, 1000);
  };

  cleanup() {
    clearTimeout(this.timer);
  }
}

index.tsx

import { DataProvider, getDefaultManagers } from '@data-client/react';
import { createRoot } from 'react-dom/client';
import { get } from 'idb-keyval';
import App from './App';
import PersistManager from './PersistManager';

const managers = [...getDefaultManagers(), new PersistManager()];
const initialState = await get('data-client');

createRoot(document.body).render(
  <DataProvider initialState={initialState} managers={managers}>
    <App />
  </DataProvider>,
);

Flujo de datos con middleware (basado en push)

Agregar un manager que procese los datos enviados por el servidor mediante websockets o Server Sent Events garantiza que podamos mantener los datos actualizados cuando las actualizaciones son independientes de la acción del usuario. Por ejemplo, el precio en una aplicación de trading, o un editor colaborativo en tiempo real.

import type {
  Manager,
  Middleware,
  Controller,
  EntityInterface,
} from '@data-client/react';

export default class StreamManager implements Manager {
  declare protected controller: Controller;
  declare protected evtSource: WebSocket | EventSource;
  declare protected createEventSource: () => WebSocket | EventSource;
  declare protected entities: Record<string, EntityInterface>;

  constructor(
    createEventSource: () => WebSocket | EventSource,
    entities: Record<string, EntityInterface>,
  ) {
    this.createEventSource = createEventSource;
    this.entities = entities;
  }

  middleware: Middleware = controller => {
    this.controller = controller;
    return next => async action => next(action);
  };

  connect() {
    this.evtSource = this.createEventSource();
    this.evtSource.onmessage = (event: MessageEvent) => {
      try {
        const msg: { type: string; args: [any]; data: any } = JSON.parse(
          event.data,
        );
        if (msg.type in this.entities)
          this.controller.set(
            this.entities[msg.type],
            ...msg.args,
            msg.data,
          );
      } catch (e) {
        console.error('Failed to handle message');
        console.error(e);
      }
    };
  }

  init() {
    this.connect();
  }

  cleanup() {
    this.evtSource?.close();
  }
}

Controller.set() permite actualizar directamente los Schemas Querables con event.data.

Agrupar actualizaciones de alta frecuencia

Los flujos como los tickers de un exchange pueden enviar cientos de mensajes por segundo, y las conexiones suelen comenzar con una instantánea grande. En lugar de llamar a set() por cada mensaje, guárdalos en un búfer y escribe cada lote con un schema Array. Controller.set([Entity], rows) normaliza todas las filas en una sola actualización del store.

export default class StreamManager implements Manager {
  // ...
  protected buffer: Record<string, any[]> = {};
  declare protected flushTimeout?: ReturnType<typeof setTimeout>;

  connect() {
    this.evtSource = this.createEventSource();
    this.evtSource.onmessage = event => {
      const msg = JSON.parse(event.data);
      if (msg.type in this.entities) {
        (this.buffer[msg.type] ??= []).push(msg.data);
        this.flushTimeout ??= setTimeout(this.flush, 50);
      }
    };
  }

  flush = () => {
    const buffer = this.buffer;
    this.buffer = {};
    this.flushTimeout = undefined;
    for (const type in buffer) {
      this.controller.set([this.entities[type]], buffer[type]);
    }
  };

  cleanup() {
    this.evtSource?.close();
    clearTimeout(this.flushTimeout);
    this.flushTimeout = undefined;
    this.buffer = {};
  }
}

Las filas de un mismo lote que comparten una pk se combinan en orden y omiten Entity.shouldReorder(), así que guarda en el búfer solo el último mensaje por pk cuando el orden importa.

Prueba ambos botones a continuación. Esta comprobación en el navegador parte de un store vacío y mide el tiempo de un Promise.all de 500 llamadas a set() frente a un único set() por lotes. Ambos caminos son un solo commit de React, y cada uno escribe 500 precios nuevos.

Ticker

PriceStream

import React from 'react';
import { useController, useQuery } from '@data-client/react';
import { Ticker, newPrices } from './Ticker';

function PriceStream() {
  const ctrl = useController();
  const [timing, setTiming] = React.useState('');
  const first = useQuery(Ticker, { product_id: 'COIN-0' });

  const time = async (
    label: string,
    write: (rows: ReturnType<typeof newPrices>) => Promise<unknown>,
  ) => {
    const rows = newPrices();
    const start = performance.now();
    await write(rows);
    setTiming(`${label}: ${(performance.now() - start).toFixed(1)} ms`);
  };
  const perRow = () =>
    time('500 set() calls', rows =>
      Promise.all(
        rows.map(row => ctrl.set(Ticker, { product_id: row.product_id }, row)),
      ),
    );
  // highlight-next-line
  const batch = () => time('1 batch set()', rows => ctrl.set([Ticker], rows));

  return (
    <div>
      <button onClick={perRow}>set() per row</button>{' '}
      <button onClick={batch}>batch set()</button>
      <p>COIN-0: {first ? `$${first.price}` : 'no data yet'}</p>
      <p>{timing}</p>
    </div>
  );
}
render(<PriceStream />);

Resultado

Store▶

Omitir DevTools en actualizaciones de alta frecuencia

Al usar WebSockets u otras fuentes de datos en tiempo real, quizás quieras omitir el registro de ciertas acciones de alta frecuencia en DevToolsManager para evitar saturar la extensión del navegador.

import { getDefaultManagers, actionTypes } from '@data-client/react';
import StreamManager from './StreamManager';
import { Ticker } from './Ticker';

export default function getManagers() {
  return [
    new StreamManager(() => new WebSocket('wss://ws-feed.example.com'), {
      ticker: Ticker,
    }),
    ...getDefaultManagers({
      devToolsManager: {
        // Increase latency buffer for high-frequency updates
        latency: 1000,
        // Skip WebSocket SET actions to avoid log spam
        // (batched writes use the [Ticker] schema)
        predicate: (state, action) =>
          action.type !== actionTypes.SET ||
          (action.schema !== Ticker && action.schema[0] !== Ticker),
      },
    }),
  ];
}

Aplicación Coin

Explora el ejemplo coin-app

Abrir en StackBlitzVer código fuente

Loading demo…

More Demos