jarust_interface 1.10.0

//! A [`JanusInterface`] that speaks the full Janus protocol over any [`Transport`].
//!
//! This is the generic counterpart to the native WebSocket and RESTful interfaces: it
//! owns transaction generation, request decoration, inbound demultiplexing, per-handle
//! routing and response polling, and delegates only the raw byte movement to a
//! [`Transport`]. Bring your own transport (including one that lives across an FFI
//! boundary) and hand it to [`CustomInterface::new`].

use crate::handle_msg::HandleMessage;
use crate::handle_msg::HandleMessageWithJsep;
use crate::janus_interface::ConnectionParams;
use crate::janus_interface::JanusInterface;
use crate::japrotocol::JaResponse;
use crate::japrotocol::ResponseType;
use crate::japrotocol::ServerInfoRsp;
use crate::tgenerator::GenerateTransaction;
use crate::tgenerator::TransactionGenerator;
use crate::transport::demuxer::Demuxer;
use crate::transport::interface_support;
use crate::transport::napmap::NapMap;
use crate::transport::router::Router;
use crate::transport::tmanager::TransactionManager;
use crate::transport_trait::Transport;
use crate::Error;
use jarust_rt::JaTask;
use serde_json::json;
use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::sync::Mutex;

#[derive(Debug)]
struct Shared {
    tasks: Vec<JaTask>,
    server_root: String,
    apisecret: Option<String>,
    transaction_generator: TransactionGenerator,
    transport: Arc<dyn Transport>,
    ack_map: Arc<NapMap<String, JaResponse>>,
    rsp_map: Arc<NapMap<String, JaResponse>>,
}

#[derive(Debug)]
struct Exclusive {
    router: Router,
    transaction_manager: TransactionManager,
}

#[derive(Debug)]
struct InnerCustomInterface {
    shared: Shared,
    exclusive: Mutex<Exclusive>,
}

/// A [`JanusInterface`] layered over a user-provided [`Transport`].
#[derive(Debug, Clone)]
pub struct CustomInterface {
    inner: Arc<InnerCustomInterface>,
}

impl CustomInterface {
    /// Builds a [`CustomInterface`] driving `transport`, connecting it to
    /// [`ConnectionParams::url`] and wiring up the demultiplexing machinery.
    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    pub async fn new(
        transport: impl Transport,
        conn_params: ConnectionParams,
        transaction_generator: impl GenerateTransaction,
    ) -> Result<Self, Error> {
        tracing::debug!("Creating custom interface");
        let transport: Arc<dyn Transport> = Arc::new(transport);
        let router = Router::new(&conn_params.server_root);
        let receiver = transport.connect(&conn_params.url).await?;
        let transaction_manager = TransactionManager::new(conn_params.capacity);
        let transaction_generator = TransactionGenerator::new(transaction_generator);

        let ack_map = Arc::new(NapMap::<String, JaResponse>::new(conn_params.capacity));
        let rsp_map = Arc::new(NapMap::<String, JaResponse>::new(conn_params.capacity));

        let (rsp_sender, mut rsp_receiver) = mpsc::unbounded_channel::<JaResponse>();
        let (ack_sender, mut ack_receiver) = mpsc::unbounded_channel::<JaResponse>();

        let rsp_task = jarust_rt::spawn("Responses gathering task", {
            let rsp_map = rsp_map.clone();
            async move {
                while let Some(rsp) = rsp_receiver.recv().await {
                    if let Some(transaction) = rsp.transaction.clone() {
                        rsp_map.insert(transaction, rsp).await;
                    }
                }
            }
        });

        let ack_task = jarust_rt::spawn("ACKs gathering task", {
            let ack_map = ack_map.clone();
            async move {
                while let Some(rsp) = ack_receiver.recv().await {
                    if let Some(transaction) = rsp.transaction.clone() {
                        ack_map.insert(transaction, rsp).await;
                    }
                }
            }
        });

        let demux_task = jarust_rt::spawn("Demultiplexing task", {
            let router = router.clone();
            let transaction_manager = transaction_manager.clone();
            let demuxer = Demuxer {
                inbound_stream: receiver,
                router,
                rsp_sender,
                ack_sender,
                transaction_manager,
            };
            async move { demuxer.start().await }
        });

        let shared = Shared {
            tasks: vec![demux_task, rsp_task, ack_task],
            server_root: conn_params.server_root,
            apisecret: conn_params.apisecret,
            transaction_generator,
            transport,
            ack_map,
            rsp_map,
        };
        let exclusive = Exclusive {
            router,
            transaction_manager,
        };
        let inner = InnerCustomInterface {
            shared,
            exclusive: Mutex::new(exclusive),
        };
        Ok(Self {
            inner: Arc::new(inner),
        })
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn send(&self, message: Value) -> Result<String, Error> {
        let (message, transaction) = self.decorate_request(message);

        let path =
            Router::path_from_request(&message).unwrap_or(self.inner.shared.server_root.clone());

        let guard = self.inner.exclusive.lock().await;
        guard.transaction_manager.insert(&transaction, &path).await;
        self.inner
            .shared
            .transport
            .send(message.to_string().as_bytes(), &path)
            .await?;
        tracing::trace!("Sending {message:#?}");
        Ok(transaction)
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip(self, timeout))]
    async fn poll_response(
        &self,
        transaction: &str,
        timeout: Duration,
    ) -> Result<JaResponse, Error> {
        tracing::trace!("Polling response");
        interface_support::poll_transaction(&self.inner.shared.rsp_map, transaction, timeout).await
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip(self, timeout))]
    async fn poll_ack(&self, transaction: &str, timeout: Duration) -> Result<JaResponse, Error> {
        tracing::trace!("Polling ack");
        interface_support::poll_transaction(&self.inner.shared.ack_map, transaction, timeout).await
    }

    fn decorate_request(&self, request: Value) -> (Value, String) {
        interface_support::decorate_request(
            &self.inner.shared.transaction_generator,
            self.inner.shared.apisecret.as_deref(),
            request,
        )
    }
}

