Real-Time Communication

Firefly supports WebSockets, Server-Sent Events (SSE), and streaming JSON responses.

WebSockets

Use WS.handler to create a WebSocket endpoint:

let chat =
    WS.handler (fun conn req -> task {
        while conn.IsOpen do
            match! conn.Receive() with
            | WsText msg ->
                do! conn.Send $"Echo: {msg}"
            | WsBinary data ->
                do! conn.SendBytes data
            | WsClose ->
                ()
    })
 
Route.start
|> Route.get "/ws/chat" chat

WsConn API

Method	Description
`conn.Send(text)`	Send a text message
`conn.SendBytes(data)`	Send binary data
`conn.Receive()`	Receive the next message (`WsText`, `WsBinary`, or `WsClose`)
`conn.Close(?status, ?reason)`	Close the connection
`conn.IsOpen`	Check if the connection is still open
`conn.CancellationToken`	Token cancelled when the client disconnects

The handler automatically accepts the WebSocket upgrade, manages the connection lifecycle, and closes the socket when the function returns. Non-WebSocket requests receive a 400 response.

Rooms and broadcast (`WsHub<'T>`)

For chat, notifications, or presence, use a WsHub<'T> to broadcast a typed message to many clients. Each connection joins a room; messages of type 'T are JSON-serialized on the wire automatically.

type ChatMsg = { User: string; Text: string }
 
let chat = WsHub<ChatMsg>()
 
let routes =
    Route.start
    |> Route.get "/ws/chat" (WS.hub chat "lobby" (fun hub _connId msg -> task {
        // Re-broadcast every inbound message to everyone in the room
        hub.Broadcast("lobby", msg)
    }))

Every client connected to /ws/chat joins the "lobby" room. When one sends a JSON frame (e.g. {"user":"ada","text":"hi"}), onMessage runs and hub.Broadcast fans it out to all members.

The hub is safe to share across connections — each socket is written only by its own pump, so concurrent broadcasts never interleave on a single connection.

Member	Description
`hub.Broadcast(room, msg)`	Send `msg` to every connection in `room`
`hub.BroadcastAll(msg)`	Send `msg` to every connection in every room
`hub.Send(connId, msg)`	Send `msg` to one connection
`hub.RoomCount(room)`	Number of connections in `room`
`hub.Count`	Total connected clients

The onMessage callback receives (hub, connId, msg); malformed inbound frames are ignored. Use the same hub instance across multiple routes to model multiple rooms.

Distributed broadcast (across nodes)

By default a hub is local — Broadcast only reaches sockets in the same process. To fan out across a cluster, give the hub a shared name and an IPubSub backplane. Every node that constructs WsHub<'T>(name, backplane) with the same name participates: a broadcast on one node reaches members on all of them.

// Each node builds the hub with the same name and a shared backplane.
let bus  = PubSub.inProcess ()          // process-local default (single node / tests)
let chat = WsHub<ChatMsg>("chat", bus)
 
// hub.Broadcast / hub.BroadcastAll now reach every node on the backplane.

PubSub.inProcess () is the built-in backplane — it fans out within a process (and lets tests simulate multiple nodes by sharing one instance), but does not cross process boundaries. For real multi-node deployments, supply an IPubSub implementation backed by a cross-process transport (Redis pub/sub, NATS, …):

type IPubSub =
    abstract Publish: topic: string * payload: byte[] -> unit
    abstract Subscribe: topic: string * handler: (byte[] -> unit) -> IDisposable

Firefly core never depends on Redis — the transport is an opt-in implementation you register. hub.Send(connId, msg) targets a connection on the local node only.

Presence

Presence tracks who is in each topic, with metadata, and notifies on join/leave. Like the hub, it is single-node by default and replicates across nodes when given a shared name + IPubSub backplane.

type UserMeta = { Name: string; Online: bool }
 
let presence = Presence("chat", bus)   // or Presence() for a single node
 
// when a user joins a room:
presence.Track("room-42", userId, { Name = "Ada"; Online = true })
 
// who's here (metadata deserialized to your type):
let here : (string * UserMeta) list = presence.List<UserMeta>("room-42")
 
// react to changes (broadcast "X joined", update a roster, …):
use _sub = presence.OnChange(fun d ->
    printfn "%s %s room %s" d.Key (if d.Joined then "joined" else "left") d.Topic)
 
// when they leave:
presence.Untrack("room-42", userId)

Member	Description
`presence.Track(topic, key, meta)`	Mark `key` present with metadata
`presence.Untrack(topic, key)`	Remove `key`
`presence.List<'M>(topic)`	Everyone present, metadata typed as `'M`
`presence.Count(topic)`	Number present in `topic`
`presence.OnChange(handler)`	Join/leave callback (`PresenceDiff`); dispose to stop

Presence shares the same backplane story: across nodes it merges each node's tracked entries. The current version has no heartbeat/expiry, so a crashed node's entries linger until untracked — single-node use is exact; multi-node is best-effort.

Server-Sent Events (SSE)

Handler-Driven SSE

Push events from within the handler function:

let countdown =
    Sse.handler (fun writer req -> task {
        for i in 10..-1..1 do
            do! writer.Event("countdown", string i)
            do! System.Threading.Tasks.Task.Delay(1000)
        do! writer.Data("Done!")
    })
 
Route.start
|> Route.get "/events/countdown" countdown

The SseWriter provides:

Method	Description
`writer.Event(event, data)`	Send a named event
`writer.Data(data)`	Send a data-only message

Channel-Driven SSE

Stream events from a ChannelReader:

open System.Threading.Channels
 
let channel = Channel.CreateUnbounded<SseEvent>()
 
// Producer (e.g., background service)
let sendEvent () = task {
    do! channel.Writer.WriteAsync({ Event = "update"; Data = """{"status":"ok"}""" })
}
 
Route.start
|> Route.get "/events/updates" (Sse.stream channel.Reader)

Each message is consumed by one client only.

Broadcast SSE

Multi-client broadcast where every connected client receives every event:

let hub = SseBroadcast()
 
// Broadcast to all connected clients
let notify (req: Request) = task {
    do! hub.Send({ Event = "notification"; Data = "Hello everyone!" })
    return Response.ok
}
 
Route.start
|> Route.get "/events/live" (Sse.broadcast hub)
|> Route.post "/notify" notify

The SseBroadcast manages per-client channels internally:

Member	Description
`hub.Send(event)`	Send an event to all connected clients
`hub.ClientCount`	Number of currently connected clients

let hub = SseBroadcast()
 
App.defaults
|> App.services [ Service.instance hub ]

Streaming JSON (NDJSON)

Stream a sequence of JSON objects as newline-delimited JSON:

// From a sequence
let streamItems (req: Request) = task {
    let items = seq {
        for i in 1..100 do
            yield {| id = i; name = $"Item {i}" |}
    }
    return Response.streamJson items
}
 
// From an async enumerable
let streamAsync (req: Request) = task {
    let items = getItemsAsyncEnumerable()
    return Response.streamJsonAsync items
}

The response uses application/x-ndjson content type with one JSON object per line, flushed after each item.

Custom Stream Callback

For full control over the response stream:

let customStream (req: Request) = task {
    return Response.streamCallback (fun ctx -> task {
        ctx.Response.ContentType <- "text/plain"
        for i in 1..10 do
            let bytes = System.Text.Encoding.UTF8.GetBytes($"Line {i}\n")
            do! ctx.Response.Body.WriteAsync(System.ReadOnlyMemory(bytes))
            do! ctx.Response.Body.FlushAsync()
            do! System.Threading.Tasks.Task.Delay(100)
        return ()
    })
}