talexander/record_watch
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Fix hang issue by moving loop logic into separate function to catch all errors.

Browse Source
This commit is contained in:
Tom Alexander 2022-02-20 22:09:02 -05:00
parent 601c1a5cbc
commit e85f9d3d41
Signed by: talexander
GPG Key ID: D3A179C9A53C0EDE
1 changed files with 50 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
src/mpvctl/mpvctl.rs
View File

@ -33,47 +33,12 @@ impl MpvCtl {
tokio::spawn(async move { tokio::spawn(async move {
loop { loop {
match framed_read.read_frame().await { match Self::read_loop(&mut framed_read, &read_db_handle).await {
Ok(Some(frame)) => { Ok(_) => (),
println!("Read {}", frame);
// get the request id and push the result into the channel
let reqid = {
let obj = match &frame {
serde_json::Value::Object(obj) => obj,
_ => {
return Err(
"Got back a json value that wasn't an object.".to_string()
);
}
};
match obj.get("request_id") {
Some(serde_json::Value::Number(reqid)) if reqid.is_u64() => {
reqid.as_u64().unwrap()
}
_ => {
return Err("Unrecognized request_id".to_string());
}
}
};
{
let mut db_handle = read_db_handle.lock().unwrap();
if let Entry::Occupied(o) = (*db_handle).entry(reqid) {
o.remove()
.send(frame)
.map_err(|e| "Failed to send frame".to_string())?;
} else {
return Err("No entry found for request id".to_string());
}
}
}
Ok(None) => {
return Ok(());
}
Err(e) => { Err(e) => {
return Err("Ran into a problem".to_string()); eprintln!("Error in read loop: {}", e);
}
} }
};
} }
}); });
@ -84,6 +49,52 @@ impl MpvCtl {
}) })
} }
async fn read_loop(
framed_read: &mut MpvFramed,
db: &CommandDb,
) -> Result<(), Box<dyn std::error::Error>> {
match framed_read.read_frame().await {
Ok(Some(frame)) => {
println!("Read {}", frame);
// get the request id and push the result into the channel
let reqid = {
let obj = match &frame {
serde_json::Value::Object(obj) => obj,
_ => {
return Err("Got back a json value that wasn't an object.".into());
}
};
match obj.get("request_id") {
Some(serde_json::Value::Number(reqid)) if reqid.is_u64() => {
reqid.as_u64().unwrap()
}
_ => {
return Err("Unrecognized request_id".into());
}
}
};
{
let mut db_handle = db.lock().unwrap();
if let Entry::Occupied(o) = (*db_handle).entry(reqid) {
o.remove()
.send(frame)
.map_err(|e| "Failed to send frame".to_string())?;
} else {
return Err("No entry found for request id".into());
}
}
}
Ok(None) => {
return Ok(());
}
Err(e) => {
return Err("Ran into a problem".into());
}
}
Ok(())
}
fn grab_request_id(&mut self) -> u64 { fn grab_request_id(&mut self) -> u64 {
let ret = self.next_request_id; let ret = self.next_request_id;
self.next_request_id += 1; self.next_request_id += 1;