#[cfg_attr(not(target_family = "wasm"), async_trait::async_trait)]
#[cfg_attr(target_family = "wasm", async_trait::async_trait(?Send))]
impl JanusInterface for CustomInterface {
    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn make_interface(
        _conn_params: ConnectionParams,
        _transaction_generator: impl GenerateTransaction,
    ) -> Result<Self, Error> {
        // `CustomInterface` needs a caller-supplied `Transport`, which the fixed
        // `make_interface` signature can't carry. Use `CustomInterface::new` instead.
        Err(Error::UnsupportedInterfaceConstructor)
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn create(&self, timeout: Duration) -> Result<u64, Error> {
        let request = json!({
            "janus": "create"
        });

        let transaction = self.send(request).await?;
        let response = self.poll_response(&transaction, timeout).await?;
        let session_id = interface_support::extract_id(response)?;
        Ok(session_id)
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn server_info(&self, timeout: Duration) -> Result<ServerInfoRsp, Error> {
        let request = json!({
            "janus": "info"
        });
        let transaction = self.send(request).await?;
        let response = self.poll_response(&transaction, timeout).await?;
        match response.janus {
            ResponseType::ServerInfo(info) => Ok(*info),
            ResponseType::Error { error } => Err(Error::JanusError {
                code: error.code,
                reason: error.reason,
            }),
            _ => Err(Error::IncompletePacket),
        }
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn attach(
        &self,
        session_id: u64,
        plugin_id: String,
        timeout: Duration,
    ) -> Result<(u64, mpsc::UnboundedReceiver<JaResponse>), Error> {
        let request = json!({
            "janus": "attach",
            "session_id": session_id,
            "plugin": plugin_id
        });
        let transaction = self.send(request).await?;
        let response = self.poll_response(&transaction, timeout).await?;
        let handle_id = interface_support::extract_id(response)?;
        let receiver = self
            .inner
            .exclusive
            .lock()
            .await
            .router
            .add_subroute(&format!("{session_id}/{handle_id}"))
            .await;
        Ok((handle_id, receiver))
    }

    fn has_keep_alive(&self) -> bool {
        true
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn keep_alive(&self, session_id: u64, timeout: Duration) -> Result<(), Error> {
        let request = json!({
            "janus": "keepalive",
            "session_id": session_id
        });
        let transaction = self.send(request).await?;
        self.poll_ack(&transaction, timeout).await?;
        Ok(())
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn destroy(&self, session_id: u64, timeout: Duration) -> Result<(), Error> {
        let request = json!({
            "janus": "destroy",
            "session_id": session_id
        });
        let transaction = self.send(request).await?;
        self.poll_response(&transaction, timeout).await?;
        Ok(())
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn fire_and_forget_msg(&self, message: HandleMessage) -> Result<String, Error> {
        let transaction = self.send(message.to_message_envelope()).await?;
        Ok(transaction)
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn send_msg_waiton_ack(
        &self,
        message: HandleMessage,
        timeout: Duration,
    ) -> Result<String, Error> {
        let transaction = self.send(message.to_message_envelope()).await?;
        self.poll_ack(&transaction, timeout).await?;
        Ok(transaction)
    }

    async fn internal_send_msg_waiton_rsp(
        &self,
        message: HandleMessage,
        timeout: Duration,
    ) -> Result<JaResponse, Error> {
        let transaction = self.send(message.to_message_envelope()).await?;
        self.poll_response(&transaction, timeout).await
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn fire_and_forget_msg_with_jsep(
        &self,
        message: HandleMessageWithJsep,
    ) -> Result<String, Error> {
        let transaction = self.send(message.to_message_envelope()).await?;
        Ok(transaction)
    }

    #[tracing::instrument(level = tracing::Level::TRACE, skip_all)]
    async fn send_msg_waiton_ack_with_jsep(
        &self,
        message: HandleMessageWithJsep,
        timeout: Duration,
    ) -> Result<String, Error> {
        let transaction = self.send(message.to_message_envelope()).await?;
        self.poll_ack(&transaction, timeout).await?;
        Ok(transaction)
    }

    async fn send_handle_request(&self, request: HandleMessage) -> Result<(), Error> {
        let mut req = request.body;
        interface_support::merge_json(
            &mut req,
            &json!({
                "session_id": request.session_id,
                "handle_id": request.handle_id,
            }),
        );
        _ = self.send(req).await?;
        Ok(())
    }

    async fn send_handle_request_waiton_ack(
        &self,
        request: HandleMessage,
        timeout: Duration,
    ) -> Result<String, Error> {
        let mut req = request.body;
        interface_support::merge_json(
            &mut req,
            &json!({
                "session_id": request.session_id,
                "handle_id": request.handle_id,
            }),
        );
        let transaction = self.send(req).await?;
        self.poll_ack(&transaction, timeout).await?;
        Ok(transaction)
    }

    fn name(&self) -> Box<str> {
        "Custom Interface".to_string().into_boxed_str()
    }
}

impl Drop for InnerCustomInterface {
    fn drop(&mut self) {
        self.shared.tasks.iter().for_each(|task| {
            task.cancel();
        });
    }
}