Throttling for the WS listeners

[present-g]

Hi! Is there a recommended way to implement throttling for WebSocket listeners?

This is an important feature for handling high-frequency WS data. Should throttling support multiple modes, such as:

• Latest / sampling — only the latest message within the interval is processed.
• Batching — all messages received within the interval are accumulated and passed to the listener as a single batch.

This would allow us to skip intermediate updates for state-based data while ensuring that no messages are lost for event-based data.

Is this model supported, and what would be the recommended way to implement it?

[NotAFlightRisk]

Not built in, as of @hyper-fetch/sockets 9.0.0. The WebSocket adapter takes no listener options, and every message calls every listen callback straight away (adapter.ts). socket.onMessage won't do it either, it has to return data so it can't drop or hold a message.

It's a small wrapper round listener.listen though, and one buffer covers both modes:

import type { ListenerInstance, ExtractListenerResponseType } from "@hyper-fetch/sockets";

export const listenEvery = <L extends ListenerInstance>(
  listener: L,
  ms: number,
  onFlush: (messages: ExtractListenerResponseType<L>[]) => void,
) => {
  let buffer: ExtractListenerResponseType<L>[] = [];
  let timer: ReturnType<typeof setTimeout> | undefined;

  const stop = listener.listen(({ data }) => {
    buffer.push(data);
    timer ??= setTimeout(() => {
      const messages = buffer;
      buffer = [];
      timer = undefined;
      onFlush(messages);
    }, ms);
  });

  return () => {
    clearTimeout(timer);
    stop();
  };
};

// batching - every message, grouped per 100ms
listenEvery(onTrade, 100, (trades) => handleTrades(trades));

// latest/ sampling - just the last one per 100ms
listenEvery(onPrice, 100, (prices) => setPrice(prices.at(-1)!));

Its a trailing throttle, so the first message waits up to ms, and calling the returned function drops anything still bufered.

[present-g]

Unfortunately, wrapping the listener seems to be too late in the pipeline. At that point, we would lose integration with features such as useListener and other listener-related APIs.

Ideally, buffering or throttling should happen earlier, before events are propagated through the listener pipeline. That would preserve the existing HyperFetch APIs while also reducing unnecessary processing and resource usage for high-frequency WebSocket streams.

From the API perspective, it would be great if this could be configured directly on createListener, for example:

const listener = socket.createListener()({
  endpoint: "messages",
  options: {
    delivery: {
      interval: 16,
      strategy: "latest",
    },
  },
});

This way, useListener(listener), listener.listen(), and the rest of the listener API could continue to work unchanged, while the listener itself defines how frequently updates should be propagated.

[present-g]

@prc5, hi, do you have any suggestions about it?

[prc5]

I will solve this directly in the listener 👍🏻
Thanks guys for your feedback, you're the best!

[prc5]

@present-g @NotAFlightRisk

Done ✅ #147

It's a delivery option on the listener, almost exactly as you sketched it (top-level instead of inside options, since options is adapter-specific):

// sampling - first message immediately, then the newest one per interval
const onPrice = socket.createListener<Price>()({
  topic: "prices",
  delivery: { strategy: "latest", interval: 16 },
});

// batching - everything from the window as one array, nothing dropped
const onTrade = socket.createListener<Trade>()({
  topic: "trades",
  delivery: { strategy: "batch", interval: 100, maxSize: 500 },
});
onTrade.listen(({ data }) => {}); // data: Trade[] - inferred

// anything else - custom strategy
delivery: createDeliveryStrategy<Tick, Summary>(({ deliver }) => ({ push, dispose }))

// change at runtime, returns a clone like setOptions
onPrice.setDelivery({ strategy: "latest", interval: 1000 });

It runs inside the listener, right before your callback, so listen(), useListener, SDK defaults and every adapter work unchanged, and there's zero overhead when you don't set it. Types follow the strategy (batch → Response[], also in useListener().data). Unsubscribing drops pending messages, so nothing fires after unmount.

Guides for sockets and React are in the PR too. Ships with the next sockets/react release, would love to hear how it works for your streams.

[NotAFlightRisk]

Awesome, thank you so much @prc5 💛

[present-g]

@prc5, thank you so much for your incredible engagement - you are the best! We're going to migrate a big project and start a new one with HyperFetch.

I'll check this approach over the next few days and share my feedback.

I sincerely wish that HyperFetch becomes the most popular React approach for working with network requests. HyperFetch deserves it, and you deserve it!

[prc5]

Thank you guys 🥹

Feel free to open as many issues as you need if something feels off - I will handle this immediately 🫡 🙏🏻

[present-g]

Hi! I've just finished testing the solution. It's amazing! In my synthetic environment, I hit 1M msgs/sec with only 1–10 calls per second for listeners and 120 fps.

I was a bit skeptical at first about the extra steps for every single message after receiving it. But after testing alternatives like JSON with prefixes, protobuf, msgpackr, etc., I realized there's not much variants for optimization.

For example, @bufbuild/protobuf performed almost the same or slower than JSON.parse due to JSON's low-level engine optimization, compared to protobuf's encode/decode and schema handling overhead.