Server Sent Events support in Remix

[rossipedia]

One of the areas that Remix is lacking currently is in support for real-time data from the server.

It's common these days to implement real-time features on top of Web Sockets. They're a robust, if somewhat technically complicated solution.

I believe a full Web Socket abstraction in Remix is out-of-scope (for now), but there's an alternative method that I think fits Remix's philosophy quite nicely:

Server Sent Events

If you're already familiar with what Server Sent Events ("SSE") are, please skip ahead to the next section.

In a nutshell, SSE is a way of informing the browser that an event happened on the server using a long-running request with a specific content type: text/event-stream. They're kind of like "formalized long-polling", but unlike regular long polling the browser will not time these requests out if they take too long.

SSE responses from the server are standard HTTP responses:

HTTP/1.1 200 OK
Cache-Control: no-cache, no-transform
Content-Type: text/event-stream

event: message
data: hello from sse

event: ping
data: ping

data: without an 'event' field it will default to 'message'

In the browser, the EventSource API can be used to subscribe to these events:

let source = new EventSource('/my-event-source');
source.addEventListener('ping', (event) => {
  console.log(`Got ping from server!`);
});

source.addEventListener('message', (event) => {
  console.log(`Got )
});

For a more in depth discussion and details, see MDN's guide on SSE.

SSE in Remix

I think SSE fits a niche in Remix for simpler real-time use-cases. For example: a status update from a back-end CI/CD process. Things that require notification from the server, but don't need the development overhead of a Web Socket integration.

I propose that Remix supports a serverEvents export from Route modules. The type for this export would look like this:

interface ServerEventsFunction {
  (args: DataFunctionArgs, send: SendFunction):
    | void
    | Promise<void>
    | CleanupFunction
    | Promise<CleanupFunction>
    | AsyncIterator<any>;
}

type SendFunction = (message: any, meta: {
  // text/event-stream things
  event?: string; // defaults to 'message',
  id?: string; // the event id
  retry?: number; // the reconnection time

  // this is allowed as an escape-hatch incase the raw data you're
  // sending is already JSON encoded, since it will be parsed on the client
  // if true, `message` will be JSON.stringify-ed before sending
  // defaults to `true`
  json?: boolean;
});

type CleanupFunction = () => (void | Promise<void>);

A few things to unpack there:

First off, the ServerEventsFunction export receives the same args parameter that loader and action do. However, it also receives a second send parameter that is used to send events to the client.

The serverEvents handler can also respond to the connection being closed when the client no longer needs it (eg: a component consuming these events unmounts, or the tab is closed, etc).

Much like useEffect, the handler can return a cleanup function. For example, say you wanted to hook up to a Redis Pub/Sub channel:

export let serverEvents: ServerEventsFunction = async (args, send) => {
  let subscriber = await getRedisConnection();
  subscriber.subscribe("channel", (message) => {
    // Assume messages are already being published as JSON
    send(message, { json: false });
  });
  return () => subscriber.quit();
};

The cleanup function can be synchronous or asynchronous.

Another way,of sending events would be to return an AsyncIterable. The most straightforward method of doing that would be to use an async generator function:

export let serverEvents: ServerEventsFunction = async function* ({ request }) {
  let count = 0;
  while (true) {
    await new Promise((resolve) => setTimeout(resolve, 1000));
    if (request.signal.aborted) {
      break;
    }
    yield count;
  }
};

When returning an AsyncIterable, you can't also return a cleanup function, so instead we rely on the AbortSignal that's attached to the request to tell when to clean up.

Instead of checking aborted, one can also call addEventListener('abort', ...) on the signal instead/as well.

The various methods of sending events can be combined, as well:

export let serverEvents: ServerEventsFunction = async function*({request}, send) {
  let subscriber = await getRedisConnection();
  send({message: 'connection established', from: 'system'});
  subscriber.subscribe('channel', (message) => {
    // Assume messages are already being published as JSON
    send(message, {json: false});
  });
  request.signal.addEventListener('abort', () => subscriber.quit());
  while (true) {
    if (request.signal.aborted) {
      break;
    }
    await new Promise((resolve) => setTimeout(resolve, 1000));
    yield ({message: 'ping', from: 'system'});
  }
}

Consuming the serverEvents export.

As this is a Route module export, we need some way of distinguishing a request for SSE from a "UI" request (like navigating to the route in your browser) or a "data" request (loaders/actions). We can use the "magic" querystring key _serverevents much in the same way that loader/action requests use _data to tell Remix that it shouldn't server render the UI.

On the client, a useServerEvents (or possibly useServerEventsData) hook will be available in @remix-run/react.

In its simplest form, it will provide the "last event received" as React state, and the component will re-render whenever a new event arrives:

function MyCounter() {
  let count = useServerEvents<number>();
  return (
    <div>
      Count: {count}
    </div>
  )
}

The hook will default to using the current route's serverEvents as the source, and listen for the default message event.

However, you can customize the behavior by passing an options object:

function ChatList() {
  const [messages, setMessages] = useState<ChatMessage[]>([]);
  useServerEvents<ChatMessage>({
    // Must be a valid route. That route's `serverEvents` export will be used
    // Defaults to the current route
    /** @type {string} */
    source: `/chat`,
    // The event to subscribe to
    /** @type {string} */
    event: `chat-message`,
    // Handle the message directly using onEvent. In this case, the state value
    // returned by the hook isn't automatically set, but can be set using
    // the second parameter of the `onEvent` callback.
    onEvent(event, setState) {
      // do something with the raw event here
      // `event` is the immediate output of JSON.parse on the raw event data
      setMessages(previous => [...previous, event]);
    },
    // whether or not to include credentials/cookies in the request
    // passed directly to the EventSource constructor
    // optional: defaults to `undefined` (false)
    withCredentials: true;
  },
   // You can pass extra dependencies here for `onEvent` for fine grained control of its lifetime
  [setMessages]);

  return (
    <ChatMessageList>
      {messages.map((msg) => <ChatMessage {...msg} />)}
    </ChatMessageList>
  )
}

In either case, the raw event data will be parsed using JSON.parse before handling.

Under the hood, this hook manages instances of EventSource and subscribes/unsubscribes handlers for consuming clients.

Some notes on behavior:

• Much like loaders and actions, JSON is the standardized non-UI data format used. Events are JSON.stringify-ed when sent, and JSON.parse-ed when received.
• The underlying EventSource instances are unique based on source + withCredentials. This means that you can use multiple useServerEvent hooks for the same source, and at most two EventSource instances will be created (they are reference counted by source).

This will help to avoid hitting the rather severe limits that SSE have around multiple connections:

Only 6 connections to a given domain may be open at a time. This is across all open tabs of the browser.

(For more information, see the warning here.)

Implementation Status

I have the above actually functioning, and I will be creating a PR for further implementation discussion. Here's the sample videos I posted in Discord to demonstrate my POC (ignore the names, they're old):

Screen_Recording_2022-03-29_at_1.45.38_AM.mov

Screen_Recording_2022-03-29_at_9.48.44_PM.mov

Currently there's some patterns I've had to use that will need more robust solutions than the.. er... "hacky" methods I used.

I'm curious to know what the Remix team and community at large think of this, and whether or not it would be a valuable addition to the framework.

Thanks for reading!

[ryanhiebert]

This looks awesome!

As this is a Route module export, we need some way of distinguishing a request for SSE from a "UI" request (like navigating to the route in your browser) or a "data" request (loaders/actions). We can use the "magic" querystring key _serverevents much in the same way that loader/action requests use _data to tell Remix that it shouldn't server render the UI.

I wonder if the browser gives us an indication that this is an EventSource request in the Accept header?

[sergiodxa]

What you're describing is working around inconsistent implementations in the platform or the ecosystem

The problem is that this is an issue with the CDN implementations, if you only cached on browsers you could add the header Vary: Accept to the response and now browsers will have a different cache key for the HTML and JSON or SSE request/response match.

So the issue is CDNs are not always supporting the Vary header as they should, but this is not an issue all CDNs have, and some CDNs even have worst issues, like Netlify CDN ignoring search params when caching.

I think, CDNs may eventually fix this and Remix could drop the ?_data search param, which would be possible since that is an implementation detail of Remix.

[kiliman]

Ideally Remix would have used _remix prefix for its query strings. As we add more cases, I'm concerned of naming collisions. It would be reasonable to say: Remix owns anything starting with _remix.

[rossipedia]

@ryanhiebert I'm fine to have slightly differing priorities here :) I personally value consistency in this case, but regardless I think the fact that we cannot rely on CDNs to do the right thing here is forcing it on us. What I would truly prefer is to ditch magic querystrings altogether, and use content negotiation exclusively. Hopefully that becomes a viable option sooner rather than later.

[ryanhiebert]

Me too. This needs a caniuse page or something. I wouldn't be surprised if some kind of a page like that exists already, even.

[ryanhiebert]

Here's an explanation of why a lot of HTML stuff doesn't rely on Content Negotiation: https://wiki.whatwg.org/wiki/Why_not_conneg

[smerchek]

First of all, I did not know about Server Sent Events and presumed that Websockets were just the only way to solve this... so, thanks for "using the platform" and teaching me something new!

For UDisc, I think this would perfect for our most common use-case, displaying real-time scores.

For example, https://udisclive.com follows the progress of a tournament by listening for scorecard changes. There are really only two scenarios:

1. The tournament is finished: use the data loader to send all the data necessary to display the historical results.
2. The tournament is active: use the data loader to get the point-in-time results and subscribe to updates for the tournament, processing each new scorecard update to update the results in real-time.

I'd expect to use Mongo ChangeStreams for this as our first implementation, since we already have these changes flowing through Mongo.

Overall, I think this is a perfect fit and wanted to put this use-case out there to help validate the proposal.

One question/concern:

How would I ensure that I don't miss any updates between the data loader point-in-time and the subscription starting for new events?

While it's probably not likely, I always have to think through these edge cases. I think I would approach this by returning the most recent scorecard timestamp in the data loader (or derive this from the results) and then pass that timestamp to the subscription for the event stream. So, the event stream would send all events that happened since timestamp plus any new events that come in via the change stream.

[sergiodxa]

What you say of sending the timestamp is what I would do too. Another option is to only fetch data client side for active tournaments instead of using the route loader.

[ryanhiebert]

You're right that implementations will need to consider how to synchronize things. Basically by timestamp is right, or some other time-based complete ordering of things. I don't think that we need to solve it here, exactly, it's probably a higher level concern, but yeah it definitely needs to be coordinated in many cases. Some cases won't need it, if the page only shows ephemeral events, but I think that use-case is probably less typical than wanting to stream something that does also have a persistent record.

[smerchek]

Ya, that all sounds right. 👍

I agree it doesn't really impact the proposed design in any way, but could be helpful to point out in docs or in one of the examples/stacks.

[afoures]

when learning about SSE, i stumbled across this article https://dev.to/miketalbot/server-sent-events-are-still-not-production-ready-after-a-decade-a-lesson-for-me-a-warning-for-you-2gie.

what are your thoughts about this ?

[smerchek]

Thanks for pointing this out. I have some similar questions as some other commenters on there.

Is this a concern only for internal networks, non-HTTPS traffic, or unsupported browsers like IE11?

https://caniuse.com/eventsource - there is pretty wide support, excluding IE11.

I think for use-cases that need to consider these types of scenarios, it is worth considering carefully. However, I'd be quite curious to learn how seriously to take these concerns for modern applications (e.g. required SSL, HTTP2, public cloud, etc).

[ryanhiebert]

From the comments there, it appears that it’s for non-tls connections, and there may be a reasonable workaround to degrade SSE to disconnect after each send to force delivery and reconnection, which would have it working very similar to long polling.

it’s good to keep these issues in mind, though, so they’re easier to spot if we run into this kind of trouble.

[rossipedia]

Since the latest version has support for proper web streams, I think this discussion is moot at this point.

[epeterson320]

I was able to get SSE working in a loader function with the following approach. This was implemented in a resource route's loader but it could possibly be implemented in a regular route's loader if it respected hypermedia controls, responding only to a specific value in the accept header (if it didn't, the main component would never load).

export const loader: LoaderFunction = async ({ request }) => {
  const body = new ReadableStream({
    async start(c) {
      c.enqueue(": sending messages\n\n");
      const subscription = someObservable.subscribe(
        (item) => {
          c.enqueue("event: item\n");
          c.enqueue(`data: ${JSON.stringify(item)}\n\n`);
        },
        (error) => {
          c.error(error);
        }
      );
      request.signal.onabort = () => {
        subscription.unsubscribe();
        c.close();
      };
    },
  });

  const headers = new Headers();
  headers.set("Cache-Control", "no-store, no-transform");
  headers.set("Content-Type", "text/event-stream");
  return new Response(body, { headers, status: 200 });
};

[tylermmorton]

Thanks for sharing this. I was able to get a resource route working with an event stream like you've shown above.

Do you know how one would use this type of resource route with the useFetcher API? When I try this the fetcher is always in the loading state.

[epeterson320]

Since useFetcher is built for a request/response model, it would probably be more effort than it's worth to use it for the publish/subscribe model of SSE.

How you use the resource will depend greatly on your use case. Do you only care about the most recent event? The most recent 20? Will there be thousands, and of different types? Keyed by ID, with partial updates? Trigger side effects?

But for the basic "subscribe to most recent value" case, you can use or adapt this hook:

/**
 * Subscribe to server sent events and return the latest one.
 */
function useLatestEvent<T>(eventSourceUrl: string, initialValue: T): T {
  const [latestEvent, setLatestEvent] = useState(initialValue);

  useEffect(() => {
    const eventSource = new EventSource(eventSourceUrl, {
      withCredentials: true // if you need it
    });

    const onMessage = (messageEvent: MessageEvent) => {
      const event = JSON.parse(messageEvent.data);
      setLatestEvent(event); // validate it first, if you're into that
    });

    eventSource.addEventListener("message", onMessage);

    return () => {
      eventSource.removeEventListener("message", onMessage);
      eventSource.close();
    };
  }, [eventSourceUrl]);

  return latestEvent;
}

[tylermmorton]

Awesome!! Thank you 🚀

I was able to get this working with some minor modifications. I had missed this part in your original post, but when enqueuing data into the ReadableStream, you must format it as data: ${JSON.stringify(payload)}\n\n in order for the EventSource to pick it up on the other side.

Resource route loader:

export const loader: LoaderFunction = async ({ request, params }) => {
  const id = `stream_${Math.random().toString(16).slice(2)}`;
  const headers = new Headers();

  headers.set("Connection", "keep-alive");
  headers.set("Content-Type", "text/event-stream");
  headers.set("Cache-Control", "no-store, no-transform");

  const stream = new ReadableStream({
    async start(streamController) {
      console.log(`[${id}] Starting`);

      request.signal.onabort = () => {
        console.log(`[${id}] Abort Signal Received - Closing`);
        streamController.close();
      };

      var i = 0;
      async function pump(): Promise<void> {
        await new Promise((r) => setTimeout(r, 2000));

        const payload = {
          content: "test_" + i++,
        };

        console.log(`[${id}] Sending payload...`);
        streamController.enqueue(`data: ${JSON.stringify(payload)}\n\n`);

        return pump();
      }
      return pump();
    },
  });

  return new Response(stream, {
    headers,
    status: 200,
  });
};

And the hook (I updated it to use an array for a chat-like application)

function useDataStream<T>(eventSourceUrl: string, initialValue: T[]): Array<T> {
  const [eventData, setEventData] = useState(initialValue);

  useEffect(() => {
    console.log("creating event source");
    const eventSource = new EventSource(eventSourceUrl, {
      withCredentials: true, // if you need it
    });

    eventSource.onmessage = (event: MessageEvent) => {
      console.log("message received: " + event.data);
      const payload = JSON.parse(event.data);
      setEventData((eventData) => [...eventData, payload]);
    };

    return () => {
      eventSource.close();
    };
  }, [eventSourceUrl]);

  return eventData;
}

[ryanflorence]

Now that we've got SSE support, it's a bit cumbersome to use, in my limited experience it looks like this:

Set up some pub/sub on the server

import { EventEmitter } from "events";
export let emitter = new EventEmitter();

Set up an event stream with cleanup and queues and stuff that subscribes to it and streams the events when new stuff comes through:

import { emitter } from "../some-emitter.server";

type InitFunction = (send: SendFunction) => CleanupFunction;
type SendFunction = (event: string, data: string) => void;
type CleanupFunction = () => void;

export function eventStream(request: Request, init: InitFunction) {
  let stream = new ReadableStream({
    start(controller) {
      let encoder = new TextEncoder();
      let send = (event: string, data: string) => {
        controller.enqueue(encoder.encode(`event: ${event}\n`));
        controller.enqueue(encoder.encode(`data: ${data}\n\n`));
      };
      let cleanup = init(send);

      let closed = false;
      let close = () => {
        if (closed) return;
        cleanup();
        closed = true;
        request.signal.removeEventListener("abort", close);
        controller.close();
      };

      request.signal.addEventListener("abort", close);
      if (request.signal.aborted) {
        close();
        return;
      }
    },
  });

  return new Response(stream, {
    headers: { "Content-Type": "text/event-stream" },
  });
}

Return the event stream from a loader in a resource route:

import { eventStream } from "./event-stream";

export let loader: LoaderFunction = ({ request }) => {
  return eventStream(request, send => {
    emitter.addListener("messageReceived", handleChatMessage);

    function handleChatMessage(chatMessage: string) {
      send("message", chatMessage);
    }

    return () => {
      emitter.removeListener("messageReceived", handleChatMessage);
    };
  });
};

Push into the event emitter in actions:

import { emitter } from "./some-emitter.server";

export let action: ActionFunction = async ({ request }) => {
  let formData = await request.formData();
  emitter.emit("messageReceived", formData.get("something");
  return { ok: true };
};

And finally, set up an EventSource in the browser

function useEventSource(href: string) {
  let [data, setData] = useState("");

  useEffect(() => {
    let eventSource = new EventSource(href);
    eventSource.addEventListener("message", handler);

    function handler(event: MessageEvent) {
      setData(event.data || "unknown");
    }

    return () => {
      eventSource.removeEventListener("message", handler);
    };
  }, []);

  return data;
}

I'd love to see some community helpers around this stuff for some happy-path SSE workflows, particularly the eventStream function and useEventSource.

My code up there is stuff I cobbled together looking at @jacob-ebey's demos and mdn docs, but I literally have zero experience with SSE otherwise.

I'd love to see a community effort around a package that adds a little abstraction over all of these APIs to make SSE in a Remix a breeze 🙏

[dan-cooke]

I tested on 1.7.2 @ryanflorence and my loaders seem to freeze up when returning text/event-stream . Is there another build I should be testing this on?

Edit: I'm just noticing it has to be a resource route

[dan-cooke]

I created a simple library for SSE that implements most of the patterns discussed in this thread.

https://github.com/dan-cooke/remix-sse

There is plenty of room for improvement so PR's are welcome!

I will be using this library in production so no doubt I will be adding features over time and mantaining it.

[kiliman]

Awesome. Definitely the best way to improve an API is to use it in anger.

[wehriam]

I have also put together a small library - @wehriam/remix-sse

In /apps/routes/counter.ts:

import { SseResponse } from "@wehriam/remix-sse";

export let loader = ({ request }) => {
  const response = new SseResponse(request);

  let count = 0;

  const interval = setInterval(() => {
    response.send("counter", { count });
    count += 1;
  }, 1000);

  function handleAbort() {
    clearInterval(interval);
  }

  response.signal.addEventListener("abort", handleAbort);

  return response;
};

In /apps/routes/index.tsx:

import { useSse } from "@wehriam/remix-sse";

export default function Index() {
  const data = useSse<{ count: number }>("counter", "/counter");
  return (
    <div>
      <h1>SSE Counter Example</h1>
      <h2>Count: {data ? data.count : "Unknown"}</h2>
    </div>
  );
}

[davyrolink]

For everyone interested and unaware of the following; two utilities for server-sent events has been added to https://github.com/sergiodxa/remix-utils#server-sent-events

[intuji-kirankumal]

Has anyone successfully implemented Server-Sent Events (SSE) using Shopify Hydrogen and hosted on Shopify Oxygen? I'm encountering a 'promise hanging' error during implementation. Any insights or best practices to resolve this issue and potential challenges would be greatly appreciated.

[phaux]

It would be nice if you could pass an Observable or AsyncIterable to defer and it would turn into an auto-updating value on the client side when passed to Await:

const kv = await Deno.openKv()

export async function loader() {
  const updates = kv.watch<Room>(["rooms", 123]) // returns AsyncIterator
  return defer({ updates })
}

export default function RoomPage() {
  const { updates } = useLoaderData<typeof loader>()

  return (
    <Suspense fallback={<Skeleton />}>
      <Await resolve={updates}>
        {(room) => 
          <RoomView data={room} /> // automatically updates
        }
      </Await>
    </Suspense>
  )
}

[ivosabev]

That would be the dream.