I am currently sending closures/functions across threads.
This works perfectly fine for sync functions.
I am specifically passing
pub type WSMethod<T> = Box<dyn Fn(WSReq, PgConn, &mut WSConnections<T>, Uuid) -> Result<String, BoxError> + Send + Sync>;
Example function being sent
pub fn update_league(req: WSReq, conn: PgConn, _: &mut WSConnections_, _: Uuid) -> Result<String, BoxError>{
let deserialized = serde_json::from_value(req.data)?;
let league = db::update_league(&conn, deserialized)?;
let resp_msg = WSMsgOut::resp(req.message_id, req.method, league);
serde_json::to_string(&resp_msg).map_err(|e| e.into())
}
However Now I would like to switch to sending async functions,
i.e.
pub async fn upsert_competitions(req: WSReq, conn: PgConn, ws_conns: &mut WSConnections_, user_ws_id: Uuid) -> Result<String, BoxError>{
let deserialized: Vec<NewCompetition> = serde_json::from_value(req.data)?;
let competitions_out= db::upsert_competitions(&conn, deserialized.into_iter().map(transform_from).collect_vec())?;
if let Some(ws_user) = ws_conns.lock().await.get_mut(&user_ws_id){
sub_to_competitions(ws_user, competitions_out.iter().map(|c| &c.competition_id)).await;
}
publish_competitions(ws_conns, &competitions_out).await;
let resp_msg = WSMsgOut::resp(req.message_id, req.method, competitions_out);
serde_json::to_string(&resp_msg).map_err(|e| e.into())
}
It's the exact same function signature, it's just async.
Where I box the functions so they can be sent around, I get this error
Box::new(upsert_competitions))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ expected enum `std::result::Result`, found opaque type
full:
288 | pub async fn upsert_competitions(req: WSReq, conn: PgConn, ws_conns: &mut WSConnections_, user_ws_id: Uuid) -> Result<String, BoxError>{
| ------------------------ the `Output` of this `async fn`'s found opaque type
|
= note: expected enum `std::result::Result<std::string::String, std::boxed::Box<dyn std::error::Error + std::marker::Send + std::marker::Sync>>`
found opaque type `impl core::future::future::Future`
= note: required for the cast to the object type `dyn for<'r> std::ops::Fn(warp_ws_server::WSReq, diesel::r2d2::PooledConnection<diesel::r2d2::ConnectionManager<diesel::PgConnection>>, &'r mut std::sync::Arc<tokio::sync::mutex::Mutex<std::collections::HashMap<uuid::Uuid, warp_ws_server::WSConnection<subscriptions::Subscriptions>>>>, uuid::Uuid) -> std::result::Result<std::string::String, std::boxed::Box<dyn std::error::Error + std::marker::Send + std::marker::Sync>> + std::marker::Send + std::marker::Sync`
I have tried attaching .await
to method(req, conn, ws_conns, user_ws_id).await
, the call-site of the passed method.
This causes compiler errors here due to Future
not being implemented for Result
.
Therefore
I change type from: Box<dyn Fn(WSReq, PgConn, &mut WSConnections<T>, Uuid) -> Result<String, BoxError> + Send + Sync>
-> Box<dyn (Fn(WSReq, PgConn, &mut WSConnections<T>, Uuid) -> Future<Output=Result<String, BoxError>>) + Send + Sync>
it complains about sizing of futures, so I box the Future, then another error (see unpin), so I pin the error.
Eventually leading to
Box<dyn (Fn(WSReq, PgConn, &mut WSConnections<T>, Uuid) -> Pin<Box<dyn Future<Output=Result<String, BoxError>> + Send + Sync >>) + Send + Sync>
Error now is
Box::new(upsert_competitions)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ expected struct `std::pin::Pin`, found opaque type
expected struct `std::pin::Pin<std::boxed::Box<dyn core::future::future::Future<Output = std::result::Result<std::string::String, std::boxed::Box<dyn std::error::Error + std::marker::Send + std::marker::Sync>>> + std::marker::Send + std::marker::Sync>>`
found opaque type `impl core::future::future::Future
I dont understand how to go from here. I dont think I should be pinning/boxing the functions results, I want to be pin/boxing the future returned when the function is called, but I dont think I can do this,
as surely i want to be boxing/pinning the future after its created when I call the func, not earlier.
I also tried stuff like
Box::new(Pin::new(Box::new(upsert_competitions))))
based on above error,
and it gives me expected an Fn<blah>
....rather than a Pin<Box<....
Source of full up-to-date code:
closure being successfully passed as a regular function
closure being unsuccessfully passed as an async func
Edit:
Latest updates (progressed the error)
pub fn upsert_competitions(req: WSReq, conn: PgConn, ws_conns: &mut WSConnections_, user_ws_id: Uuid) -> Pin<Box<dyn Future<Output=Result<String, BoxError>> + Send + Sync>>{
async fn hmmm(req: WSReq, conn: PgConn, ws_conns: &mut WSConnections_, user_ws_id: Uuid) -> Result<String, BoxError>{
let deserialized: Vec<NewCompetition> = serde_json::from_value(req.data).expect("fuck");
println!("{:?}", &deserialized);
let competitions_out= db::upsert_competitions(&conn, deserialized.into_iter().map(transform_from).collect_vec()).expect("fuck");
// assume anything upserted the user wants to subscribe to
if let Some(ws_user) = ws_conns.lock().await.get_mut(&user_ws_id){
sub_to_competitions(ws_user, competitions_out.iter().map(|c| &c.competition_id)).await;
}
// TODO ideally would return response before awaiting publishing going out
publish_competitions(ws_conns, &competitions_out).await;
println!("{:?}", &competitions_out);
let resp_msg = WSMsgOut::resp(req.message_id, req.method, competitions_out);
let out = serde_json::to_string(&resp_msg).map_err(|e| e.into());
out
}
Box::pin(hmmm(req, conn, ws_conns, user_ws_id))
}
305 | Box::pin(hmmm(req, conn, ws_conns, user_ws_id))
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ future returned by
`hmmm` is not `Sync`
So now just need to work out how to make this future sync
note: future is not `Sync` as this value is used across an await
gives me good clue
299 | publish_competitions(ws_conns, &competitions_out).await;
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ await
occurs here, with conn
maybe used later
Worked out that I have to keep use of conn
outside of inner-async function, and not use across an await.
After fixing variables across await, I now arrive at
error[E0621]: explicit lifetime required in the type of `ws_conns`
--> src/handlers.rs:305:5
|
289 | pub fn upsert_competitions(req: WSReq, conn: PgConn, ws_conns: &mut WSConnections_, user_ws_id: Uuid) -> Pin<Box<dyn Future<Output=Result<String, BoxError>> + Send + Sync>>{
| ------------------- help: add explicit lifetime `'static` to the type of `ws_conns`: `&'static mut std::sync::Arc<tokio::sync::mutex::Mutex<std::collections::HashMap<uuid::Uuid, warp_ws_server::WSConnection<subscriptions::Subscriptions>>>>`
...
305 | Box::pin(hmmm(req, competitions_out, ws_conns, user_ws_id))
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ lifetime `'static` required
Tried making &'static references, but eventually I get to point where that's not ok.
I also tried using upsert_competitions<U: lock_api::RawMutex + 'static>
generic types instead,
however getting the trait lock_api::mutex::RawMutex
is not implemented for std::sync::Arc<tokio::sync::mutex::Mutex<std::collections::HashMap<uuid::Uuid, warp_ws_server::WSConnection<subscriptions::Subscriptions>>>>
I need to find a U that implements .lock(), but is also a trait that Arc implements.