wallet: construct app and carefully manage resources

bin/darkwallet/src/app.rs

@@ -0,0 +1,261 @@
+use async_lock::Mutex;
+use futures::{stream::FuturesUnordered, StreamExt};
+use std::{
+    sync::{mpsc, Arc, Weak},
+    thread,
+};
+
+use crate::{
+    chatapp,
+    error::{Error, Result},
+    expr::Op,
+    gfx2::{GraphicsEvent, RenderApiPtr},
+    prop::{Property, PropertySubType, PropertyType},
+    pubsub::PublisherPtr,
+    scene::{
+        MethodResponseFn, Pimpl, SceneGraph, SceneGraphPtr2, SceneNode, SceneNodeId, SceneNodeInfo,
+        SceneNodeType,
+    },
+};
+
+trait Stoppable {
+    async fn stop(self);
+}
+
+pub struct App {
+    sg: SceneGraphPtr2,
+    ex: Arc<smol::Executor<'static>>,
+    render_api: RenderApiPtr,
+    event_pub: PublisherPtr<GraphicsEvent>,
+}
+
+impl App {
+    pub fn new(
+        sg: SceneGraphPtr2,
+        ex: Arc<smol::Executor<'static>>,
+        render_api: RenderApiPtr,
+        event_pub: PublisherPtr<GraphicsEvent>,
+    ) -> Arc<Self> {
+        debug!("App::new()");
+        Arc::new(Self { sg, ex, render_api, event_pub })
+    }
+
+    pub async fn start(self: Arc<Self>) {
+        debug!("App::start()");
+        // Setup UI
+        let mut sg = self.sg.lock().await;
+
+        let window = sg.add_node("window", SceneNodeType::Window);
+
+        let mut prop = Property::new("screen_size", PropertyType::Float32, PropertySubType::Pixel);
+        prop.set_array_len(2);
+        // Window not yet initialized so we can't set these.
+        //prop.set_f32(0, screen_width);
+        //prop.set_f32(1, screen_height);
+        window.add_property(prop).unwrap();
+
+        let mut prop = Property::new("scale", PropertyType::Float32, PropertySubType::Pixel);
+        prop.set_defaults_f32(vec![1.]).unwrap();
+        window.add_property(prop).unwrap();
+
+        let window_id = window.id;
+
+        // Create Window
+        // Window::new(window, weak sg)
+        drop(sg);
+        let pimpl = Window::new(
+            self.sg.clone(),
+            window_id,
+            self.ex.clone(),
+            self.render_api.clone(),
+            self.event_pub.clone(),
+        )
+        .await;
+        // -> reads any props it needs
+        // -> starts procs
+        let mut sg = self.sg.lock().await;
+        let node = sg.get_node_mut(window_id).unwrap();
+        node.pimpl = pimpl;
+
+        sg.link(window_id, SceneGraph::ROOT_ID).unwrap();
+
+        self.make_me_a_schema_plox().await;
+
+        // Access drawable in window node and call draw()
+        self.trigger_redraw().await;
+
+        let node = sg.get_node(window_id).unwrap();
+        node.set_property_f32("scale", 2.).unwrap();
+    }
+
+    async fn make_me_a_schema_plox(&self) {
+        let mut sg = self.sg.lock().await;
+        let layer_node_id = chatapp::create_layer(&mut sg, "view");
+
+        // Customize our layer
+        let node = sg.get_node(layer_node_id).unwrap();
+        let prop = node.get_property("rect").unwrap();
+        prop.set_u32(0, 0).unwrap();
+        prop.set_u32(1, 0).unwrap();
+        let code = vec![Op::Float32ToUint32((Box::new(Op::LoadVar("sw".to_string()))))];
+        prop.set_expr(2, code).unwrap();
+        let code = vec![Op::Float32ToUint32((Box::new(Op::LoadVar("sh".to_string()))))];
+        prop.set_expr(3, code).unwrap();
+
+        // Setup the pimpl
+        let node_id = node.id;
+        drop(sg);
+        let pimpl = RenderLayer::new().await;
+        let mut sg = self.sg.lock().await;
+        let node = sg.get_node_mut(node_id).unwrap();
+        node.pimpl = pimpl;
+
+        let window_id = sg.lookup_node("/window").unwrap().id;
+        sg.link(node_id, window_id).unwrap();
+    }
+
+    async fn trigger_redraw(&self) {
+        let sg = self.sg.lock().await;
+        let window_node = sg.lookup_node("/window").expect("no window attached!");
+        match &window_node.pimpl {
+            Pimpl::Window(win) => win.draw().await,
+            _ => panic!("wrong pimpl"),
+        }
+    }
+
+    pub async fn stop(&self) {
+        // Go through event graph and call stop on everything
+        // Depth first
+        debug!("Stopping app...");
+    }
+}
+
+fn print_type_of<T>(_: &T) {
+    println!("{}", std::any::type_name::<T>())
+}
+
+pub type WindowPtr = Arc<Window>;
+
+pub struct Window {
+    sg: SceneGraphPtr2,
+    node_id: SceneNodeId,
+    render_api: RenderApiPtr,
+    resize_task: smol::Task<()>,
+    modify_task: smol::Task<()>,
+}
+
+impl Window {
+    pub async fn new(
+        sg: SceneGraphPtr2,
+        node_id: SceneNodeId,
+        ex: Arc<smol::Executor<'static>>,
+        render_api: RenderApiPtr,
+        event_pub: PublisherPtr<GraphicsEvent>,
+    ) -> Pimpl {
+        debug!("Window::new()");
+
+        let screen_size_prop = {
+            let sg = sg.lock().await;
+            let node = sg.get_node(node_id).unwrap();
+            node.get_property("screen_size").unwrap()
+        };
+
+        // Start a task monitoring for window resize events
+        // which updates screen_size
+        let ev_sub = event_pub.clone().subscribe();
+        let screen_size_prop2 = screen_size_prop.clone();
+        let resize_task = ex.spawn(async move {
+            loop {
+                let Ok(ev) = ev_sub.receive().await else {
+                    debug!("Event relayer closed");
+                    break
+                };
+                let (w, h) = match ev {
+                    GraphicsEvent::Resize((w, h)) => (w, h),
+                    _ => continue,
+                };
+
+                // Now update the properties
+                screen_size_prop2.set_f32(0, w);
+                screen_size_prop2.set_f32(1, h);
+            }
+        });
+
+        // Monitor for changes to screen_size or scale properties
+        // If so then trigger draw
+        let scale_sub = {
+            let sg = sg.lock().await;
+            let node = sg.get_node(node_id).unwrap();
+            let prop = node.get_property("scale").unwrap();
+            prop.subscribe_modify()
+        };
+        let screen_size_sub = screen_size_prop.subscribe_modify();
+
+        let self_ = Arc::new_cyclic(|me: &Weak<Self>| {
+            // Modify task needs a Weak<Self>
+            let me2 = me.clone();
+            let modify_task = ex.spawn(async move {
+                loop {
+                    let mut futures = FuturesUnordered::new();
+                    futures.push(scale_sub.receive());
+                    futures.push(screen_size_sub.receive());
+
+                    while let Some(ev) = futures.next().await {
+                        let Ok(_) = ev else {
+                            debug!("prop sub closed");
+                            break
+                        };
+
+                        let Some(self_) = me2.upgrade() else {
+                            // Should not happen
+                            panic!("self destroyed before modify_task was stopped!");
+                        };
+
+                        self_.draw().await;
+                    }
+                }
+            });
+
+            Self { sg, node_id, render_api, resize_task, modify_task }
+        });
+
+        Pimpl::Window(self_)
+    }
+
+    async fn draw(&self) {
+        // This should remain locked for the entire draw
+        let sg = self.sg.lock().await;
+        let self_node = sg.get_node(self.node_id).unwrap();
+
+        for child_inf in self_node.get_children2() {
+            let node = sg.get_node(child_inf.id).unwrap();
+
+            let sg_ref: &SceneGraph = &sg;
+            match &node.pimpl {
+                //Pimpl::RenderLayer(layer) => layer.draw(sg_ref).await,
+                _ => error!("unhandled pimpl type"),
+            }
+        }
+    }
+}
+
+// Nodes should be stopped before being removed
+impl Stoppable for Window {
+    async fn stop(self) {
+        self.resize_task.cancel().await;
+    }
+}
+
+pub type RenderLayerPtr = Arc<RenderLayer>;
+
+pub struct RenderLayer {}
+
+impl RenderLayer {
+    pub async fn new() -> Pimpl {
+        let self_ = Arc::new(Self {});
+
+        Pimpl::RenderLayer(self_)
+    }
+
+    pub async fn draw(&self, sg: &SceneGraph) {}
+}

