1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

use crate::error::{Error, Result};
use crate::persistent_client::PersistentClient;
use mpd_client::client::{CommandError, ConnectionEvent};
use mpd_client::responses::{PlayState, SongInQueue, Status};
use mpd_client::Client;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

pub struct MultiHostClient<'a> {
    clients: Vec<PersistentClient<'a>>,
}

impl<'a> MultiHostClient<'a> {
    pub fn new(hosts: &'a [&'a str], retry_interval: Duration) -> Self {
        let hosts = hosts
            .iter()
            .map(|&host| PersistentClient::new(host, retry_interval))
            .collect();

        Self { clients: hosts }
    }

    /// Initialises each of the clients.
    pub fn init(&self) {
        for client in &self.clients {
            client.init();
        }
    }

    /// Waits until any of the clients
    /// make a valid connection to their host.
    pub async fn wait_for_any_client(&self) -> Arc<Client> {
        let waits = self
            .clients
            .iter()
            .map(|client| Box::pin(client.wait_for_client()));
        futures::future::select_all(waits).await.0
    }

    /// Waits until all of the clients
    /// make a valid connection to their host.
    pub async fn wait_for_all_clients(&self) -> Vec<Arc<Client>> {
        let waits = self.clients.iter().map(|client| client.wait_for_client());
        futures::future::join_all(waits).await
    }

    /// Attempts to find the current most relevant client.
    /// This checks for, in order:
    ///
    /// - A currently playing client
    /// - A paused client (ie has items in the playlist)
    /// - A connected client
    async fn get_current_client(
        &self,
    ) -> std::result::Result<Option<&PersistentClient>, CommandError> {
        self.wait_for_any_client().await;

        let connected_clients = self
            .clients
            .iter()
            .filter(|client| client.is_connected())
            .collect::<Vec<_>>();

        if connected_clients.is_empty() {
            Ok(None)
        } else {
            let player_states = connected_clients.iter().map(|&client| async move {
                client.status().await.map(|status| (client, status.state))
            });

            let player_states = futures::future::join_all(player_states)
                .await
                .into_iter()
                .collect::<std::result::Result<Vec<_>, _>>();

            player_states.map(|player_states| {
                player_states
                    .iter()
                    .find(|(_, state)| state == &PlayState::Playing)
                    .or_else(|| {
                        player_states
                            .iter()
                            .find(|(_, state)| state == &PlayState::Paused)
                    })
                    .or_else(|| {
                        player_states
                            .iter()
                            .find(|(_, state)| state == &PlayState::Stopped)
                    })
                    .map(|(client, _)| *client)
            })
        }
    }

    /// Runs the provided callback as soon as a connected client is available,
    /// using the most relevant client (see `get_current_client`).
    pub async fn with_client<F, Fut, T>(&self, f: F) -> Result<T>
    where
        F: FnOnce(Arc<Client>) -> Fut,
        Fut: Future<Output = T>,
    {
        let client = self.get_current_client().await;

        match client {
            Ok(Some(client)) => Ok(client.with_client(f).await),
            Ok(None) => Err(Error::NoHostConnectedError),
            Err(err) => Err(Error::CommandError(err)),
        }
    }

    pub async fn recv(&mut self) -> Option<ConnectionEvent> {
        let waits = self.clients.iter().map(|client| Box::pin(client.recv()));
        futures::future::select_all(waits).await.0
    }

    /// Runs the `status` command on the MPD server.
    pub async fn status(&self) -> Result<Status> {
        let client = self.get_current_client().await;
        match client {
            Ok(Some(client)) => client.status().await.map_err(Error::CommandError),
            Ok(None) => Err(Error::NoHostConnectedError),
            Err(err) => Err(Error::CommandError(err)),
        }
    }

    /// Runs the `currentsong` command on the MPD server.
    pub async fn current_song(&self) -> Result<Option<SongInQueue>> {
        match self.get_current_client().await {
            Ok(Some(client)) => client.current_song().await.map_err(Error::CommandError),
            Ok(None) => Err(Error::NoHostConnectedError),
            Err(err) => Err(Error::CommandError(err)),
        }
    }
}

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

    #[tokio::test]
    async fn test() {
        let client =
            MultiHostClient::new(&["localhost:6600", "chloe:6600"], Duration::from_secs(5));

        client.init();
        client.wait_for_all_clients().await;

        let current_client = client.get_current_client().await;
        println!("{current_client:?}");
    }
}