thfmr-protocol 0.2.2

use {
    crate::{channels, decenc, error::Result, message},
    futures::future,
    log::info,
    std::net::IpAddr,
    thfmr_util::DropResult,
    tokio::{io, net, sync, sync::mpsc},
};

/// Producer facing side of the protocol. It is used to setup communication to producers.
///
/// A producer picks songs to play and delivers information about playing songs. It responds to
/// [crate::message::Message::Want] with [crate::message::Message::Next].
pub struct Producer {
    producer_address: (IpAddr, u16),
}

/// An individual connected producer instance
pub struct ProducerClient();

impl Producer {
    /// Construct a new [Producer] listening to the given producer_address.
    pub fn new(producer_address: (IpAddr, u16)) -> Producer {
        Producer { producer_address }
    }

    pub async fn run(
        &self,
        mut channels: channels::ChanPair<message::Message>,
        producer_conn: mpsc::Sender<mpsc::Sender<message::Message>>,
    ) -> Result<()> {
        let client_channels: sync::RwLock<Vec<mpsc::Sender<message::Message>>> =
            sync::RwLock::new(Vec::new());

        let (send_back, recv) = channels.split();

        let fan_out_future = Self::run_fan_out(&client_channels, recv);
        let runner_future = self.run_listener(&client_channels, send_back, producer_conn);

        future::try_join(fan_out_future, runner_future)
            .await
            .drop_result()
    }

    async fn run_fan_out(
        client_channels: &sync::RwLock<Vec<mpsc::Sender<message::Message>>>,
        recv: &mut mpsc::Receiver<message::Message>,
    ) -> Result<()> {
        let mut to_delete = Vec::new();

        loop {
            let received_message = match recv.recv().await {
                Some(s) => s,
                None => return Ok(()),
            };

            let client_channels_guard = client_channels.read().await;

            for (pos, client_channel) in client_channels_guard.iter().enumerate() {
                let res = client_channel.send(received_message.clone()).await;

                if res.is_err() {
                    to_delete.push(pos);
                }
            }
            drop(client_channels_guard);

            let mut client_channels_guard = client_channels.write().await;
            for idx in to_delete.iter().rev() {
                client_channels_guard.remove(*idx);
            }
            to_delete.clear();
            drop(client_channels_guard);
        }
    }

    async fn run_listener(
        &self,
        client_channels: &sync::RwLock<Vec<mpsc::Sender<message::Message>>>,
        send_back: &mpsc::Sender<message::Message>,
        producer_conn: mpsc::Sender<mpsc::Sender<message::Message>>,
    ) -> Result<()> {
        let listener = net::TcpListener::bind(&self.producer_address).await?;

        loop {
            let (socket, address) = listener.accept().await?;

            {
                let (send, recv) = mpsc::channel(1);

                let mut client_channels = client_channels.write().await;
                producer_conn.send(send.clone()).await?;
                client_channels.push(send);

                info!("Connected: {:?}", address);

                let mut channels = channels::ChanPair::from(send_back.clone(), recv);

                tokio::spawn(async move {
                    let mut client = ProducerClient {};

                    if let Err(e) = client.run(socket, &mut channels).await {
                        info!("Terminated: {:?}", e);
                    } else {
                        info!("Connection closed gracefully");
                    };
                });
            }
        }
    }
}

impl ProducerClient {
    async fn run(
        &mut self,
        socket: net::TcpStream,
        channels: &mut channels::ChanPair<message::Message>,
    ) -> Result<()> {
        let (send, mut recv) = channels.split();

        let (r, w) = io::split(socket);

        let read_future = decenc::run_read(r, &send);
        let write_future = decenc::run_write(w, &mut recv);

        future::try_join(read_future, write_future)
            .await
            .drop_result()
    }
}