Skip to content

WebSocketFeed

WebSocketFeed is one WebSocket connection as a WorkManager worker. It replaces the WebSocketManager service connection by connection (workmanager#103): the Runtime gives it a lifecycle (StartAsync, StopAsync), an activation lease and an outbox, so the feed is an ordinary worker in ListWorkers, with tags, lifecycle events, dead letters and worker secrets. Until the cutover (phase 4) it runs next to the WebSocketManager.

The worker is Virtufin.Worker.WebSocketFeed (worker type websocketfeed, engine application/x-dotnet-dll); the connection code it is built on is the library Virtufin.WebSocket.DevKit, both in this repo (WebSocketFeed, WebSocketDevKit).

How a frame travels

flowchart LR
    venue[(Venue<br/>wss://...)]
    subgraph runtime[WorkManager Runtime pod]
        subgraph feed[WebSocketFeed worker]
            session[WebSocketSession<br/>login, subscribe,<br/>receive, reconnect]
            router[PublishRouter<br/>key_path, keys,<br/>split_arrays]
            queue[[frame queue<br/>one batch per burst]]
        end
        gate{lease<br/>gate}
        outbox[[outbox]]
    end
    api[API gateway<br/>Pubsub]
    consumers[Consumers<br/>BinanceOrderBook,<br/>AlpacaOrderBook, ...]
    tui[TUI, notebooks,<br/>scripts]
    wm[WorkManager]

    venue -- frames --> session
    session -- one JSON value --> router --> queue
    queue -- ctx.PublishAsync --> gate --> outbox --> api
    api -- publishtopic or<br/>routed sub-topic --> consumers
    tui -- InvokeWorkerCommand<br/>send, publish, status --> wm --> feed
    consumers -. attach / detach<br/>wrk.websocketfeed.command.id .-> feed
    session -- send, subscribe --> venue
    feed -. ReportStatusAsync<br/>connected, reconnecting, ... .-> wm
  1. Start. The Runtime claims the worker's lease and calls StartAsync. The feed parses its config, checks the URL against the SSRF guard, reads its saved session (publish topic, routing, replayed frames, attachments) and starts the session in the background, reporting connecting.
  2. Open. The session connects and runs the open handler before anything is relayed: the login (handshake), the Alpaca subscription of the union of symbols, the subscribe frames. Then it reports connected.
  3. Frames. The receive loop hands every whole message to the feed. A frame that is one JSON value is routed (routing) and queued; one publisher task hands everything queued to ctx.PublishAsync in one batch, into the Runtime's outbox, which publishes only while the lease is current.
  4. Drop. A dropped socket reports reconnecting and is reopened with backoff (1 s doubling to 30 s), running the login and the subscriptions again. It keeps trying; reconnectmaxattempts caps it.
  5. Stop. Ownership ends (stop, delete, reload, a lost lease, a move): the stop token fires and the feed publishes nothing more; StopAsync sends a normal close and aborts within its deadline. The next activation, on this pod or another, finds the saved session and opens as a handoff: a rejected login (Alpaca's 406 while the old socket still holds the account's single slot) is retried with backoff.

Config, commands and events

The worker's README lists every config key, command and attribute. In short:

Config Meaning
url ws:// / wss://, required
publishtopic, routing where frames go; routing is {key_path, split_arrays, keys}
handshake none, script (handshakescript steps {text\|base64, expect?, fail?}), alpaca (worker secrets alpacakey, alpacasecret)
subscribe frames sent after the login on every open
symbols, currencypairs, channels, orderbooklimit the Alpaca subscription; beyond 10 order books the feed reports degraded
autoreconnect, reconnectmaxattempts reconnect policy (default: forever)
pinned, releaseafterminutes an unpinned feed without consumers closes and reports released
resource=websocketfeed, autoreload=false for requires=websocketfeed/<name> consumers; no bounce on library releases
Command Effect
send one frame; replay: true also on every later open
publish, unpublish the publish topic and routing (saved)
attach, detach a consumer and its symbols (saved; Alpaca resubscribes the delta)
status state, URL, topic, routing, attachments, counters

Frames are published as com.virtufin.websocketfeed.message from urn:com.virtufin:worker:websocketfeed with the WebSocketManager's attributes (ingesttime, websocketid = the worker id, websocketurl). Status goes out as the worker's lifecycle status event.

Mapping from the WebSocketManager

WebSocketManager WebSocketFeed
Connect CreateWorker (config, secrets, tags) + StartWorker; wait for status connected
Disconnect DeleteWorker
SendRaw / Send InvokeWorkerCommand send (no correlated request/reply)
StartPublish / StopPublish InvokeWorkerCommand publish / unpublish
List, tags ListWorkers, the WorkManager tag RPCs
connection.lifecycle the worker's lifecycle events (status)
handshake request/reply on pub/sub the login computed in the worker from its secrets

Using it from a notebook

examples/dotnet/platform.cs:

var feedId = await platform.OpenFeedWorkerAsync("binance-feed", "wss://stream.binance.com:9443/stream",
    publishTopic: unroutedTopic,
    routing: new JsonObject { ["key_path"] = "stream", ["keys"] = keys },
    subscribe: [new JsonObject { ["method"] = "SUBSCRIBE", ["params"] = streams, ["id"] = 1 }]);
await platform.SendAsync(feedId, """{"method":"LIST_SUBSCRIPTIONS","id":2}""");

OpenFeedWorkerAsync creates the worker (autoreload=false, resource=websocketfeed, tag name), starts it, waits until it reports connected and removes it again in CleanupAsync, before the workers that consume it.

The DevKit

Virtufin.WebSocket.DevKit holds what a worker needs to own a socket, with no platform code in it:

Type What it does
WebSocketSession connect, open handler, receive loop, reconnect with capped backoff, handoff login retry, graceful close, status callback
HandshakeStep, HandshakeRunner the login steps (send, expect, fail, timeouts)
PublishRouting, PublishRouter content-based routing to sub-topics
JsonFrames the one-JSON-value publish rule
WebSocketUrlValidator the SSRF guard (operator: WEBSOCKETFEED_ALLOWED_HOSTS, WEBSOCKETFEED_ALLOW_PRIVATE_HOSTS)
WebSocketFeedNames the virtufin.worker.websocketfeed naming fragment it embeds

Not yet (phase 4 and later)

  • Consumers still read the venue topics as before; nothing has moved off the WebSocketManager yet (TUI, scripts, the app, the live notebooks).
  • A non-Alpaca feed records attachments (for status and the release) but derives no subscription from them: its subscriptions are the subscribe frames and send replay.
  • Commands sent on the command topic while the feed moves to another Runtime pod are lost: the new activation subscribes at the stream's tail (InvokeWorkerCommand fails instead, and can be retried).