Data Distribution Service (DDS)

Built-in pub/sub system for cross-instance communication and state persistence

Yazi’s Data Distribution Service (DDS) provides a built-in publish-subscribe system that enables communication between multiple Yazi instances and plugins. It’s built on a client-server architecture with no additional server process required.

Overview

DDS enables:

• Cross-instance communication - Share data between multiple Yazi instances
• State persistence - Maintain state across all instances
• Plugin messaging - Plugins can publish and subscribe to events
• Lua integration - Full Lua API for pub/sub operations
• Automatic discovery - Instances find each other automatically

Architecture

DDS uses a hybrid client-server model:

┌─────────────┐      ┌─────────────┐      ┌─────────────┐
│   Yazi #1   │◄────►│  DDS Server │◄────►│   Yazi #2   │
│  (Client)   │      │ (in Yazi #1)│      │  (Client)   │
└─────────────┘      └─────────────┘      └─────────────┘
       ▲                                          ▲
       │                                          │
       └──────────────────────────────────────────┘
              Direct communication via Unix socket

• First instance - Starts DDS server automatically
• Additional instances - Connect as clients to existing server
• No separate daemon - Server embedded in first Yazi process
• Unix socket - Communication via filesystem socket

Message Types (Embers)

DDS messages are called “embers” and include:

System Events

• hi - Client announces capabilities
• hey - Server broadcasts peer list
• bye - Client disconnects gracefully

File Events

• cd - Directory change
• load - Folder loading complete
• hover - File hover/selection
• rename - File renamed
• move - Files moved
• duplicate - Files duplicated
• trash - Files moved to trash
• delete - Files permanently deleted
• download - Remote files downloaded

Special Events

• yank - Files yanked to clipboard (cut/copy)
• bulk - Bulk file changes
• tab - Tab switched
• mount - Filesystem mounted/unmounted
• custom - User-defined events

Client-Server Implementation

The DDS system is implemented in yazi-dds/:

Server

The server (yazi-dds/src/server.rs:17) manages connections:

pub(super) struct Server;

impl Server {
    pub(super) async fn make() -> Result<JoinHandle<()>> {
        CLIENTS.write().clear();
        let listener = Stream::bind().await?;

        Ok(tokio::spawn(async move {
            while let Ok((stream, _)) = listener.accept().await {
                let (tx, mut rx) = mpsc::unbounded_channel::<String>();
                let (reader, mut writer) = tokio::io::split(stream);
                // Handle connection...
            }
        }))
    }
}

Client

Clients (yazi-dds/src/client.rs:34) connect and communicate:

impl Client {
    pub(super) fn serve() {
        tokio::spawn(async move {
            let mut server = None;
            let (mut lines, mut writer) = Self::connect(&mut server).await;

            loop {
                select! {
                    Some(payload) = rx.recv() => {
                        writer.write_all(payload.as_bytes()).await;
                    }
                    Ok(next) = lines.next_line() => {
                        // Process received message
                    }
                }
            }
        });
    }
}

Publish-Subscribe API

The pub/sub system (yazi-dds/src/pubsub.rs) provides:

Subscribe (Local)

Subscribe to events in the current instance:

local function on_tab_switch(id)
    ya.err("Switched to tab: " .. id)
end

ps.sub("my-plugin", "tab", on_tab_switch)

Subscribe (Remote)

Subscribe to events from all instances:

local function on_remote_cd(tab, url)
    ya.err("Another instance changed to: " .. tostring(url))
end

ps.sub_remote("my-plugin", "cd", on_remote_cd)

Publish

Send events to subscribers:

-- Publish to local subscribers
ps.pub("custom", "my-event-data")

-- Publish to specific instance
ps.pub_to(receiver_id, "custom", "my-event-data")

Unsubscribe

ps.unsub("my-plugin", "tab")
ps.unsub_remote("my-plugin", "cd")

Built-in Event Publishing

Yazi automatically publishes events using the pub_after_* pattern:

impl Pubsub {
    pub fn pub_after_cd(tab: Id, url: &UrlBuf) -> Result<()> {
        if BOOT.local_events.contains("cd") {
            EmberCd::borrowed(tab, url).with_receiver(*ID).flush()?;
        }
        if PEERS.read().values().any(|p| p.able("cd")) {
            Client::push(EmberCd::borrowed(tab, url))?;
        }
        if LOCAL.read().contains_key("cd") {
            Self::pub(EmberCd::owned(tab, url))?;
        }
        Ok(())
    }
}

This ensures events are:

1. Delivered locally if subscribed
2. Broadcast to remote peers if they subscribe
3. Persisted if static event

State Persistence

Static events (prefixed with @) are persisted:

if receiver == 0 && kind.starts_with('@') {
    let Some(body) = parts.next() else { continue };
    if !STATE.set(kind, sender, body) { continue }
}

When new clients connect, they receive all persisted state:

if let Some(state) = &*STATE.read() {
    state.values().for_each(|s| _ = tx.send(s.clone()));
}

This enables:

• Shared clipboard - Yank state synced across instances
• Plugin state - Persistent data between restarts
• Configuration sync - Runtime config changes propagated

Peer Discovery

Instances track each other via hey messages:

fn handle_hey(clients: &HashMap<Id, Client>) {
    let payload = Payload::new(EmberHey::owned(
        clients.values().map(|c| (c.id, Peer::new(&c.abilities))).collect(),
    ));
    if let Ok(s) = try_format!("{payload}\n") {
        clients.values().for_each(|c| _ = c.tx.send(s.clone()));
    }
}

Each instance knows:

• Peer IDs - Unique identifier for each instance
• Capabilities - What events each peer can receive
• Connection status - When peers join/leave

Ya CLI Integration

The ya command-line tool uses DDS to communicate with running instances:

Send Single Message

ya pub static quit --json '{"args": []}'

Implemented in yazi-dds/src/client.rs:81:

pub async fn shot(kind: &str, receiver: Id, body: &str) -> Result<()> {
    Ember::validate(kind)?;

    let payload = try_format!(
        "{}\n{kind},{receiver},{ID},{body}\n{}\n",
        Payload::new(EmberHi::borrowed(iter::empty())),
        Payload::new(EmberBye::borrowed())
    )?;

    let (mut lines, mut writer) = Stream::connect().await?;
    writer.write_all(payload.as_bytes()).await?;
    writer.flush().await?;
    drop(writer);

    // Wait for response...
}

Listen to Events

ya sub cd,hover,load

Implemented in yazi-dds/src/client.rs:141:

pub async fn draw(kinds: HashSet<&str>) -> Result<()> {
    async fn make(kinds: &HashSet<&str>) -> Result<ClientReader> {
        let (lines, mut writer) = Stream::connect().await?;
        let hi = Payload::new(EmberHi::borrowed(kinds.iter().copied()));
        writer.write_all(try_format!("{hi}\n")?.as_bytes()).await?;
        writer.flush().await?;
        Ok(lines)
    }

    let mut lines = make(&kinds).await
        .context("No running Yazi instance found")?;
    loop {
        match lines.next_line().await? {
            Some(s) => {
                let kind = s.split(',').next();
                if matches!(kind, Some(kind) if kinds.contains(kind)) {
                    println!("{s}");
                }
            }
            None => /* reconnect */,
        }
    }
}

Connection Management

Automatic Reconnection

Clients automatically reconnect if connection lost:

select! {
    Some(payload) = rx.recv() => {
        if writer.write_all(payload.as_bytes()).await.is_err() {
            (lines, writer) = Self::reconnect(&mut server).await;
            writer.write_all(payload.as_bytes()).await.ok(); // Retry once
        }
    }
    Ok(next) = lines.next_line() => {
        let Some(line) = next else {
            (lines, writer) = Self::reconnect(&mut server).await;
            continue;
        };
        // Process line...
    }
}

Heartbeat

Server sends periodic heartbeats to detect dead connections:

_ = time::sleep(Duration::from_secs(5)) => {
    if writer.write_u8(b'\n').await.is_err() {
        break;  // Connection dead
    }
}

Graceful Shutdown

Clients send bye message before disconnecting:

async fn handle_bye(id: Id, mut rx: UnboundedReceiver<String>, mut writer: ClientWriter) {
    // Flush pending messages
    while let Ok(payload) = rx.try_recv() {
        writer.write_all(payload.as_bytes()).await.ok();
    }

    // Send bye
    let bye = EmberBye::borrowed().with_receiver(id).with_sender(Id(0));
    if let Ok(s) = try_format!("{bye}") {
        writer.write_all(s.as_bytes()).await.ok();
        writer.flush().await.ok();
    }
}

Environment Variables

DDS sets environment variables for child processes:

pub fn init() {
    // ...
    unsafe {
        if let Some(s) = std::env::var("YAZI_ID").ok().filter(|s| !s.is_empty()) {
            std::env::set_var("YAZI_PID", s);  // Parent ID
        }
        std::env::set_var("YAZI_ID", ID.to_string());  // Current ID
        std::env::set_var(
            "YAZI_LEVEL",
            (std::env::var("YAZI_LEVEL").unwrap_or_default().parse().unwrap_or(0u16) + 1).to_string(),
        );
    }
}

This enables:

• Nested detection - Know if running inside another Yazi
• Parent tracking - Find parent Yazi instance
• Nesting level - Track recursion depth

Version Compatibility

DDS checks version compatibility:

if version.as_deref() != Some(EmberHi::version()) {
    bail!(
        "Incompatible version (Ya {}, Yazi {}). Restart all `ya` and `yazi` processes if you upgrade either one.",
        EmberHi::version(),
        version.as_deref().unwrap_or("Unknown")
    );
}

This prevents communication errors between different Yazi versions.

Plugin Examples

Cross-Instance Notification

-- In plugin init
ps.sub_remote("my-plugin", "custom", function(data)
    ya.notify({
        title = "Remote Event",
        content = "Another instance sent: " .. data,
    })
end)

-- In another instance
ps.pub("custom", "Hello from instance A!")

Synchronized State

local state = {}

ps.sub_remote("my-plugin", "custom", function(data)
    state = data
    -- Update UI
end)

function update_state(new_state)
    state = new_state
    ps.pub("custom", state)
end

File Synchronization

ps.sub_remote("sync-plugin", "cd", function(tab, url)
    -- When any instance changes directory, sync others
    if should_sync(url) then
        ya.manager_emit("cd", { url = tostring(url) })
    }
end)

Performance

DDS is designed for efficiency:

• Async I/O - Non-blocking communication
• Unix sockets - Fast local IPC
• Binary format - Compact message encoding
• Connection pooling - Reuse connections
• Selective delivery - Only send to interested peers

Debugging

Monitor DDS activity:

# Watch all events
ya sub cd,hover,load,tab,yank,bulk,move,trash,delete

# Send test message
ya pub custom --str "test message"

# Check connected instances
ya pub hey  # Triggers peer list update

Security Considerations

• Local only - Unix socket accessible only to same user
• No authentication - Processes under same UID trusted
• No encryption - Data not encrypted (local socket)

See Also

• Async Tasks - Task coordination across instances
• Plugin System - Using DDS in plugins
• Ya CLI - Command-line DDS interaction