Update to tokio v1, warp v0.3

Cargo.toml

@@ -10,8 +10,9 @@ edition = "2018"
 futures-util = "0.3"
 log = "0.4.11"
 pretty_env_logger = "0.4.0"
-reqwest = { version = "0.10", default-features = false }
+reqwest = { version = "0.11", default-features = false }
 rppal = { git = "https://github.com/golemparts/rppal/", rev = "2e980caf76756c97bb0b18fb3ab08fb51ed1f90e" }
 smallvec = "1.4.2"
-tokio = { version = "0.2", features = ["macros"] }
-warp = "0.2.5"
+tokio = { version = "1", features = ["macros"] }
+tokio-stream = "0.1.2"
+warp = "0.3"

src/main.rs

@@ -9,8 +9,9 @@ use futures_util::{FutureExt, select, stream, StreamExt};
 use rppal::gpio::{Gpio, Trigger};
 use smallvec::SmallVec;
 use tokio::{ sync::mpsc,
-             time::{delay_for, interval} };
-use warp::{Filter, sse, sse::ServerSentEvent};
+             time::{interval, sleep} };
+use tokio_stream::wrappers::{IntervalStream, ReceiverStream};
+use warp::{Filter, sse};
 
 const HUE_ADDRESS: &str = "philips-hue.local";
 const OUTDOOR_CHIME_FILE: &str = "static/outdoor.mp3";
@@ -19,7 +20,7 @@ const HUE_KEY: &str = include_str!("../hue.key");
 const BUTTON_PIN: u8 = 26;
 const CHANNEL_VEC_SIZE: usize = 32;
 
-#[tokio::main(basic_scheduler)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() {
   pretty_env_logger::init_timed();
 
@@ -38,7 +39,7 @@ async fn main() {
   let reqwest = &reqwest::Client::new();
   let audio_child = &RefCell::new(None);
   let hue_busy = &Cell::new(false);
-  let pushes = rx.for_each_concurrent(2, |()| async move {
+  let pushes = ReceiverStream::new(rx).for_each_concurrent(2, |()| async move {
     if !audio_busy(&audio_child) {
       play_chime(&audio_child)
     } else { debug!("doorbell still ringing, not playing new chime"); }
@@ -82,14 +83,14 @@ async fn warp(clients: Arc<Mutex<SmallVec<[mpsc::Sender<()>; CHANNEL_VEC_SIZE]>>
       clients.lock().unwrap().push(tx);
 
       let stream = stream::select(
-        interval(Duration::from_secs(3)).map(move |_| {
+        IntervalStream::new(interval(Duration::from_secs(3))).map(move |_| {
           trace!("sending sse keepalive ping");
-          Ok::<_, Infallible>((sse::event("ping"), sse::data("")).into_a())
+          Ok::<_, Infallible>(sse::Event::default().event("ping").data(""))
         }),
 
-        rx.map(|()| {
+        ReceiverStream::new(rx).map(|()| {
           debug!("sending ring sse");
-          Ok::<_, Infallible>((sse::event("ring"), sse::data("")).into_b())
+          Ok::<_, Infallible>(sse::Event::default().event("ring").data(""))
         })
       );
       sse::reply(stream)
@@ -129,7 +130,7 @@ fn play_chime(audio_child: &RefCell<Option<process::Child>>) {
 async fn hue_base(reqwest: &reqwest::Client, body: &'static str, delay_millis: u64) {
   let _ = reqwest.put(&format!("http://{}/api/{}/lights/10/state", HUE_ADDRESS, HUE_KEY))
                  .body(body).send().await;
-  delay_for(Duration::from_millis(delay_millis)).await;
+  sleep(Duration::from_millis(delay_millis)).await;
 }
 
 async fn flash_porch(reqwest: &reqwest::Client) {
@@ -147,7 +148,7 @@ async fn flash_porch(reqwest: &reqwest::Client) {
              r#"{"transitiontime":0,"hue":56228}"#,  // purple
              250).await;
   }
-  delay_for(Duration::from_millis(250)).await;
+  sleep(Duration::from_millis(250)).await;
   hue_base(reqwest,
            r#"{"transitiontime":20,"bri":20}"#,  // fade
            2_250).await;