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

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

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

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

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

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

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

            
95
3075
        parse_mpv_response_data(response?, command)
96
3077
    }
97

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

            
106
3148
    pub(crate) async fn set_mpv_property(
107
3148
        &mut self,
108
3148
        property: &str,
109
3148
        value: Value,
110
3148
    ) -> Result<Option<Value>, MpvError> {
111
1574
        self.send_command(&[json!("set_property"), json!(property), value])
112
1574
            .await
113
1573
    }
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
32
    async fn handle_event(&mut self, event: Result<Value, MpvError>) {
130
16
        match &event {
131
16
            Ok(event) => {
132
16
                log::trace!("Parsed event: {:?}", event);
133
                if let Err(broadcast::error::SendError(_)) =
134
16
                    self.event_channel.send(MpvIpcEvent(event.to_owned()))
135
                {
136
3
                    log::trace!("Failed to send event to channel, ignoring");
137
13
                }
138
            }
139
            Err(e) => {
140
                log::trace!("Error parsing event, ignoring:\n  {:?}\n  {:?}", &event, e);
141
            }
142
        }
143
16
    }
144

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

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

            
157
16
                self.handle_event(parsed_event).await;
158
              }
159
3115
              Some((cmd, tx)) = self.command_channel.recv() => {
160
3079
                  log::trace!("Handling command: {:?}", cmd);
161
3079
                  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
1489
                      MpvIpcCommand::GetProperty(property) => {
168
1489
                          let response = self.get_mpv_property(&property).await;
169
1488
                          tx.send(MpvIpcResponse(response)).unwrap()
170
                      }
171
1574
                      MpvIpcCommand::SetProperty(property, value) => {
172
1574
                          let response = self.set_mpv_property(&property, value).await;
173
1573
                          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
6150
fn parse_mpv_response_data(value: Value, command: &[Value]) -> Result<Option<Value>, MpvError> {
198
6150
    log::trace!("Parsing mpv response data: {:?}", value);
199
6150
    let result = value
200
6150
        .as_object()
201
6150
        .ok_or(MpvError::ValueContainsUnexpectedType {
202
6150
            expected_type: "object".to_string(),
203
6150
            received: value.clone(),
204
6150
        })
205
6150
        .and_then(|o| {
206
6150
            let error = o
207
6150
                .get("error")
208
6150
                .ok_or(MpvError::MissingKeyInObject {
209
6150
                    key: "error".to_string(),
210
6150
                    map: o.clone(),
211
6150
                })?
212
6150
                .as_str()
213
6150
                .ok_or(MpvError::ValueContainsUnexpectedType {
214
6150
                    expected_type: "string".to_string(),
215
6150
                    received: o.get("error").unwrap().clone(),
216
6150
                })?;
217

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

            
220
6150
            Ok((error, data))
221
6150
        })
222
6150
        .and_then(|(error, data)| match error {
223
6150
            "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
6150
        });
230

            
231
6150
    match &result {
232
5880
        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
6150
    result.map(|opt| opt.cloned())
237
6150
}