1
//! IPC handling thread/task. Handles communication between [`Mpv`](crate::Mpv) instances and mpv's unix socket
2

            
3
use futures::{SinkExt, StreamExt};
4
use serde_json::{Value, json};
5
use tokio::{
6
    net::UnixStream,
7
    sync::{broadcast, mpsc, oneshot},
8
};
9
use tokio_util::codec::{Framed, LinesCodec};
10

            
11
use crate::MpvError;
12

            
13
/// Container for all state that regards communication with the mpv IPC socket
14
/// and message passing with [`Mpv`](crate::Mpv) controllers.
15
pub(crate) struct MpvIpc {
16
    socket: Framed<UnixStream, LinesCodec>,
17
    command_channel: mpsc::Receiver<(MpvIpcCommand, oneshot::Sender<MpvIpcResponse>)>,
18
    event_channel: broadcast::Sender<MpvIpcEvent>,
19
}
20

            
21
/// Commands that can be sent to [`MpvIpc`]
22
#[derive(Debug, Clone, PartialEq, Eq)]
23
pub(crate) enum MpvIpcCommand {
24
    Command(Vec<String>),
25
    GetProperty(String),
26
    SetProperty(String, Value),
27
    ObserveProperty(u64, String),
28
    UnobserveProperty(u64),
29
    Exit,
30
}
31

            
32
/// [`MpvIpc`]'s response to a [`MpvIpcCommand`].
33
#[derive(Debug)]
34
pub(crate) struct MpvIpcResponse(pub(crate) Result<Option<Value>, MpvError>);
35

            
36
/// A deserialized and partially parsed event from mpv.
37
#[derive(Debug, Clone)]
38
pub(crate) struct MpvIpcEvent(pub(crate) Value);
39

            
40
impl MpvIpc {
41
44
    pub(crate) fn new(
42
44
        socket: UnixStream,
43
44
        command_channel: mpsc::Receiver<(MpvIpcCommand, oneshot::Sender<MpvIpcResponse>)>,
44
44
        event_channel: broadcast::Sender<MpvIpcEvent>,
45
44
    ) -> Self {
46
44
        MpvIpc {
47
44
            socket: Framed::new(socket, LinesCodec::new()),
48
44
            command_channel,
49
44
            event_channel,
50
44
        }
51
44
    }
52

            
53
7440
    pub(crate) async fn send_command(
54
7440
        &mut self,
55
7440
        command: &[Value],
56
7440
    ) -> Result<Option<Value>, MpvError> {
57
3720
        let ipc_command = json!({ "command": command });
58
3720
        let ipc_command_str =
59
3720
            serde_json::to_string(&ipc_command).map_err(MpvError::JsonParseError)?;
60

            
61
3720
        log::trace!("Sending command: {}", ipc_command_str);
62

            
63
3720
        self.socket
64
3720
            .send(ipc_command_str)
65
3720
            .await
66
3720
            .map_err(|why| MpvError::MpvSocketConnectionError(why.to_string()))?;
67

            
68
3716
        let response = loop {
69
3718
            let response = self
70
3718
                .socket
71
3718
                .next()
72
3718
                .await
73
3716
                .ok_or(MpvError::MpvSocketConnectionError(
74
3716
                    "Could not receive response from mpv".to_owned(),
75
3716
                ))?
76
3716
                .map_err(|why| MpvError::MpvSocketConnectionError(why.to_string()))?;
77

            
78
3716
            let parsed_response =
79
3716
                serde_json::from_str::<Value>(&response).map_err(MpvError::JsonParseError);
80

            
81
3716
            if parsed_response
82
3716
                .as_ref()
83
3716
                .ok()
84
3716
                .and_then(|v| v.as_object().map(|o| o.contains_key("event")))
85
3716
                .unwrap_or(false)
86
            {
87
                self.handle_event(parsed_response).await;
88
            } else {
89
3716
                break parsed_response;
90
            }
91
        };
92

            
93
3716
        log::trace!("Received response: {:?}", response);
94

            
95
3716
        parse_mpv_response_data(response?, command)
96
3718
    }
97

            
98
3636
    pub(crate) async fn get_mpv_property(
99
3636
        &mut self,
100
3636
        property: &str,
101
3636
    ) -> Result<Option<Value>, MpvError> {
102
1818
        self.send_command(&[json!("get_property"), json!(property)])
103
1818
            .await
104
1817
    }
105

            
106
3772
    pub(crate) async fn set_mpv_property(
107
3772
        &mut self,
108
3772
        property: &str,
109
3772
        value: Value,
110
3772
    ) -> Result<Option<Value>, MpvError> {
111
1886
        self.send_command(&[json!("set_property"), json!(property), value])
112
1886
            .await
113
1885
    }
114

            
115
12
    pub(crate) async fn observe_property(
116
12
        &mut self,
117
12
        id: u64,
118
12
        property: &str,
119
12
    ) -> Result<Option<Value>, MpvError> {
120
6
        self.send_command(&[json!("observe_property"), json!(id), json!(property)])
121
6
            .await
122
6
    }
123

            
124
2
    pub(crate) async fn unobserve_property(&mut self, id: u64) -> Result<Option<Value>, MpvError> {
125
1
        self.send_command(&[json!("unobserve_property"), json!(id)])
126
1
            .await
127
1
    }
128

            
129
34
    async fn handle_event(&mut self, event: Result<Value, MpvError>) {
130
17
        match &event {
131
17
            Ok(event) => {
132
17
                log::trace!("Parsed event: {:?}", event);
133
                if let Err(broadcast::error::SendError(_)) =
134
17
                    self.event_channel.send(MpvIpcEvent(event.to_owned()))
135
                {
136
                    log::trace!("Failed to send event to channel, ignoring");
137
17
                }
138
            }
139
            Err(e) => {
140
                log::trace!("Error parsing event, ignoring:\n  {:?}\n  {:?}", &event, e);
141
            }
142
        }
143
17
    }
144

            
145
44
    pub(crate) async fn run(mut self) -> Result<(), MpvError> {
146
        loop {
147
3757
            tokio::select! {
148
3757
              Some(event) = self.socket.next() => {
149
17
                log::trace!("Got event: {:?}", event);
150

            
151
17
                let parsed_event = event
152
17
                    .map_err(|why| MpvError::MpvSocketConnectionError(why.to_string()))
153
17
                    .and_then(|event|
154
17
                        serde_json::from_str::<Value>(&event)
155
17
                        .map_err(MpvError::JsonParseError));
156

            
157
17
                self.handle_event(parsed_event).await;
158
              }
159
3757
              Some((cmd, tx)) = self.command_channel.recv() => {
160
3720
                  log::trace!("Handling command: {:?}", cmd);
161
3720
                  match cmd {
162
9
                      MpvIpcCommand::Command(command) => {
163
9
                          let refs = command.iter().map(|s| json!(s)).collect::<Vec<Value>>();
164
9
                          let response = self.send_command(refs.as_slice()).await;
165
9
                          tx.send(MpvIpcResponse(response)).unwrap()
166
                      }
167
1818
                      MpvIpcCommand::GetProperty(property) => {
168
1818
                          let response = self.get_mpv_property(&property).await;
169
1817
                          tx.send(MpvIpcResponse(response)).unwrap()
170
                      }
171
1886
                      MpvIpcCommand::SetProperty(property, value) => {
172
1886
                          let response = self.set_mpv_property(&property, value).await;
173
1885
                          tx.send(MpvIpcResponse(response)).unwrap()
174
                      }
175
6
                      MpvIpcCommand::ObserveProperty(id, property) => {
176
6
                          let response = self.observe_property(id, &property).await;
177
6
                          tx.send(MpvIpcResponse(response)).unwrap()
178
                      }
179
1
                      MpvIpcCommand::UnobserveProperty(id) => {
180
1
                          let response = self.unobserve_property(id).await;
181
1
                          tx.send(MpvIpcResponse(response)).unwrap()
182
                      }
183
                      MpvIpcCommand::Exit => {
184
                        tx.send(MpvIpcResponse(Ok(None))).unwrap();
185
                        return Ok(());
186
                      }
187
                  }
188
              }
189
            }
190
        }
191
    }
192
}
193

            
194
/// This function does the most basic JSON parsing and error handling
195
/// for status codes and errors that all responses from mpv are
196
/// expected to contain.
197
7432
fn parse_mpv_response_data(value: Value, command: &[Value]) -> Result<Option<Value>, MpvError> {
198
7432
    log::trace!("Parsing mpv response data: {:?}", value);
199
7432
    let result = value
200
7432
        .as_object()
201
7432
        .ok_or(MpvError::ValueContainsUnexpectedType {
202
7432
            expected_type: "object".to_string(),
203
7432
            received: value.clone(),
204
7432
        })
205
7432
        .and_then(|o| {
206
7432
            let error = o
207
7432
                .get("error")
208
7432
                .ok_or(MpvError::MissingKeyInObject {
209
7432
                    key: "error".to_string(),
210
7432
                    map: o.clone(),
211
7432
                })?
212
7432
                .as_str()
213
7432
                .ok_or(MpvError::ValueContainsUnexpectedType {
214
7432
                    expected_type: "string".to_string(),
215
7432
                    received: o.get("error").unwrap().clone(),
216
7432
                })?;
217

            
218
7432
            let data = o.get("data");
219

            
220
7432
            Ok((error, data))
221
7432
        })
222
7432
        .and_then(|(error, data)| match error {
223
7432
            "success" => Ok(data),
224
274
            "property unavailable" => Ok(None),
225
270
            err => Err(MpvError::MpvError {
226
270
                command: command.to_owned(),
227
270
                message: err.to_string(),
228
270
            }),
229
7432
        });
230

            
231
7432
    match &result {
232
7162
        Ok(v) => log::trace!("Successfully parsed mpv response data: {:?}", v),
233
270
        Err(e) => log::trace!("Error parsing mpv response data: {:?}", e),
234
    }
235

            
236
7432
    result.map(|opt| opt.cloned())
237
7432
}