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

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

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

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

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

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

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

            
95
3027
        parse_mpv_response_data(response?, command)
96
3029
    }
97

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

            
106
3072
    pub(crate) async fn set_mpv_property(
107
3072
        &mut self,
108
3072
        property: &str,
109
3072
        value: Value,
110
3072
    ) -> Result<Option<Value>, MpvError> {
111
1536
        self.send_command(&[json!("set_property"), json!(property), value])
112
1536
            .await
113
1535
    }
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
3068
            tokio::select! {
148
3068
              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
3068
              Some((cmd, tx)) = self.command_channel.recv() => {
160
3031
                  log::trace!("Handling command: {:?}", cmd);
161
3031
                  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
1479
                      MpvIpcCommand::GetProperty(property) => {
168
1479
                          let response = self.get_mpv_property(&property).await;
169
1478
                          tx.send(MpvIpcResponse(response)).unwrap()
170
                      }
171
1536
                      MpvIpcCommand::SetProperty(property, value) => {
172
1536
                          let response = self.set_mpv_property(&property, value).await;
173
1535
                          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
6054
fn parse_mpv_response_data(value: Value, command: &[Value]) -> Result<Option<Value>, MpvError> {
198
6054
    log::trace!("Parsing mpv response data: {:?}", value);
199
6054
    let result = value
200
6054
        .as_object()
201
6054
        .ok_or(MpvError::ValueContainsUnexpectedType {
202
6054
            expected_type: "object".to_string(),
203
6054
            received: value.clone(),
204
6054
        })
205
6054
        .and_then(|o| {
206
6054
            let error = o
207
6054
                .get("error")
208
6054
                .ok_or(MpvError::MissingKeyInObject {
209
6054
                    key: "error".to_string(),
210
6054
                    map: o.clone(),
211
6054
                })?
212
6054
                .as_str()
213
6054
                .ok_or(MpvError::ValueContainsUnexpectedType {
214
6054
                    expected_type: "string".to_string(),
215
6054
                    received: o.get("error").unwrap().clone(),
216
6054
                })?;
217

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

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

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