I have a struct where, you can call a function to get a sender or receiver channel, and a spawn function, the spawn spawns a proxy so you when you ask for the receiver it will give you an unbounded receiver, only by adding the coorosponding unbounded sender to some array. Before this struct has its own internal channels changed so it can start running, I have to decide whether to filter messages from another channel in a stream::unfold or to pass in that same channel directly and let spawn filter it there Basically what I am asking is, is there any time where there is any added benefit of using unfold vs transferring whatever channel you want to use there, and filtering it directly.
impl StreamTransportable for StartServerRequest {
type Output = ();
async fn stream_transport(
&mut self,
arc_state: Arc<ArcSwap<AppState>>,
) -> Result<Self::Output, Box<dyn Error + Send + Sync>> {
let state = arc_state.load();
let connection = state.current_node.connection.clone();
let Some(connection_handler) = connection else {
return Err("No connection working".into());
};
let bytes = convert_into_request(&MessagePayload {
r#type: "command".to_string(),
message: "start_server".to_string(),
authcode: "".to_string(),
}).unwrap();
if *connection_handler.current_active_priority.lock().await > 0 {
return Err("high priority task is occuring and cant be interfered with".into())
}
if connection_handler.proxy_tx.is_none(){
return Err("no stream".into());
}
let proxy_tx = connection_handler.proxy_tx.clone().unwrap();
let _ = proxy_tx.send(bytes);
println!("after sending message");
drop(state);
// let (server_tx, server_rx) = tokio::sync::mpsc::channel(32);
let (tx, proxy_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
let share_tx_guard = connection_handler.share_tx.clone();
let mut share_tx = share_tx_guard.lock().await;
let index = share_tx.len();
share_tx.insert(index, tx);
drop(share_tx);
let console_task = ConsoleTask {
share_tx_guard: share_tx_guard.clone(),
input: proxy_rx,
};
let stream = stream::unfold(console_task, move |mut console_task| {
async move {
loop {
match console_task.input.recv().await {
Some(bytes) => {
if let Ok(value) = serde_json::from_slice::<ConsoleData>(&bytes) {
return Some((value, console_task));
}
}
None => {
let mut share_tx = console_task.share_tx_guard.lock().await;
share_tx.remove(&index);
return None;
}
}
}
}
});
*self.interface.console_out.lock().await = Some(stream.boxed());
*self.interface.console_in.lock().await = Some(proxy_tx.clone());
self.interface.active_event.notify_one();
Ok(())
// Ok(server_rx)
}
}
#[derive(Clone, Default)]
pub struct ConsoleInterface {
active_event: Arc<Notify>,
active: Arc<AtomicBool>,
console_out: Arc<Mutex<Option<Pin<Box<dyn Stream<Item = ConsoleData> + Send + 'static>>>>>,
console_in: Arc<Mutex<Option<UnboundedSender<Vec<u8>>>>>,
proxy_out: Arc<RwLock<Vec<UnboundedSender<ConsoleData>>>>,
proxy_in: Arc<Mutex<Option<broadcast::Sender<ConsoleData>>>>
}
impl ConsoleInterface {
pub async fn spawn(&mut self) -> Result<(), Box<dyn Error + Send + Sync>>{
self.active_event.notified().await;
let Some(ref mut console_out) = *self.console_out.lock().await else {
return Err("no console out".into())
};
self.active.store(true, Ordering::SeqCst);
let (proxy_in_tx, mut proxy_in_rx) = broadcast::channel::<ConsoleData>(32);
*self.proxy_in.lock().await = Some(proxy_in_tx);
*self.proxy_out.write().await = Vec::new();
loop {
tokio::select! {
console_data_res = console_out.next() => {
if let Some(console_data) = console_data_res {
let proxy_out = self.proxy_out.read().await;
for tx in &*proxy_out {
let _ = tx.send(console_data.clone());
}
}
},
Ok(message) = proxy_in_rx.recv(), if self.console_in.lock().await.is_some() => {
let res = self.console_in.lock().await.as_ref().unwrap().send(serde_json::to_vec(&message).unwrap());
println!("got console in res {:#?}", res);
}
}
}
}
pub async fn stdin(&self) -> Option<broadcast::Sender<ConsoleData>> {
self.proxy_in.lock().await.clone()
}
pub async fn stdout(&mut self) -> Option<UnboundedReceiver<ConsoleData>> {
let (proxy_out_tx, proxy_out_rx) = mpsc::unbounded_channel();
self.proxy_out.write().await.push(proxy_out_tx);
if self.active.load(Ordering::SeqCst) {
Some(proxy_out_rx)
} else {
None
}
}
}
Spawn will be called, and it will wait for the console task to start, but I am asking if unfold makes the code more messy or if its bad for polling efficiency or if its worse efficiency or if its just better overall.