In-memory LiveMap

LiveMap is an in-memory keyed collection that bridges live producing and consuming logic — declare (key, value) entries as target states on one side, consume it as a LiveMapView with ctx.mount_each_live on the other.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

LiveMap<K, V> (cocoindex::resources::live_map, also cocoindex::LiveMap) is an in-memory, keyed collection that sits between a producing part of your pipeline and a consuming part. The producing side declares (key, value) entries; the consuming side reads the map as a LiveMapView via ctx.mount_each_live, getting one processing component per entry that stays in sync as entries appear, change, and disappear.

Think of it as a connector whose “external system” is an in-process map: entries are declared as target states (so they participate in CocoIndex’s declarative change detection and ownership) and that same map is simultaneously exposed as a live source for downstream components. It lets you split a pipeline into a producer half and a consumer half that are decoupled through one shared, incrementally-maintained collection.

It is designed for live mode: the producer and consumer run concurrently, and the consumer reacts as the producer updates the map.

Example

use std::sync::{Arc, Mutex};

use cocoindex::{Ctx, LiveComponent, LiveComponentOperator, LiveMap, Result, async_trait};
use tokio::sync::Notify;

/// Declares every entry of `source`; re-runs its full pass on `trigger`.
struct Producer {
    lm: LiveMap<String, String>,
    source: Arc<Mutex<Vec<(String, String)>>>,
    trigger: Arc<Notify>,
}

#[async_trait]
impl LiveComponent for Producer {
    async fn process(&self, ctx: Ctx) -> Result<()> {
        let entries = self.source.lock().unwrap().clone();
        for (key, value) in entries {
            self.lm.declare_entry(&ctx, key, value)?;
        }
        Ok(())
    }

    async fn process_live(&self, operator: LiveComponentOperator) -> Result<()> {
        operator.update_full().await?;
        operator.mark_ready().await;
        loop {
            self.trigger.notified().await;
            operator.update_full().await?;
        }
    }
}

/// Consumes the map: one child component per entry.
struct Consumer {
    lm: LiveMap<String, String>,
    seen: Arc<Mutex<Vec<String>>>,
}

#[async_trait]
impl LiveComponent for Consumer {
    async fn process(&self, ctx: Ctx) -> Result<()> {
        let seen = self.seen.clone();
        ctx.mount_each_live(&"each", self.lm.clone(), move |_ctx, value: String| {
            let seen = seen.clone();
            async move {
                seen.lock().unwrap().push(value);
                Ok(())
            }
        })
        .await
    }

    async fn process_live(&self, operator: LiveComponentOperator) -> Result<()> {
        operator.update_full().await?;
        operator.mark_ready().await;
        std::future::pending::<()>().await;
        Ok(())
    }
}

// In the app body:
let lm: LiveMap<String, String> = LiveMap::create(&ctx).await?;
ctx.mount_live(&"producer", Producer { lm: lm.clone(), source, trigger }).await?;
ctx.mount_live(&"consumer", Consumer { lm, seen }).await?;

The producer and the consumer are live components here, so both keep running in live mode. A plain component can declare entries too: declare_entry only needs a Ctx.

Creating a LiveMap

let lm: LiveMap<String, String> = LiveMap::create(&ctx).await?;

create is an async factory and must be called from inside the app’s component tree (it mounts a backing target).

K is any LiveMapKey: String, i64, u64 or a StableKey, so a composite key works too. V must be Serialize + DeserializeOwned + Clone + PartialEq + Send + Sync + 'static (the LiveMapValue bound, implemented automatically for every such type). Values are compared with ==; the serialization bound is what makes an entry a target state. A LiveMap is cheap to clone, and all clones share one map.

Producing entries

lm.declare_entry(&ctx, "a".to_string(), "alpha".to_string())?;

Call declare_entry from inside any component. The entry is a target state owned by the declaring component, which gives it normal declarative semantics:

• Declaring a key makes it present, or updates its value.
• The consumer is notified only when the value actually changes (compared with ==) — re-declaring the same value is a no-op for downstream.
• An entry is removed when the component that declared it stops declaring it (on a re-run) or disappears. There is no explicit delete verb; deletion follows target-state ownership, exactly like other CocoIndex targets.

Multiple components may declare into the same map, as long as each key has a single owner (two components declaring the same key is a conflict, the same as for any target).

Consuming entries

A LiveMap is a LiveMapView, so pass it straight to ctx.mount_each_live — it behaves just like a live source:

ctx.mount_each_live(&"each", self.lm.clone(), move |_ctx, value: String| {
    let seen = seen.clone();
    async move {
        seen.lock().unwrap().push(value);
        Ok(())
    }
})
.await

ctx.mount_each_live(&key, feed, process_fn) takes the map as its feed: LiveMap<K, V> implements LiveMapView<K, V> and LiveMapFeed<K, V>, so it is a live feed like a watched directory. lm.scan().await? (from the LiveMapView trait) returns the current entries as Vec<(K, V)>.

mount_each_live mounts one processing component per entry (keyed by the entry key), scans the current entries first, then reacts to incremental changes — re-mounting a component when its value changes and removing it when its entry is deleted. The processor receives the entry value.

A LiveMap supports one active consumer at a time; a second concurrent watch fails with an Error. A consumer that finished, failed or was cancelled releases the map, so it can be watched again.

Semantics

• Live-mode first. LiveMap is built for live mode: the consumer subscribes and reacts as producers update the map. In catch-up mode (a plain update), the consumer scans whatever entries happen to exist when it runs, which can race producers running concurrently in the same session — so for predictable one-shot results, order the producer ahead of the consumer (wait for the producer’s mount handle to be ready before consuming).
• In-memory and per-session. The map lives in process memory and is rebuilt from its producers each time the app starts; its contents do not persist across restarts.
• Empty in a preview. A preview (app.preview(..), cocoindex update --preview) applies no target actions, so it does not fill the map either: its consumers see an empty map, and the preview lists no downstream actions for them. Preview the producers’ own targets instead.

When to use it

Reach for a LiveMap when you want to decouple a producing stage from a consuming stage through a shared, incrementally-maintained collection. For example: one part of your pipeline watches a stream, extracts entities, and declares them into a map keyed by entity ID; another part builds an index or enrichment for each entity, reacting automatically as entities are added, updated, or removed.