Propagate tile stream errors further before unwrapping

src/config.rs

@@ -1,4 +1,5 @@
 use crate::tiles;
+use crate::tiles::TileResult;
 use dbus::nonblock::SyncConnection;
 use futures::{stream::BoxStream, Stream};
 use serde::{Deserialize, Deserializer};
@@ -11,7 +12,6 @@ use structopt::StructOpt;
 use tokio::fs::File;
 use tokio::prelude::*;
 use tokio::time::{self, Duration};
-use crate::tiles::TileResult;
 
 #[derive(Deserialize, Clone, Debug, Default)]
 #[serde(default)]
@@ -119,17 +119,14 @@ pub fn process_tile(
     let five_secs = Duration::from_secs(5);
     match &tile.config_type {
         TileConfigType::Battery => wrap(tiles::battery_stream(), tile.update.or(Some(five_secs))),
-        TileConfigType::Hostname => wrap(tiles::hostname_stream(connection.clone()), tile.update),
+        TileConfigType::Hostname => wrap(tiles::hostname_stream(connection.as_ref()), tile.update),
         TileConfigType::Load => wrap(tiles::load_stream(), tile.update.or(Some(five_secs))),
         TileConfigType::Memory => wrap(tiles::memory_stream(), tile.update.or(Some(five_secs))),
         TileConfigType::Time(c) => wrap(tiles::time_stream(c.clone()), tile.update),
     }
 }
 
-fn wrap<'a, S>(
+fn wrap<'a, S>(stream: S, duration: Option<Duration>) -> BoxStream<'a, TileResult>
-    stream: S,
-    duration: Option<Duration>,
-) -> BoxStream<'a, TileResult>
 where
     S: Stream<Item = TileResult> + Send + 'a,
 {

src/main.rs

@@ -45,20 +45,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 fn spawn_stream<E: 'static>(
     index: usize,
     stream: BoxStream<'static, Result<tile::Block, E>>,
-    sender: Sender<tile::TileData>,
+    sender: Sender<Result<tile::TileData, E>>,
 ) where
-    E: Debug,
+    E: Debug + Send,
 {
     tokio::spawn(async move {
         let instance: Arc<str> = Uuid::new_v4().to_string().into();
-        let stream = stream.map(|block| {
+        let stream = stream.map(|block: Result<_, _>| {
-            Ok(tile::TileData {
+            Ok(block.map(|block| tile::TileData {
                 block: tile::Block {
                     instance: instance.clone(),
-                    ..block.unwrap()
+                    ..block
                 },
                 sender_id: index,
-            })
+            }))
         });
         let future = stream.forward(sender);
         future.await

src/output.rs

@@ -5,11 +5,14 @@ use futures::StreamExt;
 use std::convert::Infallible;
 use tokio::io::{self, AsyncWriteExt};
 
-pub async fn launch(
+pub async fn launch<E>(
     num_tiles: usize,
-    mut receiver: Receiver<TileData>,
+    mut receiver: Receiver<Result<TileData, E>>,
     _default: DefaultSection,
-) -> io::Result<Infallible> {
+) -> io::Result<Infallible>
+where
+    E: Send + std::fmt::Debug,
+{
     let mut stdout = io::stdout();
     stdout.write_all(b"{ \"version\": 1 }\n[").await?;
 
@@ -17,6 +20,7 @@ pub async fn launch(
     blocks.resize_with(num_tiles, Default::default);
     loop {
         let message = receiver.next().await.unwrap();
+        let message = message.unwrap();
         if message.sender_id < num_tiles {
             blocks[message.sender_id] = Some(message.block);
         } else {

src/tiles/hostname.rs

@@ -3,10 +3,9 @@ use crate::tiles::TileResult;
 use dbus::nonblock::stdintf::org_freedesktop_dbus::Properties;
 use dbus::nonblock::{Proxy, SyncConnection};
 use futures::{FutureExt, Stream};
-use std::sync::Arc;
 use std::time::Duration;
 
-pub fn hostname_stream(connection: Arc<SyncConnection>) -> impl Stream<Item = TileResult> {
+pub fn hostname_stream(connection: &SyncConnection) -> impl Stream<Item = TileResult> {
     let proxy = Proxy::new(
         "org.freedesktop.hostname1",
         "/org/freedesktop/hostname1",