- io.rs: factor the pending-RPC-replies map into a PendingReplies type alias (type_complexity) - lua_env.rs: drop a redundant i64->i64 cast and an unneeded borrow before create_string (unnecessary_cast, needless_borrows_for_generic_args) - state_engine.rs: iterate watches.values() instead of discarding the key from watches.iter() (for_kv_map) - module_host_sandbox.rs: collapse nested if into a single condition (collapsible_if) (cherry picked from commit 49063cb98ddfeae35d842a6cf439123cfda763d9)
This commit is contained in:
parent
e073a17353
commit
72afe790fd
4 changed files with 10 additions and 9 deletions
|
|
@ -33,6 +33,10 @@ pub enum IoCommand {
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// In-flight RPC calls awaiting a response, keyed by request id: each entry
|
||||||
|
/// is the reply channel for the call that's blocked waiting on it.
|
||||||
|
type PendingReplies = Arc<Mutex<HashMap<String, mpsc::Sender<Result<Value, String>>>>>;
|
||||||
|
|
||||||
/// Something the IO thread has for the Lua-driving thread: either an
|
/// Something the IO thread has for the Lua-driving thread: either an
|
||||||
/// unsolicited push (a subscribed event fired, a timer fired) or "the
|
/// unsolicited push (a subscribed event fired, a timer fired) or "the
|
||||||
/// connection is gone" (breadd exited, socket closed, etc).
|
/// connection is gone" (breadd exited, socket closed, etc).
|
||||||
|
|
@ -148,8 +152,7 @@ pub fn run(
|
||||||
// Lua-side call is blocked waiting for them; a dedicated thread
|
// Lua-side call is blocked waiting for them; a dedicated thread
|
||||||
// bridges the synchronous `cmd_rx` (fed from the Lua thread) onto an
|
// bridges the synchronous `cmd_rx` (fed from the Lua thread) onto an
|
||||||
// async channel this task can select on.
|
// async channel this task can select on.
|
||||||
let pending: Arc<Mutex<HashMap<String, mpsc::Sender<Result<Value, String>>>>> =
|
let pending: PendingReplies = Arc::new(Mutex::new(HashMap::new()));
|
||||||
Arc::new(Mutex::new(HashMap::new()));
|
|
||||||
|
|
||||||
let (async_cmd_tx, mut async_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<IoCommand>();
|
let (async_cmd_tx, mut async_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<IoCommand>();
|
||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
|
|
|
||||||
|
|
@ -93,7 +93,7 @@ fn json_to_lua<'lua>(lua: &'lua Lua, value: &JsonValue) -> mlua::Result<LuaValue
|
||||||
JsonValue::Bool(b) => LuaValue::Boolean(*b),
|
JsonValue::Bool(b) => LuaValue::Boolean(*b),
|
||||||
JsonValue::Number(n) => {
|
JsonValue::Number(n) => {
|
||||||
if let Some(i) = n.as_i64() {
|
if let Some(i) = n.as_i64() {
|
||||||
LuaValue::Integer(i as i64)
|
LuaValue::Integer(i)
|
||||||
} else {
|
} else {
|
||||||
LuaValue::Number(n.as_f64().unwrap_or(0.0))
|
LuaValue::Number(n.as_f64().unwrap_or(0.0))
|
||||||
}
|
}
|
||||||
|
|
@ -243,7 +243,7 @@ impl ModuleHostLua {
|
||||||
let decode_fn = lua.create_function(|lua, s: String| {
|
let decode_fn = lua.create_function(|lua, s: String| {
|
||||||
match serde_json::from_str::<JsonValue>(&s) {
|
match serde_json::from_str::<JsonValue>(&s) {
|
||||||
Ok(v) => Ok((json_to_lua(lua, &v)?, LuaValue::Nil)),
|
Ok(v) => Ok((json_to_lua(lua, &v)?, LuaValue::Nil)),
|
||||||
Err(e) => Ok((LuaValue::Nil, LuaValue::String(lua.create_string(&e.to_string())?))),
|
Err(e) => Ok((LuaValue::Nil, LuaValue::String(lua.create_string(e.to_string())?))),
|
||||||
}
|
}
|
||||||
})?;
|
})?;
|
||||||
json_tbl.set("decode", decode_fn)?;
|
json_tbl.set("decode", decode_fn)?;
|
||||||
|
|
|
||||||
|
|
@ -260,7 +260,7 @@ pub async fn run_state_engine(
|
||||||
}
|
}
|
||||||
|
|
||||||
if let (Some(before), Some(after)) = (before_snapshot, after_snapshot) {
|
if let (Some(before), Some(after)) = (before_snapshot, after_snapshot) {
|
||||||
for (_id, path) in watches.iter() {
|
for path in watches.values() {
|
||||||
let old_val = value_at_path(&before, path).unwrap_or(Value::Null);
|
let old_val = value_at_path(&before, path).unwrap_or(Value::Null);
|
||||||
let new_val = value_at_path(&after, path).unwrap_or(Value::Null);
|
let new_val = value_at_path(&after, path).unwrap_or(Value::Null);
|
||||||
if old_val != new_val {
|
if old_val != new_val {
|
||||||
|
|
|
||||||
|
|
@ -137,10 +137,8 @@ enabled = false
|
||||||
async fn wait_until_ready(&self) -> Result<()> {
|
async fn wait_until_ready(&self) -> Result<()> {
|
||||||
let deadline = Instant::now() + Duration::from_secs(8);
|
let deadline = Instant::now() + Duration::from_secs(8);
|
||||||
while Instant::now() < deadline {
|
while Instant::now() < deadline {
|
||||||
if self.socket_path.exists() {
|
if self.socket_path.exists() && self.send_request("ping", json!({})).await.is_ok() {
|
||||||
if self.send_request("ping", json!({})).await.is_ok() {
|
return Ok(());
|
||||||
return Ok(());
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
sleep(Duration::from_millis(100)).await;
|
sleep(Duration::from_millis(100)).await;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue