aboutsummaryrefslogtreecommitdiffstats
path: root/lampada/src/connection/read.rs
diff options
context:
space:
mode:
authorLibravatar cel 🌸 <cel@bunny.garden>2025-04-17 11:03:51 +0100
committerLibravatar cel 🌸 <cel@bunny.garden>2025-04-17 11:03:51 +0100
commitb9d75f38743113c054be3d97af36bdd2a7dd0d69 (patch)
tree537623664010d26d5f2574e2d51d03f8c25e08ac /lampada/src/connection/read.rs
parentcf51dcf052af89f8742d887bde2c93d735309bdd (diff)
downloadluz-b9d75f38743113c054be3d97af36bdd2a7dd0d69.tar.gz
luz-b9d75f38743113c054be3d97af36bdd2a7dd0d69.tar.bz2
luz-b9d75f38743113c054be3d97af36bdd2a7dd0d69.zip
feat(filamento): compiles on wasm
Diffstat (limited to 'lampada/src/connection/read.rs')
-rw-r--r--lampada/src/connection/read.rs55
1 files changed, 33 insertions, 22 deletions
diff --git a/lampada/src/connection/read.rs b/lampada/src/connection/read.rs
index 4451d99..591a2cb 100644
--- a/lampada/src/connection/read.rs
+++ b/lampada/src/connection/read.rs
@@ -16,6 +16,8 @@ use tokio::{
sync::{mpsc, oneshot, Mutex},
task::{JoinHandle, JoinSet},
};
+#[cfg(target_arch = "wasm32")]
+use tokio_with_wasm::alias as tokio;
use tracing::info;
use crate::{Connected, Logic, WriteMessage};
@@ -29,7 +31,7 @@ pub struct Read<Lgc> {
disconnect_timedout: Fuse<oneshot::Receiver<()>>,
// all the threads spawned by the current connection session
- tasks: JoinSet<()>,
+ // tasks: JoinSet<()>,
// for handling incoming stanzas
// jabber server must be able to both terminate the connection from error, and ask for data from the client (such as supported XEPs)
@@ -46,13 +48,13 @@ pub struct Read<Lgc> {
pub struct ReadState {
pub supervisor_control: SupervisorSender,
// TODO: when a stream dies, the iq gets from the server should not be replied to on the new stream
- pub tasks: JoinSet<()>,
+ // pub tasks: JoinSet<()>,
}
impl<Lgc> Read<Lgc> {
fn new(
stream: BoundJabberReader,
- tasks: JoinSet<()>,
+ // tasks: JoinSet<()>,
connected: Connected,
logic: Lgc,
supervisor_control: SupervisorSender,
@@ -64,7 +66,7 @@ impl<Lgc> Read<Lgc> {
stream,
disconnecting: false,
disconnect_timedout: recv.fuse(),
- tasks,
+ // tasks,
connected,
logic,
supervisor_control,
@@ -74,7 +76,12 @@ impl<Lgc> Read<Lgc> {
}
}
-impl<Lgc: Clone + Logic + Send + 'static> Read<Lgc> {
+impl<Lgc> Read<Lgc>
+where
+ Lgc: Logic + Clone + 'static,
+ #[cfg(not(target_arch = "wasm32"))]
+ Lgc: Send,
+{
async fn run(mut self) {
println!("started read thread");
// let stanza = self.stream.read::<Stanza>().await;
@@ -100,7 +107,7 @@ impl<Lgc: Clone + Logic + Send + 'static> Read<Lgc> {
})
},
ReadControl::Abort(sender) => {
- let _ = sender.send(ReadState { supervisor_control: self.supervisor_control, tasks: self.tasks });
+ let _ = sender.send(ReadState { supervisor_control: self.supervisor_control });
break;
},
};
@@ -112,11 +119,11 @@ impl<Lgc: Clone + Logic + Send + 'static> Read<Lgc> {
match s {
Stanza::Error(error) => {
self.logic.clone().handle_stream_error(error).await;
- self.supervisor_control.send(SupervisorCommand::Reconnect(ReadState { supervisor_control: self.supervisor_control.clone(), tasks: self.tasks })).await;
+ self.supervisor_control.send(SupervisorCommand::Reconnect(ReadState { supervisor_control: self.supervisor_control.clone() })).await;
break;
},
_ => {
- self.tasks.spawn(self.logic.clone().handle_stanza(s, self.connected.clone()));
+ tokio::spawn(self.logic.clone().handle_stanza(s, self.connected.clone()));
}
};
},
@@ -143,7 +150,7 @@ impl<Lgc: Clone + Logic + Send + 'static> Read<Lgc> {
_ => None,
};
- let _ = self.on_crash.send((stream_error, ReadState { supervisor_control: self.supervisor_control, tasks: self.tasks }));
+ let _ = self.on_crash.send((stream_error, ReadState { supervisor_control: self.supervisor_control }));
}
break;
},
@@ -189,27 +196,29 @@ impl DerefMut for ReadControlHandle {
}
impl ReadControlHandle {
- pub fn new<Lgc: Clone + Logic + Send + 'static>(
+ pub fn new<Lgc>(
stream: BoundJabberReader,
connected: Connected,
logic: Lgc,
supervisor_control: SupervisorSender,
on_crash: oneshot::Sender<(Option<StreamError>, ReadState)>,
- ) -> Self {
+ ) -> Self
+ where
+ Lgc: Logic + Clone + 'static,
+ #[cfg(not(target_arch = "wasm32"))]
+ Lgc: Send,
+ {
let (control_sender, control_receiver) = mpsc::channel(20);
let actor = Read::new(
stream,
- JoinSet::new(),
+ // JoinSet::new(),
connected,
logic,
supervisor_control,
control_receiver,
on_crash,
);
- #[cfg(target_arch = "wasm32")]
- wasm_bindgen_futures::spawn_local(async move { actor.run().await });
- #[cfg(not(target_arch = "wasm32"))]
tokio::spawn(async move { actor.run().await });
Self {
@@ -217,28 +226,30 @@ impl ReadControlHandle {
}
}
- pub fn reconnect<Lgc: Clone + Logic + Send + 'static>(
+ pub fn reconnect<Lgc>(
stream: BoundJabberReader,
- tasks: JoinSet<()>,
+ // tasks: JoinSet<()>,
connected: Connected,
logic: Lgc,
supervisor_control: SupervisorSender,
on_crash: oneshot::Sender<(Option<StreamError>, ReadState)>,
- ) -> Self {
+ ) -> Self
+ where
+ Lgc: Logic + Clone + 'static,
+ #[cfg(not(target_arch = "wasm32"))]
+ Lgc: Send,
+ {
let (control_sender, control_receiver) = mpsc::channel(20);
let actor = Read::new(
stream,
- tasks,
+ // tasks,
connected,
logic,
supervisor_control,
control_receiver,
on_crash,
);
- #[cfg(target_arch = "wasm32")]
- wasm_bindgen_futures::spawn_local(async move { actor.run().await });
- #[cfg(not(target_arch = "wasm32"))]
tokio::spawn(async move { actor.run().await });
Self {