Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Streaming RTD

Enable the rtd feature in the application’s xlfn dependency. The handles feature alone does not expose the generic streaming API.

Real-Time Data (RTD) is the appropriate model for a formula that should update repeatedly from a push source. xlfn hides the COM server transport and exposes typed sources, topics, sinks, and subscriptions.

Data flow

worksheet formula
    -> MainThreadContext::rtd().subscribe(&source_handle, &parts)
    -> RtdChannelSource starts a producer and publisher
    -> producer calls sender.try_send(value)
    -> framework publisher forwards queued values through RtdSink
    -> framework batches RefreshData
    -> Excel recalculates the dependent formula

The initial subscription returns the current value, which can be empty while the asynchronous producer starts. Later publications update the RTD topic and notify Excel.

Define a topic

#![allow(unused)]
fn main() {
let parts = [
    "events",
    topic.as_str(),
    metric.as_str(),
];
context.rtd().subscribe(&source_handle, &parts)?;
}

For one part:

#![allow(unused)]
fn main() {
context.rtd().subscribe(&source_handle, &["service-health"])?;
}

The subscription call borrows its parts and copies them only when creating a new subscription. Source callbacks receive the canonical owned RtdTopic.

Use topic.len() to inspect the part count, topic.part(index) for an optional &str, and topic.parts() for an allocation-free RtdTopicParts iterator over &str values. The iterator implements ExactSizeIterator. RtdTopic::new and RtdTopic::single accept string inputs through AsRef<str>; their storage representation is private.

A topic must contain at least one non-empty part. Each part must fit Excel’s 32,767 UTF-16-unit counted-string representation. Topic parts are identity, not display labels; use stable, canonical values.

The runtime also applies bounded admission limits. The standard limits are 253 topic parts, 1 MiB of UTF-8 text per topic, 64 MiB of pending-topic text in aggregate, 4,096 pending preparations, 4,096 active streams, 4,096 queued updates, and 4,096 distinct live source identities. A custom RuntimeConfig::with_rtd_limits can choose lower limits during Addin::open; use RtdCapacity::bounded or RtdCapacity::disabled for each resource class so a disabled limit is explicit rather than an untyped zero. Exceeding a limit returns XllError::Overloaded (or a topic input error for an invalid topic).

Implement a source

Use RtdChannelSource for ordinary producers. The source owns a factory that runs synchronously during subscription setup and returns an owned producer job for the topic. Each job runs once on a framework-owned thread and receives only a typed, bounded RtdSender. Its captures may include resources that are neither Clone nor Sync. A separate publisher thread owns the internal sink. The capacity passed to RtdChannelSource::new limits queued values per subscription.

#![allow(unused)]
fn main() {
use std::{num::NonZeroUsize, time::Duration};

use xlfn::{error::InputError, prelude::*, rtd::RtdValue};

use super::Client;

pub(crate) type MetricSource = RtdChannelSource<RtdValue>;

pub(crate) fn metric_source() -> MetricSource {
    RtdChannelSource::new(NonZeroUsize::new(64).unwrap(), |topic| {
        let mut parts = topic.parts();
        let (Some(kind), Some(symbol), None) = (parts.next(), parts.next(), parts.next()) else {
            return Err(XllError::input(
                "RTD topic",
                InputError::Malformed("expected [kind, symbol]"),
            ));
        };
        if kind != "last" {
            return Err(XllError::input(
                "RTD topic",
                InputError::Malformed("unsupported metric topic"),
            ));
        }
        let symbol = symbol.to_owned();
        // Each job owns its client; real integrations can open a connection here.
        let client = Client;
        Ok(move |sender: RtdSender<RtdValue>| {
            while !sender.is_closed() {
                match client.try_next_metric(&symbol) {
                    Ok(Some(value)) => sender.try_send(RtdValue::Number(value))?,
                    Ok(None) => {
                        sender.wait_closed(Duration::from_millis(50));
                    }
                    Err(_) => {
                        sender
                            .try_send(RtdValue::Error(ExcelErrorValue(ExcelError::NotAvailable)))?;
                        break;
                    }
                }
            }
            Ok(())
        })
    })
}
}

In this illustrative loop, try_next_metric is a non-blocking poll and wait_closed makes the polling delay interruptible. Use bounded or cancellation-aware I/O so the callback returns promptly after admission closes. The factory validates the topic before any worker starts. Producer errors and panics close the channel and are reported during disconnect.

Shutdown and advanced sources

The channel adapter closes admission, discards pending values, and joins both workers during disconnect. Sender clones can remain alive afterward: their try_send calls return XllError::Closing, even after the RTD runtime has been reclaimed. Reference counting shares only the empty, closed queue storage; the subscription owns admission, payload disposal, and worker joins. Source configuration and each producer job have separate unique owners, so dropping the source neither drops a running job’s resources nor waits for a final shared callback reference. Jobs release their captures before their workers are joined. Successful producer completion drains accepted values before the publisher stops. The producer must stop any additional threads or callbacks before it returns. A producer that ignores cancellation delays disconnect and unload.

