Real-Time Subscriptions
Turn any query into a live stream with one suffix.
.listen() — Subscribe to Live Data
Append .listen() to any pipeline query. Liven returns the historical snapshot first, then streams newly matching records in real-time.
from("orders").listen()
This subscribes to all new records in the orders stream. The response begins with all existing records, followed by a live stream delivered via WebSocket.
Filtered Subscriptions
Only records matching the pipeline are streamed. This reduces bandwidth and client-side processing:
from("orders") | filter(amount > 100) .listen()
Subscription with Any Pipeline
You can chain any pipeline stages before .listen(). The server evaluates each incoming record against the full pipeline before delivering it:
from("embeddings") | vector_filter(value, [12, -45, 98], 0.85) .listen()
tail() — Low-Level Stream Tail
The tail("stream_name") command provides a raw, unfiltered stream of all new records in a stream. Useful for scripting and terminal use:
tail("orders")
Unlike .listen(), tail() does not return a historical snapshot — it begins streaming live records immediately.
How Subscriptions Work
- The server parses the pipeline and executes it against the full historical dataset.
- Historical results are sent to the client immediately.
- The server subscribes to the internal broadcast channel.
- Every new write is evaluated against the pipeline. Matching records are streamed to the client.
- If the client disconnects, the subscription is dropped — no cleanup needed.