tehuti 0.22.0

use crate::{
    channel::Channel,
    engine::{EngineId, EnginePeerDescriptor},
    event::Duplex,
    peer::{Peer, PeerBuildResult, PeerFactory, PeerId, PeerInfo, PeerRoleId},
};
use std::{
    collections::BTreeMap,
    error::Error,
    ops::Deref,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
};

#[derive(Debug)]
pub enum MeetingEngineEvent {
    /// Notification that the meeting has been destroyed.
    MeetingDestroyed,
    /// Notification that a peer has been created.
    PeerCreated(EnginePeerDescriptor),
    /// Notification that a peer has been destroyed.
    PeerDestroyed(PeerId),
    /// Notification that a peer has joined.
    PeerJoined(EngineId, PeerId, PeerRoleId),
    /// Notification that a peer has left.
    PeerLeft(PeerId),
}

#[derive(Debug)]
pub enum MeetingUserEvent {
    /// Request to create a peer with the given ID and role.
    PeerCreate(PeerId, PeerRoleId),
    /// Request to destroy the peer with the given ID.
    PeerDestroy(PeerId),
    /// Response that a peer has been added.
    PeerAdded(Peer),
    /// Response that a peer has been removed.
    PeerRemoved(PeerId),
}

/// User-side container for peers and channels.
/// The goal of a meeting is to replicate a multi-peer environment.
pub struct Meeting {
    factory: Arc<PeerFactory>,
    engine_event: Duplex<MeetingEngineEvent>,
    user_event: Duplex<MeetingUserEvent>,
    peers: BTreeMap<PeerId, Vec<Channel>>,
    name: String,
    peer_kill_event: Duplex<PeerId>,
}

impl Drop for Meeting {
    fn drop(&mut self) {
        let _ = self
            .engine_event
            .sender
            .send(MeetingEngineEvent::MeetingDestroyed);
        tracing::event!(
            target: "tehuti::meeting",
            tracing::Level::TRACE,
            "Meeting closed: {}",
            self.name
        );
    }
}

impl Meeting {
    pub fn new(
        factory: Arc<PeerFactory>,
        engine_event: Duplex<MeetingEngineEvent>,
        user_event: Duplex<MeetingUserEvent>,
        name: impl ToString,
    ) -> Self {
        let name = name.to_string();
        tracing::event!(
            target: "tehuti::meeting",
            tracing::Level::TRACE,
            "Meeting opened: {}",
            name
        );
        Self {
            factory,
            engine_event,
            user_event,
            peers: Default::default(),
            name,
            peer_kill_event: Duplex::unbounded(),
        }
    }

    pub fn name(&self) -> &str {
        &self.name
    }

    pub fn factory(&self) -> &Arc<PeerFactory> {
        &self.factory
    }

    pub fn peers(&self) -> impl Iterator<Item = PeerId> {
        self.peers.keys().copied()
    }

    pub fn pump(&mut self) -> Result<bool, Box<dyn Error>> {
        let mut result = false;
        if let Some(event) = self.peer_kill_event.receiver.try_recv() {
            self.handle_peer_kill(event)?;
            result = true;
        }
        if let Some(event) = self.engine_event.receiver.try_recv() {
            self.handle_engine_event(event)?;
            result = true;
        }
        if let Some(event) = self.user_event.receiver.try_recv() {
            self.handle_user_event(event)?;
            result = true;
        }
        for peer in self.peers.values_mut() {
            for channel in peer {
                let pumped = channel.pump()?;
                result = result || pumped;
            }
        }
        Ok(result)
    }

    pub fn pump_all(&mut self) -> Result<bool, Box<dyn Error>> {
        let mut result = false;
        if let Some(event) = self.peer_kill_event.receiver.try_recv() {
            self.handle_peer_kill(event)?;
            result = true;
        }
        while let Some(event) = self.engine_event.receiver.try_recv() {
            self.handle_engine_event(event)?;
            result = true;
        }
        while let Some(event) = self.user_event.receiver.try_recv() {
            self.handle_user_event(event)?;
            result = true;
        }
        for peer in self.peers.values_mut() {
            for channel in peer {
                let pumped = channel.pump_all()? > 0;
                result = result || pumped;
            }
        }
        Ok(result)
    }

