no need for lifetimes if you have owned sync primitives :)

speaker_actor.rs

use eyre::Result;
use sonor::Speaker;
use std::sync::{Arc};
use tokio::sync::Mutex;
use std::thread;
use tokio::sync::mpsc;
use tokio::time::Duration;

pub struct SpeakerActor {
    receiver: mpsc::UnboundedReceiver<SpeakerMessage>,
    sender: mpsc::UnboundedSender<SpeakerMessage>,
    state: Arc<Mutex<State>>,
}

pub struct State {
    remaining_steps: i32,
    direction: i16,
    speaker: Speaker,
}

pub enum SpeakerMessage {
    VolumeUp,
    VolumeDown,
    VolumeStop,
    Play,
    Pause,
    PlayPause,
    Tick,
}

impl SpeakerActor {
    fn new(
        receiver: mpsc::UnboundedReceiver<SpeakerMessage>,
        sender: mpsc::UnboundedSender<SpeakerMessage>,
        speaker: Speaker,
    ) -> Self {
        SpeakerActor {
            receiver,
            sender,
            state: Arc::new(Mutex::new(State {
                speaker,
                remaining_steps: 0,
                direction: 1,
            })),
        }
    }

    async fn move_volume(&self, direction: i16) -> Result<()> {
        let mut state = self.state.lock().await;
        let running = state.remaining_steps > 0;
        state.remaining_steps = 5;
        state.direction = direction;
        if !running {
            let state_mutex = self.state.clone();
            tokio::spawn(async move {
                let mut state = state_mutex.lock().await;
                while state.remaining_steps > 0 {
                    state.remaining_steps = state.remaining_steps - 1;
                    thread::sleep(Duration::from_millis(250));
                    let state_mutex = state_mutex.clone();
                    // interval.tick().await;
                    tokio::spawn(async move {
                        let state = state_mutex.lock().await;
                        state.speaker.set_volume_relative(state.direction).await.unwrap();
                    });
                    println!("Volume +{:?}", state.direction);
                }
            });
        }
        Ok(())
    }

    async fn handle_message(&mut self, msg: SpeakerMessage) -> Result<()> {
        match msg {
            SpeakerMessage::VolumeUp => {
                self.move_volume(1).await?;
                Ok(())
            }
            SpeakerMessage::VolumeDown => {
                self.move_volume(-1).await?;
                Ok(())
            }
            SpeakerMessage::VolumeStop => {
                let mut state = self.state.lock().await;
                state.remaining_steps = 0;
                Ok(())
            }
            SpeakerMessage::Play => Ok(()),
            SpeakerMessage::Pause => Ok(()),
            SpeakerMessage::PlayPause => Ok(()),
            SpeakerMessage::Tick => Ok(()),
        }
    }

    async fn run(&mut self) {
        while let Some(msg) = self.receiver.recv().await {
            let task = self.handle_message(msg);
            /*
                        tokio::spawn(async move {
            println!("{:?}", err);
                        })
                        */
        }
    }
}

#[derive(Clone)]
pub struct SpeakerHandle {
    sender: mpsc::UnboundedSender<SpeakerMessage>,
}

impl SpeakerHandle {
    pub fn new(speaker: Speaker) -> Self {
        let (sender, receiver) = mpsc::unbounded_channel();
        let self_sender = sender.clone();
        tokio::spawn(async move {
            let mut actor = SpeakerActor::new(receiver, self_sender, speaker);
            actor.run().await;
        });

        Self { sender }
    }

    pub async fn send(&self, msg: SpeakerMessage) {
        println!("before");
        let _ = self.sender.send(msg);
        println!("after");
    }
}