clan-sidebus/sidebus-broker/src/bus.rs

use std::sync::Arc;
use tracing::trace;

pub struct HostedBus {
    peers: Arc<busd::peers::Peers>,
    guid: zbus::OwnedGuid,
    next_id: usize,
    _self_conn: zbus::Connection,
}

impl HostedBus {
    pub async fn new() -> eyre::Result<Self> {
        let guid: zbus::OwnedGuid = zbus::Guid::generate().into();

        let peers = busd::peers::Peers::new();

        let dbus = busd::fdo::DBus::new(peers.clone(), guid.clone());
        let monitoring = busd::fdo::Monitoring::new(peers.clone());

        // Create a peer for ourselves.
        trace!("Creating self-dial connection.");
        let (client_socket, peer_socket) = zbus::connection::socket::Channel::pair();
        let service_conn =
            zbus::connection::Builder::authenticated_socket(client_socket, guid.clone())?
                .p2p()
                .unique_name(busd::fdo::BUS_NAME)?
                .name(busd::fdo::BUS_NAME)?
                .serve_at(busd::fdo::DBus::PATH, dbus)?
                .serve_at(busd::fdo::Monitoring::PATH, monitoring)?
                .build()
                .await?;
        let peer_conn = zbus::connection::Builder::authenticated_socket(peer_socket, guid.clone())?
            .p2p()
            .build()
            .await?;

        peers.add_us(peer_conn).await;
        trace!("Self-dial connection created.");
        Ok(HostedBus {
            peers,
            guid,
            next_id: 0,
            _self_conn: service_conn,
        })
    }

    pub fn server_guid(&self) -> &zbus::Guid<'_> {
        self.guid.inner()
    }

    pub async fn connect_channel(&mut self, skip_hello: bool) -> eyre::Result<zbus::Connection> {
        let id = self.next_id();
        self.peers
            .add_channel(&self.guid, id, skip_hello)
            .await
            .map_err(|err| eyre::eyre!(Box::new(err))) // https://github.com/eyre-rs/eyre/issues/31 XXX: busd should not use anyhow!
    }

    pub async fn connect_unix(&mut self, socket: tokio::net::UnixStream) -> eyre::Result<()> {
        let id = self.next_id();
        self.peers
            .add(&self.guid, id, socket.into(), zbus::AuthMechanism::External)
            .await
            .map_err(|err| eyre::eyre!(Box::new(err)))
    }

    fn next_id(&mut self) -> usize {
        self.next_id += 1;
        self.next_id
    }
}