    fn handle_peer_kill(&mut self, peer_id: PeerId) -> Result<(), Box<dyn Error>> {
        if self.peers.remove(&peer_id).is_some() {
            tracing::event!(
                target: "tehuti::meeting",
                tracing::Level::TRACE,
                "Meeting {} peer killed: {:?}",
                self.name,
                peer_id
            );
            self.engine_event
                .sender
                .send(MeetingEngineEvent::PeerDestroyed(peer_id))
                .map_err(|err| format!("Engine event sender error: {err}"))?;
            self.user_event
                .sender
                .send(MeetingUserEvent::PeerRemoved(peer_id))
                .map_err(|err| format!("User event sender error: {err}"))?;
        } else {
            tracing::event!(
                target: "tehuti::meeting",
                tracing::Level::WARN,
                "Meeting {} peer kill request for non-existent peer: {:?}",
                self.name,
                peer_id
            );
        }
        Ok(())
    }

    fn handle_engine_event(&mut self, event: MeetingEngineEvent) -> Result<(), Box<dyn Error>> {
        tracing::event!(
            target: "tehuti::meeting",
            tracing::Level::TRACE,
            "Meeting {} handle engine event: {:?}",
            self.name,
            event
        );
        match event {
            MeetingEngineEvent::PeerJoined(engine_id, peer_id, role_id) => {
                let PeerBuildResult {
                    mut peer,
                    channels,
                    mut descriptor,
                } = self.factory.create(Peer::new(
                    PeerInfo {
                        peer_id,
                        role_id,
                        remote: true,
                    },
                    self.peer_kill_event.sender.clone(),
                ))?;
                descriptor.origin_engine = Some(engine_id);
                if self.peers.contains_key(&peer.info().peer_id) {
                    unsafe { peer.set_destroy_on_drop(false) };
                    return Err(
                        format!("Joined peer {:?} already exists", peer.info().peer_id).into(),
                    );
                }
                self.peers.insert(peer.info().peer_id, channels);
                tracing::event!(
                    target: "tehuti::meeting",
                    tracing::Level::TRACE,
                    "Meeting {} peer joined: {:?}",
                    self.name,
                    peer.info()
                );
                self.engine_event
                    .sender
                    .send(MeetingEngineEvent::PeerCreated(descriptor))
                    .map_err(|err| format!("Engine event sender error: {err}"))?;
                self.user_event
                    .sender
                    .send(MeetingUserEvent::PeerAdded(peer))
                    .map_err(|err| format!("User event sender error: {err}"))?;
            }
            MeetingEngineEvent::PeerLeft(peer_id) => {
                if self.peers.remove(&peer_id).is_some() {
                    tracing::event!(
                        target: "tehuti::meeting",
                        tracing::Level::TRACE,
                        "Meeting {} peer left: {:?}",
                        self.name,
                        peer_id
                    );
                    self.engine_event
                        .sender
                        .send(MeetingEngineEvent::PeerDestroyed(peer_id))
                        .map_err(|err| format!("Engine event sender error: {err}"))?;
                    self.user_event
                        .sender
                        .send(MeetingUserEvent::PeerRemoved(peer_id))
                        .map_err(|err| format!("User event sender error: {err}"))?;
                } else {
                    tracing::event!(
                        target: "tehuti::meeting",
                        tracing::Level::WARN,
                        "Meeting {} peer left event for non-existent peer: {:?}",
                        self.name,
                        peer_id
                    );
                }
            }
            _ => {
                tracing::event!(
                    target: "tehuti::meeting",
                    tracing::Level::WARN,
                    "Meeting {} received unhandled engine event: {:?}",
                    self.name,
                    event
                );
            }
        }
        Ok(())
    }