bin/darkwallet/src/chatapp.rs

@@ -94,7 +94,7 @@ impl Buffer {
     }
 }
 
-fn create_layer(sg: &mut SceneGraph, name: &str) -> SceneNodeId {
+pub fn create_layer(sg: &mut SceneGraph, name: &str) -> SceneNodeId {
     let win_id = sg.lookup_node("/window").unwrap().id;
 
     let node = sg.add_node(name, SceneNodeType::RenderLayer);
@@ -106,9 +106,7 @@ fn create_layer(sg: &mut SceneGraph, name: &str) -> SceneNodeId {
     prop.allow_exprs();
     node.add_property(prop).unwrap();
 
-    let node_id = node.id;
-    sg.link(node_id, win_id).unwrap();
-    node_id
+    node.id
 }
 
 fn create_mesh(sg: &mut SceneGraph, name: &str, layer_node_id: SceneNodeId) -> SceneNodeId {

bin/darkwallet/src/error.rs

@@ -101,4 +101,7 @@ pub enum Error {
 
     #[error("Graphics window closed")]
     GfxWindowClosed = 33,
+
+    #[error("Publisher was destroyed")]
+    PublisherDestroyed = 34,
 }

bin/darkwallet/src/gfx2.rs

@@ -23,7 +23,7 @@ use crate::{
     gfx::Rectangle,
     keysym::{KeyCodeAsStr, MouseButtonAsU8},
     prop::{Property, PropertySubType, PropertyType},
-    pubsub::Publisher,
+    pubsub::PublisherPtr,
     res::{ResourceId, ResourceManager},
     scene::{
         MethodResponseFn, Pimpl, SceneGraph, SceneGraphPtr, SceneNode, SceneNodeId, SceneNodeInfo,
@@ -40,13 +40,15 @@ pub struct Vertex {
     pub uv: [f32; 2],
 }
 
+pub type RenderApiPtr = Arc<RenderApi>;
+
 pub struct RenderApi {
-    method_sendr: mpsc::Sender<GraphicsMethod>,
+    method_req: mpsc::Sender<GraphicsMethod>,
 }
 
 impl RenderApi {
-    pub fn new(method_sendr: mpsc::Sender<GraphicsMethod>) -> Arc<Self> {
-        Arc::new(Self { method_sendr })
+    pub fn new(method_req: mpsc::Sender<GraphicsMethod>) -> Arc<Self> {
+        Arc::new(Self { method_req })
     }
 
     async fn new_texture(&self, width: u16, height: u16, data: Vec<u8>) -> Result<TextureId> {
@@ -54,7 +56,7 @@ impl RenderApi {
 
         let method = GraphicsMethod::NewTexture((width, height, data, sendr));
 
-        self.method_sendr.send(method).map_err(|_| Error::GfxWindowClosed)?;
+        self.method_req.send(method).map_err(|_| Error::GfxWindowClosed)?;
 
         let texture_id = recvr.recv().await.map_err(|_| Error::GfxWindowClosed)?;
         Ok(texture_id)
@@ -64,7 +66,7 @@ impl RenderApi {
         let method = GraphicsMethod::DeleteTexture(texture);
 
         // Ignore any error
-        let _ = self.method_sendr.send(method);
+        let _ = self.method_req.send(method);
     }
 
     pub async fn new_vertex_buffer(&self, verts: Vec<Vertex>) -> Result<BufferId> {
@@ -72,7 +74,7 @@ impl RenderApi {
 
         let method = GraphicsMethod::NewVertexBuffer((verts, sendr));
 
-        self.method_sendr.send(method).map_err(|_| Error::GfxWindowClosed)?;
+        self.method_req.send(method).map_err(|_| Error::GfxWindowClosed)?;
 
         let buffer = recvr.recv().await.map_err(|_| Error::GfxWindowClosed)?;
         Ok(buffer)
@@ -83,7 +85,7 @@ impl RenderApi {
 
         let method = GraphicsMethod::NewIndexBuffer((indices, sendr));
 
-        self.method_sendr.send(method).map_err(|_| Error::GfxWindowClosed)?;
+        self.method_req.send(method).map_err(|_| Error::GfxWindowClosed)?;
 
         let buffer = recvr.recv().await.map_err(|_| Error::GfxWindowClosed)?;
         Ok(buffer)
@@ -93,14 +95,14 @@ impl RenderApi {
         let method = GraphicsMethod::DeleteBuffer(buffer);
 
         // Ignore any error
-        let _ = self.method_sendr.send(method);
+        let _ = self.method_req.send(method);
     }
 
     pub async fn replace_draw_call(&self, loc: Vec<usize>, draw_call: DrawCall) {
         let method = GraphicsMethod::ReplaceDrawCall((loc, draw_call));
 
         // Ignore any error
-        let _ = self.method_sendr.send(method);
+        let _ = self.method_req.send(method);
     }
 }
 
@@ -205,14 +207,14 @@ struct Stage {
     root_dc: DrawCall,
     last_draw_time: Option<Instant>,
 
-    method_recvr: mpsc::Receiver<GraphicsMethod>,
-    event_pub: async_channel::Sender<GraphicsEvent>,
+    method_rep: mpsc::Receiver<GraphicsMethod>,
+    event_pub: PublisherPtr<GraphicsEvent>,
 }
 
 impl Stage {
     pub fn new(
-        method_recvr: mpsc::Receiver<GraphicsMethod>,
-        event_pub: async_channel::Sender<GraphicsEvent>,
+        method_rep: mpsc::Receiver<GraphicsMethod>,
+        event_pub: PublisherPtr<GraphicsEvent>,
     ) -> Self {
         let mut ctx: Box<dyn RenderingBackend> = window::new_rendering_backend();
 
@@ -261,7 +263,7 @@ impl Stage {
             white_texture,
             root_dc: DrawCall { instrs: vec![], dcs: vec![] },
             last_draw_time: None,
-            method_recvr,
+            method_rep,
             event_pub,
         }
     }
@@ -336,7 +338,7 @@ impl EventHandler for Stage {
         let deadline = Instant::now() + allowed_time;
 
         loop {
-            let Ok(method) = self.method_recvr.recv_deadline(deadline) else { break };
+            let Ok(method) = self.method_rep.recv_deadline(deadline) else { break };
             match method {
                 GraphicsMethod::NewTexture((width, height, data, sendr)) => {
                     self.method_new_texture(width, height, data, sendr)
@@ -387,18 +389,15 @@ impl EventHandler for Stage {
 
     fn key_down_event(&mut self, keycode: KeyCode, mods: KeyMods, repeat: bool) {
         let event = GraphicsEvent::KeyDown((keycode, mods, repeat));
-        self.event_pub.try_send(event).unwrap();
+        self.event_pub.notify(event);
     }
     fn resize_event(&mut self, width: f32, height: f32) {
         let event = GraphicsEvent::Resize((width, height));
-        self.event_pub.try_send(event).unwrap();
+        self.event_pub.notify(event);
     }
 }
 
-pub fn run_gui(
-    method_recvr: mpsc::Receiver<GraphicsMethod>,
-    event_pub: async_channel::Sender<GraphicsEvent>,
-) {
+pub fn run_gui(method_rep: mpsc::Receiver<GraphicsMethod>, event_pub: PublisherPtr<GraphicsEvent>) {
     #[cfg(target_os = "android")]
     {
         android_logger::init_once(
@@ -431,5 +430,5 @@ pub fn run_gui(
     conf.platform.apple_gfx_api =
         if metal { conf::AppleGfxApi::Metal } else { conf::AppleGfxApi::OpenGl };
 
-    miniquad::start(conf, || Box::new(Stage::new(method_recvr, event_pub)));
+    miniquad::start(conf, || Box::new(Stage::new(method_rep, event_pub)));
 }

bin/darkwallet/src/main.rs

@@ -2,54 +2,41 @@
 #![feature(str_split_whitespace_remainder)]
 
 use async_lock::Mutex;
+use futures::{stream::FuturesUnordered, StreamExt};
 use std::{
     sync::{mpsc, Arc},
     thread,
 };
 
-mod chatapp;
+#[macro_use]
+extern crate log;
+#[allow(unused_imports)]
+use log::LevelFilter;
 
+mod app;
+mod chatapp;
 mod chatview;
-
 mod editbox;
-
 mod error;
-
 mod expr;
-
 mod gfx;
-use gfx::run_gui;
-
 mod gfx2;
-
 mod keysym;
-
 mod net;
-use net::ZeroMQAdapter;
-
-mod scene;
-use scene::{SceneGraph, SceneGraphPtr};
-
 mod plugin;
-
 mod prop;
-
 mod pubsub;
-
 mod py;
-
 mod res;
-
+mod scene;
 mod shader;
-
 mod text;
 
-use crate::error::{Error, Result};
-
-#[macro_use]
-extern crate log;
-#[allow(unused_imports)]
-use log::LevelFilter;
+use crate::{
+    error::{Error, Result},
+    net::ZeroMQAdapter,
+    scene::{SceneGraph, SceneGraphPtr},
+};
 
 fn start_zmq(scene_graph: SceneGraphPtr) {
     // detach thread
@@ -167,50 +154,62 @@ async fn amain(ex: Arc<smol::Executor<'static>>, render_api: Arc<gfx2::RenderApi
 */
 
 fn main() {
-    let ex = std::sync::Arc::new(smol::Executor::new());
-    let scene_graph = Arc::new(Mutex::new(SceneGraph::new()));
+    // [x] event pub should be a Publisher
+    // [ ] properties should have post-modify hook used to redraw widgets
+
+    let ex = Arc::new(smol::Executor::new());
+    let sg = Arc::new(Mutex::new(SceneGraph::new()));
 
-    let scene_graph2 = scene_graph.clone();
+    let sg2 = sg.clone();
     let ex2 = ex.clone();
     let zmq_task = ex.spawn(async {
-        let mut zmq_rpc = ZeroMQAdapter::new(scene_graph2, ex2).await;
+        let mut zmq_rpc = ZeroMQAdapter::new(sg2, ex2).await;
         zmq_rpc.run().await;
     });
 
-    let (method_sender, method_recvr) = mpsc::channel();
-    let render_api = gfx2::RenderApi::new(method_sender);
+    let (method_req, method_rep) = mpsc::channel();
+    let render_api = gfx2::RenderApi::new(method_req);
+    let event_pub = pubsub::Publisher::new();
 
-    let (event_pub, event_sub) = async_channel::unbounded();
+    let app = app::App::new(sg.clone(), ex.clone(), render_api.clone(), event_pub.clone());
+    let app_task = ex.spawn(app.clone().start());
 
+    // Nice to see which events exist
+    let ev_sub = event_pub.clone().subscribe();
     let ev_relay_task = ex.spawn(async move {
         loop {
-            let Ok(ev) = event_sub.recv().await else {
+            let Ok(ev) = ev_sub.receive().await else {
                 debug!("Event relayer closed");
                 break
             };
             debug!("event: {:?}", ev);
         }
     });
+    // End debug code
 
     let n_threads = std::thread::available_parallelism().unwrap().get();
     let (signal, shutdown) = smol::channel::unbounded::<()>();
     let exec_threadpool = thread::spawn(move || {
         easy_parallel::Parallel::new()
-            // Executor threads
-            .each(1..n_threads, |_| smol::future::block_on(ex.run(shutdown.recv())))
-            // Run the main future on this thread
-            .finish(|| smol::future::block_on(ex.run(shutdown.recv())));
+            // N executor threads
+            .each(0..n_threads, |_| smol::future::block_on(ex.run(shutdown.recv())))
+            .run();
     });
 
-    gfx2::run_gui(method_recvr, event_pub);
+    gfx2::run_gui(method_rep, event_pub);
 
     // Close all tasks
     smol::future::block_on(async {
         // Perform cleanup code
         // If not finished in certain amount of time, then just exit
 
-        zmq_task.cancel().await;
-        ev_relay_task.cancel().await;
+        let mut futures = FuturesUnordered::new();
+        futures.push(zmq_task.cancel());
+        futures.push(ev_relay_task.cancel());
+        futures.push(app_task.cancel());
+        let _: Vec<_> = futures.collect().await;
+
+        app.stop().await;
     });
 
     drop(signal);

bin/darkwallet/src/prop/mod.rs

@@ -11,7 +11,11 @@ use std::{
     },
 };
 
-use crate::{expr::SExprCode, scene::SceneNodeId};
+use crate::{
+    expr::SExprCode,
+    pubsub::{Publisher, PublisherPtr, Subscription},
+    scene::SceneNodeId,
+};
 
 mod wrap;
 pub use wrap::{PropertyBool, PropertyColor, PropertyFloat32, PropertyStr, PropertyUint32};
@@ -177,6 +181,13 @@ impl Encodable for PropertyValue {
     }
 }
 
+#[derive(Debug, Clone)]
+pub enum ModifyAction {
+    Clear,
+    Set(usize),
+    Push,
+}
+
 pub struct Property {
     pub name: String,
     pub typ: PropertyType,
@@ -197,6 +208,8 @@ pub struct Property {
 
     // PropertyType must be Enum
     pub enum_items: Option<Vec<String>>,
+
+    on_modify: PublisherPtr<ModifyAction>,
 }
 
 impl Property {
@@ -219,6 +232,8 @@ impl Property {
             min_val: None,
             max_val: None,
             enum_items: None,
+
+            on_modify: Publisher::new(),
         }
     }
 
@@ -286,15 +301,16 @@ impl Property {
         Ok(())
     }
 
+    // Set
+
     /// This will clear all values, resetting them to the default
     pub fn clear_values(&self) {
         let vals = &mut self.vals.lock().unwrap();
         vals.clear();
         vals.resize(self.array_len, PropertyValue::Unset);
+        self.on_modify.notify(ModifyAction::Clear);
     }
 
-    // Set
-
     fn set_raw_value(&self, i: usize, val: PropertyValue) -> Result<()> {
         if self.typ != val.as_type() {
             return Err(Error::PropertyWrongType)
@@ -305,6 +321,7 @@ impl Property {
             return Err(Error::PropertyWrongIndex)
         }
         vals[i] = val;
+        self.on_modify.notify(ModifyAction::Set(i));
         Ok(())
     }
 
@@ -314,6 +331,7 @@ impl Property {
             return Err(Error::PropertyWrongIndex)
         }
         vals[i] = PropertyValue::Unset;
+        self.on_modify.notify(ModifyAction::Set(i));
         Ok(())
     }
 
@@ -326,6 +344,7 @@ impl Property {
             return Err(Error::PropertyWrongIndex)
         }
         vals[i] = PropertyValue::Null;
+        self.on_modify.notify(ModifyAction::Set(i));
         Ok(())
     }
 
@@ -390,6 +409,7 @@ impl Property {
             return Err(Error::PropertyWrongIndex)
         }
         vals[i] = PropertyValue::SExpr(Arc::new(val));
+        self.on_modify.notify(ModifyAction::Set(i));
         Ok(())
     }
 
@@ -402,6 +422,7 @@ impl Property {
         let vals = &mut self.vals.lock().unwrap();
         let i = vals.len();
         vals.push(PropertyValue::Null);
+        self.on_modify.notify(ModifyAction::Push);
         Ok(i)
     }
 
@@ -566,6 +587,12 @@ impl Property {
     pub fn get_expr(&self, i: usize) -> Result<Arc<SExprCode>> {
         self.get_value(i)?.as_sexpr()
     }
+
+    // Subs
+
+    pub fn subscribe_modify(&self) -> Subscription<ModifyAction> {
+        self.on_modify.clone().subscribe()
+    }
 }
 
 #[cfg(test)]

bin/darkwallet/src/pubsub.rs

@@ -4,45 +4,48 @@ use std::{
     sync::{Arc, Mutex},
 };
 
+use crate::error::{Error, Result};
+
 pub type SubscriptionId = usize;
 
+// Waiting for trait aliases
+trait Piped: Clone + Send + 'static {}
+impl<T> Piped for T where T: Clone + Send + 'static {}
+
 #[derive(Debug)]
 /// Subscription to the Publisher. Created using `publisher.subscribe().await`.
-pub struct Subscription<T> {
+pub struct Subscription<T: Piped> {
     id: SubscriptionId,
     recv_queue: smol::channel::Receiver<T>,
     parent: Arc<Publisher<T>>,
 }
 
-impl<T: Clone + Send + 'static> Subscription<T> {
+impl<T: Piped> Subscription<T> {
     pub fn get_id(&self) -> SubscriptionId {
         self.id
     }
 
     /// Receive message.
-    pub async fn receive(&self) -> T {
-        let message_result = self.recv_queue.recv().await;
-
-        match message_result {
-            Ok(message_result) => message_result,
-            Err(err) => {
-                panic!("Subscription::receive() recv_queue failed! {}", err);
-            }
-        }
+    pub async fn receive(&self) -> Result<T> {
+        let msg_result = self.recv_queue.recv().await;
+        msg_result.or(Err(Error::PublisherDestroyed))
     }
+}
 
-    /// Must be called manually since async Drop is not possible in Rust
-    pub fn unsubscribe(&self) {
-        self.parent.clone().unsubscribe(self.id)
+impl<T: Piped> Drop for Subscription<T> {
+    fn drop(&mut self) {
+        self.parent.unsubscribe(self.id)
     }
 }
 
+pub type PublisherPtr<T> = Arc<Publisher<T>>;
+
 #[derive(Debug)]
 pub struct Publisher<T> {
     subs: Mutex<HashMap<SubscriptionId, smol::channel::Sender<T>>>,
 }
 
-impl<T: Clone + Send + 'static> Publisher<T> {
+impl<T: Piped> Publisher<T> {
     pub fn new() -> Arc<Self> {
         Arc::new(Self { subs: Mutex::new(HashMap::new()) })
     }
@@ -57,28 +60,16 @@ impl<T: Clone + Send + 'static> Publisher<T> {
         Subscription { id: sub_id, recv_queue: recvr, parent: self.clone() }
     }
 
-    fn unsubscribe(self: Arc<Self>, sub_id: SubscriptionId) {
+    fn unsubscribe(&self, sub_id: SubscriptionId) {
         self.subs.lock().unwrap().remove(&sub_id);
     }
 
     /// Publish a message to all listening subscriptions.
-    pub fn notify_sync(&self, message_result: T) {
-        self.notify_with_exclude_sync(message_result, &[])
-    }
-
-    /// Publish a message to all listening subscriptions but exclude some subset.
-    /// Sync version.
-    pub fn notify_with_exclude_sync(&self, message_result: T, exclude_list: &[SubscriptionId]) {
+    pub fn notify(&self, msg: T) {
         for (id, sub) in self.subs.lock().unwrap().iter() {
-            if exclude_list.contains(id) {
-                continue
-            }
-
-            if let Err(e) = sub.try_send(message_result.clone()) {
-                warn!(
-                    target: "system::publisher",
-                    "[system::publisher] Error returned sending message in notify_with_exclude_sync() call! {}", e,
-                );
+            if let Err(e) = sub.try_send(msg.clone()) {
+                // This should never happen since Drop calls unsubscribe()
+                panic!("Error in notify() call for sub={}! {}", id, e);
             }
         }
     }

bin/darkwallet/src/scene.rs

@@ -13,7 +13,7 @@ use std::{
 };
 
 use crate::{
-    chatview, editbox,
+    app, chatview, editbox,
     error::{Error, Result},
     prop::{Property, PropertyType},
 };
@@ -408,6 +408,9 @@ impl SceneNode {
             .filter(move |child_inf| allowed_types.contains(&child_inf.typ))
             .collect()
     }
+    pub fn get_children2(&self) -> Vec<SceneNodeInfo> {
+        self.children.iter().cloned().collect()
+    }
 
     pub fn add_property(&mut self, prop: Property) -> Result<()> {
         if self.has_property(&prop.name) {
@@ -634,4 +637,6 @@ pub enum Pimpl {
     Null,
     EditBox(editbox::EditBoxPtr),
     ChatView(chatview::ChatViewPtr),
+    Window(app::WindowPtr),
+    RenderLayer(app::RenderLayerPtr),
 }