Radio

Subscribe to a WebSocket channel

-- Open a subscription. The URL itself is the subscription identifier —
-- pass it to the read / send / unsubscribe functions later.
CALL radio_subscribe('wss://stream.example.com/events');

-- Inspect what's subscribed
SELECT * FROM radio_subscriptions();

The URL is what every other radio_* function takes as its subscription key. If you’ll reuse it across many statements, stash it in a SQL variable and bind via :feed:

SET VARIABLE feed = 'wss://stream.example.com/events';

CALL radio_subscribe(getvariable('feed'));

SELECT * FROM radio_subscription_received_messages(getvariable('feed'));

See radio_subscribe and radio_subscriptions for the full signatures, including the optional named parameters for buffer sizing and transmit retry tuning.

Subscribe to Redis Pub/Sub

Same function, different URL — Radio routes WebSocket and Redis Pub/Sub through the same surface:

CALL radio_subscribe('redis://cache.example.com:6379/0?channel=events');

-- Drain the same way as a WebSocket subscription, keyed by URL
SELECT receive_time, message
FROM radio_subscription_received_messages('redis://cache.example.com:6379/0?channel=events')
ORDER BY receive_time DESC;

Drain received messages

-- Recent messages across all subscriptions
SELECT subscription_url, channel, receive_time, message
FROM radio_received_messages()
ORDER BY receive_time DESC
LIMIT 100;

-- One feed only — pass the same URL you subscribed with
SELECT receive_time, channel, message
FROM radio_subscription_received_messages('wss://stream.example.com/events')
ORDER BY receive_time DESC;

The full row shape from radio_received_messages() / radio_subscription_received_messages(url) is (subscription_id UBIGINT, subscription_url VARCHAR, message_id UBIGINT, message_type, receive_time TIMESTAMP_MS, seen_count UBIGINT, channel VARCHAR, message BLOB). The payload is a BLOB — cast it to VARCHAR for text or use DuckDB’s JSON functions for structured payloads. These are normal DuckDB table functions — JOIN, filter, aggregate, push to Parquet, anything you’d do with a regular table. See radio_received_messages and radio_subscription_received_messages.

Wait for fresh messages before reading

For loops that need to act as messages arrive (rather than polling on a timer), radio_listen blocks the connection until something lands in the inbox or a timeout elapses:

-- Block up to 30 seconds for a message on any subscription.
-- Args: (return_on_first_message, max_wait).
CALL radio_listen(true, INTERVAL '30 seconds');

-- Then drain
SELECT * FROM radio_received_messages() ORDER BY receive_time DESC LIMIT 1;

Pair with radio_sleep when scripting between polls.

Send a message and track delivery

-- Args: (subscription_url, channel, payload, max_attempts, retry_delay).
CALL radio_transmit_message(
  'wss://stream.example.com/events',
  /* channel       */ 'main',
  /* payload       */ '{"type":"hello","at":"2026-04-28T12:00:00Z"}'::BLOB,
  /* max_attempts  */ 3,
  /* retry_delay   */ INTERVAL '500 milliseconds'
);

-- Inspect transmit queue + delivery state
SELECT message_id, message
FROM radio_subscription_transmit_messages('wss://stream.example.com/events')
ORDER BY message_id DESC;

Per-subscription delivery counters live on radio_subscriptions (transmit_successes, transmit_failures, transmit_last_success_time, transmit_last_failure_time); poll those to confirm delivery progress. See radio_transmit_message and radio_subscription_transmit_messages.

Garbage-collect the outbox

-- Drop all already-delivered transmit messages
CALL radio_subscription_transmit_messages_delete_finished('wss://stream.example.com/events');

-- Or drop one specific pending outgoing message — message_id is the
-- UBIGINT returned by radio_subscription_transmit_messages.message_id.
CALL radio_subscription_transmit_message_delete('wss://stream.example.com/events', :message_id::UBIGINT);

See radio_subscription_transmit_messages_delete_finished and radio_subscription_transmit_message_delete.

Test pipelines without a real upstream

Inject messages straight into a subscription’s inbox to exercise downstream SQL — no broker, no real WebSocket, no flake:

CALL radio_subscription_received_message_add(
  'wss://stream.example.com/events',
  'channel-name',
  '{"type":"test","value":42}'::BLOB
);

SELECT * FROM radio_subscription_received_messages('wss://stream.example.com/events');

See radio_subscription_received_message_add. Useful for unit-testing pipelines that consume radio_* tables.

Tear down

-- Close the subscription
CALL radio_unsubscribe('wss://stream.example.com/events');

See radio_unsubscribe.

Diagnostics

SELECT radio_version();   -- e.g. '0.4.x'

Use radio_version to confirm what’s loaded when matching against bug reports. radio_flush drains all currently-buffered messages so subsequent reads see only fresh data.