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

Sending the end file event out to listeners.

Browse Source
This commit is contained in:
Tom Alexander 2022-02-21 18:30:03 -05:00
parent 0dd5cf8fe4
commit 860f5759c1
Signed by: talexander
GPG Key ID: D3A179C9A53C0EDE
1 changed files with 56 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
src/mpvctl/mpvctl.rs
View File

@ -88,75 +88,70 @@ impl MpvCtl {
db: &CommandDb,
event_listeners: &EventDb,
) -> Result<(), Box<dyn std::error::Error>> {
match framed_read.read_frame().await {
Ok(Some(frame)) => {
println!("Read {}", frame);
// check if its an event
let is_event = match &frame {
serde_json::Value::Object(obj) => obj.contains_key("event"),
_ => false,
let current_frame = framed_read.read_frame().await.map_err(|e| e.to_string())?;
if let Some(frame) = current_frame {
println!("Read {}", frame);
// check if its an event
let is_event = match &frame {
serde_json::Value::Object(obj) => obj.contains_key("event"),
_ => false,
};
if is_event {
let obj = match &frame {
serde_json::Value::Object(obj) => obj,
_ => {
return Err("Got back a json value that wasn't an object.".into());
}
};
if is_event {
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("event") {
Some(orig @ serde_json::Value::String(s)) => {
println!("Notifying listeners for event {}", s);
let listeners = {
let db_handle = event_listeners.lock().unwrap();
match (*db_handle).get(s) {
None => Vec::new(),
Some(listeners) => listeners.to_owned(),
}
};
for listener in listeners {
listener.send(frame.clone()).await;
match obj.get("event") {
Some(serde_json::Value::String(s)) => {
println!("Notifying listeners for event {}", s);
let listeners = {
let db_handle = event_listeners.lock().unwrap();
match (*db_handle).get(s) {
None => Vec::new(),
Some(listeners) => listeners.to_owned(),
}
}
_ => return Err("Event with no string value".into()),
};
return Ok(());
}
// 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());
};
for listener in listeners {
listener.send(frame.clone()).await?;
}
}
_ => return Err("Event with no string value".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());
// 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());
}
}
}
} else if let None = current_frame {
return Ok(());
};
Ok(())
}