    fn handle_user_event(&mut self, event: MeetingUserEvent) -> Result<(), Box<dyn Error>> {
        tracing::event!(
            target: "tehuti::meeting",
            tracing::Level::TRACE,
            "Meeting {} handle user event: {:?}",
            self.name,
            event
        );
        match event {
            MeetingUserEvent::PeerCreate(peer_id, role_id) => {
                let PeerBuildResult {
                    mut peer,
                    channels,
                    descriptor,
                } = self.factory.create(Peer::new(
                    PeerInfo {
                        peer_id,
                        role_id,
                        remote: false,
                    },
                    self.peer_kill_event.sender.clone(),
                ))?;
                if self.peers.contains_key(&peer.info().peer_id) {
                    unsafe { peer.set_destroy_on_drop(false) };
                    return Err(
                        format!("Created peer {:?} already exists", peer.info().peer_id).into(),
                    );
                }
                self.peers.insert(peer.info().peer_id, channels);
                tracing::event!(
                    target: "tehuti::meeting",
                    tracing::Level::TRACE,
                    "Meeting {} peer created: {:?}",
                    self.name,
                    peer.info()
                );
                self.engine_event
                    .sender
                    .send(MeetingEngineEvent::PeerCreated(descriptor))
                    .map_err(|err| format!("Engine event sender error: {err}"))?;
                self.user_event
                    .sender
                    .send(MeetingUserEvent::PeerAdded(peer))
                    .map_err(|err| format!("User event sender error: {err}"))?;
            }
            MeetingUserEvent::PeerDestroy(peer_id) => {
                if self.peers.remove(&peer_id).is_some() {
                    tracing::event!(
                        target: "tehuti::meeting",
                        tracing::Level::TRACE,
                        "Meeting {} peer destroyed: {:?}",
                        self.name,
                        peer_id
                    );
                    self.engine_event
                        .sender
                        .send(MeetingEngineEvent::PeerDestroyed(peer_id))
                        .map_err(|err| format!("Engine event sender error: {err}"))?;
                    self.user_event
                        .sender
                        .send(MeetingUserEvent::PeerRemoved(peer_id))
                        .map_err(|err| format!("User event sender error: {err}"))?;
                } else {
                    tracing::event!(
                        target: "tehuti::meeting",
                        tracing::Level::WARN,
                        "Meeting {} peer destroy request for non-existent peer: {:?}",
                        self.name,
                        peer_id
                    );
                }
            }
            _ => {
                tracing::event!(
                    target: "tehuti::meeting",
                    tracing::Level::WARN,
                    "Meeting {} received unhandled user event: {:?}",
                    self.name,
                    event
                );
            }
        }
        Ok(())
    }
}

impl Future for Meeting {
    type Output = Result<(), Box<dyn Error>>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match self.pump() {
            Ok(true) => Poll::Ready(Ok(())),
            Ok(false) => {
                cx.waker().wake_by_ref();
                Poll::Pending
            }
            Err(e) => Poll::Ready(Err(e)),
        }
    }
}

pub struct MeetingInterfaceResult {
    pub meeting: Meeting,
    pub interface: MeetingInterface,
    pub engine_event: Duplex<MeetingEngineEvent>,
}

#[derive(Clone)]
pub struct MeetingInterface {
    user_event: Duplex<MeetingUserEvent>,
}

impl MeetingInterface {
    pub fn make(factory: Arc<PeerFactory>, name: impl ToString) -> MeetingInterfaceResult {
        let (inside_engine, outside_engine) = Duplex::crossing_unbounded();
        let (inside_user, outside_user) = Duplex::crossing_unbounded();
        let meeting = Meeting::new(factory, inside_engine, inside_user, name);
        MeetingInterfaceResult {
            meeting,
            interface: Self {
                user_event: outside_user,
            },
            engine_event: outside_engine,
        }
    }
}

impl Deref for MeetingInterface {
    type Target = Duplex<MeetingUserEvent>;

    fn deref(&self) -> &Self::Target {
        &self.user_event
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_async() {
        fn is_send<T: Send>() {}

        is_send::<MeetingEngineEvent>();
        is_send::<MeetingUserEvent>();
        is_send::<Meeting>();
    }
}