Each active channel subscription uses two threads. Advanced integrations that share an event loop can implement the existing unsafe RtdSource and RtdSubscription traits. In that path, a sink must not escape an Err or panic from subscribe; the returned subscription must stop every sink user before disconnect_and_wait exits, whether it succeeds, returns an error, or unwinds. Preserve this guarantee during unwinding, for example with a cleanup guard that joins all sink users. The framework contains cleanup panics and may reclaim the runtime afterward; an error or panic does not extend sink lifetime. Cancellation must be bounded, idempotent, panic-free, and must not call Excel or re-enter framework subscription APIs. Sinks are non-owning capabilities, so this shutdown contract is a memory-safety requirement for the unsafe extension point.

Do not implement a timeout that abandons an in-process callback and then permits the XLL to unload. Put uninterruptible producers in another process.

Use the source from a function

Keep the source handle in add-in state and subscribe from a main-thread function:

#![allow(unused)]
#![deny(unsafe_op_in_unsafe_fn)]

fn main() {
use xlfn::prelude::*;
use xlfn::rtd::RtdValue;

mod metric_source;

use metric_source::{MetricSource, metric_source};

#[derive(Default)]
pub(crate) struct Client;

impl Client {
    fn try_next_metric(&self, _symbol: &str) -> Result<Option<f64>, std::io::Error> {
        Ok(None)
    }
}

pub struct State {
    metrics: RtdSourceHandle<MetricSource>,
}

#[excel_addin(name = "RTD Source Example", id = "rtd-source", category = "Examples")]
pub struct RtdSourceExample;

impl Addin for RtdSourceExample {
    type SharedState = State;
    type LifecycleState = ();
    type Error = XllError;
    type Layers = ();

    fn open(
        context: &OpenContext,
    ) -> Result<Opened<Self::SharedState, Self::LifecycleState, Self::Layers>, Self::Error> {
        Ok(Opened::new(
            State {
                metrics: context.rtd().register_source(metric_source())?,
            },
            (),
            (),
        ))
    }
}

#[excel_function(name = "METRIC.LAST")]
pub fn last_metric(
    #[excel_context(main_thread)] context: MainThreadContext<'_, RtdSourceExample>,
    symbol: String,
) -> XllResult<RtdValue> {
    context
        .rtd()
        .subscribe(&context.state().metrics, &["last", symbol.as_str()])
}
}

The source handle is the opaque RTD identity. Clone a handle when multiple functions should refer to the same source; registering a new source creates a distinct identity even when its value is equivalent. The runtime owns the source and subscription identity; handles do not extend its lifetime. Multiple formulas that observe the same active subscription share it; a failed new observation rolls back only the reservation created by that attempt, not an unrelated established subscriber. The complete compile-tested fixture, including the add-in state and Client placeholder, lives under examples/rtd-source.

RTD value types

RtdValue supports scalar transport:

  • finite number;
  • Boolean;
  • integer;
  • string;
  • Excel error;
  • empty.

IntoRtdValue is implemented for common scalar types, including f64, bool, i32, exactly representable i64, ExcelSerialDate, strings, ExcelErrorValue, and ().

RTD does not transport arrays. Publish a handle or another scalar identity and expose a separate function when a stream logically updates a complex object.

Backpressure and errors

RtdSender::try_send validates values before enqueueing and never waits for queue capacity. It returns XllError::Overloaded when the per-subscription queue is full and XllError::Closing after admission closes. Handle an error by stopping or retrying with a bounded policy. Enqueue success is not a delivery acknowledgement: disconnect can discard pending values. If the publisher encounters a runtime error, it closes admission, wakes the producer, and reports that error during disconnect. Closing is treated as normal shutdown.

For an unsafe custom source, RtdSink::publish directly reports a closing runtime, an inactive subscription, invalid values, or exhaustion of the runtime’s queued-update limit. Do not loop tightly on a permanent error.

Publishing validates and queues a value; notification and RefreshData happen through the framework. User code must never call the COM update event directly.

Temporary COM registration

The XLL registers its RTD COM server temporarily. Ownership markers include the add-in identity, schema, module path, and CLSID. On a later start after an abnormal Excel exit, xlfn removes only stale registrations whose full marker set belongs to the same XLL. It does not broadly delete similarly named registry keys.

Installation still needs appropriate user-profile registry access. Test start, normal close, forced Excel termination, and restart in the deployment environment.

Choosing RTD versus async

Use async for one eventual result. Use RTD for a value that may change repeatedly while a formula remains subscribed. Do not simulate streaming by starting an infinite async function; it prevents normal completion and complicates cancellation and unload.