I need some helping understanding how to properly create a subscription stream that will accept a multitude of subscriptions from various parts of my application and backend network logic.
My backend is using libp2p and so contains a network_event_stream that I need my application to keep track of, on top of potential application specific events.
I believe this implies I create an mpsc channel between the main app bodies fn subscription, and the nested page fn subscription(), instead of just propagating back. Is this assumption correct? Or is there a better solution.
The example given by the docs of a subsciption channel is: (not sure how to modify this to work with my program structure)
use iced_native::subscription::{self, Subscription};
use iced_native::futures::channel::mpsc;
use iced_native::futures::sink::SinkExt;
pub enum Event {
Ready(mpsc::Sender<Input>),
WorkFinished,
// ...
}
enum Input {
DoSomeWork,
// ...
}
enum State {
Starting,
Ready(mpsc::Receiver<Input>),
}
fn some_worker() -> Subscription<Event> {
struct SomeWorker;
subscription::channel(std::any::TypeId::of::<SomeWorker>(), 100, |mut output| async move {
let mut state = State::Starting;
loop {
match &mut state {
State::Starting => {
// Create channel
let (sender, receiver) = mpsc::channel(100);
// Send the sender back to the application
output.send(Event::Ready(sender)).await;
// We are ready to receive messages
state = State::Ready(receiver);
}
State::Ready(receiver) => {
use iced_native::futures::StreamExt;
// Read next input sent from `Application`
let input = receiver.select_next_some().await;
match input {
Input::DoSomeWork => {
// Do some async work...
// Finally, we can optionally produce a message to tell the
// `Application` the work is done
output.send(Event::WorkFinished).await;
}
}
}
}
}
})
}
My program is currently split into several nested parts and my current subscription method is invalid as it only subscribes to subscriptions on pages that are in focus, instead of running constantly throughout the lifetime of the application.
My implementation looks like this:
#[derive(Debug)]
pub struct ProjectGui {
pages: Pages,
}
#[derive(Debug, Clone)]
pub enum Message {
PageMessage(PageMessage),
// skipped code
}
impl Application for ProjectGui {
type Executor = executor::Default; // Can I expand on this
type Message = Message;
type Theme = Theme;
type Flags = ();
// skipped code
fn update(&mut self, message: Self::Message) -> Command<Message> {
match message {
Message::PageMessage(message) => {
return self.pages.update(message);
},
}
Command::none()
}
// skipped code
fn subscription(&self) -> Subscription<Message> {
self.pages.subscription().map(|message| Message::PageMessage(message))
}
}
#[derive(Debug)]
pub struct Pages {
pub pages: Vec<Page>,
pub current: usize,
pub debug: bool,
pub theme: Theme,
}
impl<'a> Pages {
// skipped code
pub fn update(&mut self, message: PageMessage) -> Command<Message> {
return self
.pages[self.current]
.update(message, &mut self.debug, &mut self.theme)
.map(|message| Message::PageMessage(message));
}
// skipped code
pub fn subscription(&self) -> Subscription<PageMessage>{
self.pages[self.current].subscription()
}
}
#[derive(Debug)]
pub enum Page {
SoftDemo(SoftDemo),
// skipped code
}
#[derive(Debug, Clone)]
pub enum PageMessage {
SoftDemo(SoftDemoMessage),
// skipped code
}
impl<'a> Page {
pub fn update(&mut self, message: PageMessage, debug: &mut bool, theme: &mut Theme) -> Command<PageMessage> {
match message {
PageMessage::SoftDemo(message) => {
if let Page::SoftDemo(soft_demo) = self {
return soft_demo
.update(message)
.map(|message| PageMessage::SoftDemo(message));
}
},
}
Command::none()
}
// skipped code
pub fn subscription(&self) -> Subscription<PageMessage> {
match self {
Page::SoftDemo(soft_demo) =>{
soft_demo
.subscription()
.map(|message| PageMessage::SoftDemo(message))
},
_ => Subscription::none()
}
}
}
// skipped code
#[derive(Debug)]
pub struct SoftDemo {
pub simulation: Simulation,
pub is_playing: bool,
pub queued_ticks: usize,
pub speed: usize,
pub next_speed: Option<usize>,
pub version: usize,
}
#[derive(Debug, Clone)]
pub enum SoftDemoMessage {
Tick(Instant),
// skipped code
}
impl<'a> SoftDemo {
pub fn new() -> Self {
Self {
simulation: Simulation::new(),
is_playing: Default::default(),
queued_ticks: Default::default(),
speed: 5,
next_speed: Default::default(),
version: Default::default()
}
}
pub fn update(&mut self, message: SoftDemoMessage) -> Command<SoftDemoMessage> {
match message {
SoftDemoMessage::Tick(_) | SoftDemoMessage::Next => {
self.queued_ticks = (self.queued_ticks + 1).min(self.speed);
if let Some(task) = self.simulation.tick(self.queued_ticks) {
if let Some(speed) = self.next_speed.take() {
self.speed = speed;
}
self.queued_ticks = 0;
let version = self.version;
return Command::perform(task, move |message| {
SoftDemoMessage::Simulation(message, version)
});
}
},
}
Command::none()
}
// skipped code
pub fn subscription(&self) -> Subscription<SoftDemoMessage> {
if self.is_playing {time::every(Duration::from_millis(1000 / self.speed as u64))
.map(|instant| SoftDemoMessage::Tick(instant))
} else {
Subscription::none()
}
}
}
// skipped code