diff --git a/Cargo.toml b/Cargo.toml
index 05979c3..3035bd9 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -3,7 +3,7 @@ name = "xactor"
 version = "0.7.11"
 authors = ["sunli <scott_s829@163.com>"]
 description = "Xactor is a rust actors framework based on async-std"
-edition = "2018"
+edition = "2021"
 publish = true
 license = "MIT"
 documentation = "https://docs.rs/xactor/"
@@ -14,16 +14,17 @@ categories = ["network-programming", "asynchronous"]
 readme = "README.md"
 
 [dependencies]
-futures = "0.3.8"
-async-trait = "0.1.42"
-async-std = { version = "1.8.0", features = ["attributes"], optional = true }
-tokio = { version = "1.0.1", features = ["rt-multi-thread", "macros", "time"], optional = true }
-once_cell = "1.5.2"
-xactor-derive = { path = "xactor-derive", version = "0.7"}
+futures = "0.3.21"
+async-trait = "0.1.52"
+async-std = { version = "1.10.0", features = ["attributes"], optional = true }
+tokio = { version = "1.17.0", features = ["rt-multi-thread", "macros", "time"], optional = true }
+once_cell = "1.9.0"
+xactor-derive = { path = "xactor-derive", version = "0.7" }
 fnv = "1.0.7"
-slab = "0.4.2"
-anyhow = { version = "1.0.37", optional = true }
-eyre = { version = "0.6.5", optional = true }
+slab = "0.4.5"
+anyhow = { version = "1.0.53", optional = true }
+eyre = { version = "0.6.6", optional = true }
+dyn-clone = "1.0.4"
 
 [workspace]
 members = [
diff --git a/examples/intervals.rs b/examples/intervals.rs
new file mode 100644
index 0000000..f87cd31
--- /dev/null
+++ b/examples/intervals.rs
@@ -0,0 +1,31 @@
+use std::time::Duration;
+use xactor::*;
+
+#[message]
+struct IntervalMsg;
+
+struct MyActor;
+
+#[async_trait::async_trait]
+impl Actor for MyActor {
+    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
+        // Send the IntervalMsg message 3 seconds later
+        ctx.send_later(IntervalMsg, Duration::from_millis(500));
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler<IntervalMsg> for MyActor {
+    async fn handle(&mut self, ctx: &mut Context<Self>, _msg: IntervalMsg) {
+        ctx.send_later(IntervalMsg, Duration::from_millis(500));
+    }
+}
+
+#[xactor::main]
+async fn main() -> Result<()> {
+    // Exit the program after 3 seconds
+    let addr = MyActor.start().await?;
+    addr.wait_for_stop().await;
+    Ok(())
+}
diff --git a/examples/subscriber.rs b/examples/subscriber.rs
index 7f4fda1..0b15adf 100644
--- a/examples/subscriber.rs
+++ b/examples/subscriber.rs
@@ -95,9 +95,21 @@ impl Actor for Subscriber {
         // Send subscription request message to the Message Producer
         println!("Child Subscriber Started - id {:?}", self.id);
         let self_sender = ctx.address().sender();
+
         let _ = self.message_producer_addr.send(SubscribeToProducer {
-            sender: self_sender,
+            sender: self_sender.clone(),
         });
+
+        let _ = self.message_producer_addr.send(SubscribeToProducer {
+            sender: self_sender.clone(),
+        });
+        let _ = self.message_producer_addr.send(SubscribeToProducer {
+            sender: self_sender.clone(),
+        });
+        let _ = self.message_producer_addr.send(SubscribeToProducer {
+            sender: self_sender.clone(),
+        });
+
         Ok(())
     }
 }
@@ -169,7 +181,7 @@ impl Handler<Broadcast> for MessageProducer {
 #[async_trait::async_trait]
 impl Handler<SubscribeToProducer> for MessageProducer {
     async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SubscribeToProducer) {
-        println!("Recieved Subscription Request");
+        println!("Recieved Subscription Request {:}", msg.sender.actor_id);
         self.subscribers.push(msg.sender);
     }
 }
diff --git a/src/actor.rs b/src/actor.rs
index 1e7b46c..2fd0629 100644
--- a/src/actor.rs
+++ b/src/actor.rs
@@ -1,7 +1,7 @@
 use crate::addr::ActorEvent;
+use crate::error::Result;
 use crate::runtime::spawn;
 use crate::{Addr, Context};
-use crate::error::Result;
 use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
 use futures::channel::oneshot;
 use futures::{FutureExt, StreamExt};
@@ -158,8 +158,10 @@ impl<A: Actor> ActorManager<A> {
                         ActorEvent::Exec(f) => f(&mut actor, &mut ctx).await,
                         ActorEvent::Stop(_err) => break,
                         ActorEvent::RemoveStream(id) => {
-                            if ctx.streams.contains(id) {
-                                ctx.streams.remove(id);
+                            let mut streams = ctx.streams.lock().unwrap();
+
+                            if streams.contains(id) {
+                                streams.remove(id);
                             }
                         }
                     }
diff --git a/src/addr.rs b/src/addr.rs
index a81c249..aa200dd 100644
--- a/src/addr.rs
+++ b/src/addr.rs
@@ -4,7 +4,7 @@ use futures::future::Shared;
 use futures::Future;
 use std::hash::{Hash, Hasher};
 use std::pin::Pin;
-use std::sync::{Arc, Mutex, Weak};
+use std::sync::{Arc, Weak};
 
 type ExecFuture<'a> = Pin<Box<dyn Future<Output = ()> + Send + 'a>>;
 
@@ -53,6 +53,8 @@ impl<A> PartialEq for Addr<A> {
     }
 }
 
+impl<A> Eq for Addr<A> {}
+
 impl<A> Hash for Addr<A> {
     fn hash<H: Hasher>(&self, state: &mut H) {
         self.actor_id.hash(state)
@@ -111,29 +113,33 @@ impl<A: Actor> Addr<A> {
     {
         let weak_tx = Arc::downgrade(&self.tx);
 
-        Caller {
-            actor_id: self.actor_id.clone(),
-            caller_fn: Mutex::new(Box::new(move |msg| {
-                let weak_tx_option = weak_tx.upgrade();
-                Box::pin(async move {
-                    match weak_tx_option {
-                        Some(tx) => {
-                            let (oneshot_tx, oneshot_rx) = oneshot::channel();
-
-                            mpsc::UnboundedSender::clone(&tx).start_send(ActorEvent::Exec(
-                                Box::new(move |actor, ctx| {
-                                    Box::pin(async move {
-                                        let res = Handler::handle(&mut *actor, ctx, msg).await;
-                                        let _ = oneshot_tx.send(res);
-                                    })
-                                }),
-                            ))?;
-                            Ok(oneshot_rx.await?)
-                        }
-                        None => Err(crate::error::anyhow!("Actor Dropped")),
+        let closure = move |msg: T| {
+            let weak_tx_option = weak_tx.upgrade();
+            Box::pin(async move {
+                match weak_tx_option {
+                    Some(tx) => {
+                        let (oneshot_tx, oneshot_rx) = oneshot::channel();
+
+                        mpsc::UnboundedSender::clone(&tx).start_send(ActorEvent::Exec(
+                            Box::new(move |actor, ctx| {
+                                Box::pin(async move {
+                                    let res = Handler::handle(&mut *actor, ctx, msg).await;
+                                    let _ = oneshot_tx.send(res);
+                                })
+                            }),
+                        ))?;
+
+                        let result = oneshot_rx.await?;
+                        Ok(result)
                     }
-                })
-            })),
+                    None => Err(crate::error::anyhow!("Actor Dropped")),
+                }
+            }) as Pin<Box<dyn Future<Output = Result<T::Result>>>>
+        };
+
+        Caller {
+            actor_id: self.actor_id,
+            caller_fn: Box::new(closure),
         }
     }
 
@@ -143,21 +149,25 @@ impl<A: Actor> Addr<A> {
         A: Handler<T>,
     {
         let weak_tx = Arc::downgrade(&self.tx);
+
+        let closure = move |msg| match weak_tx.upgrade() {
+            Some(tx) => {
+                mpsc::UnboundedSender::clone(&tx).start_send(ActorEvent::Exec(Box::new(
+                    move |actor, ctx| {
+                        Box::pin(async move {
+                            Handler::handle(&mut *actor, ctx, msg).await;
+                        })
+                    },
+                )))?;
+                Ok(())
+            }
+            None => Ok(()),
+        };
+
+        let sender_fn = Box::new(closure);
         Sender {
-            actor_id: self.actor_id.clone(),
-            sender_fn: Box::new(move |msg| match weak_tx.upgrade() {
-                Some(tx) => {
-                    mpsc::UnboundedSender::clone(&tx).start_send(ActorEvent::Exec(Box::new(
-                        move |actor, ctx| {
-                            Box::pin(async move {
-                                Handler::handle(&mut *actor, ctx, msg).await;
-                            })
-                        },
-                    )))?;
-                    Ok(())
-                }
-                None => Ok(()),
-            }),
+            actor_id: self.actor_id,
+            sender_fn,
         }
     }
 
@@ -183,6 +193,8 @@ impl<A> PartialEq for WeakAddr<A> {
     }
 }
 
+impl<A> Eq for WeakAddr<A> {}
+
 impl<A> Hash for WeakAddr<A> {
     fn hash<H: Hasher>(&self, state: &mut H) {
         self.actor_id.hash(state)
@@ -191,14 +203,11 @@ impl<A> Hash for WeakAddr<A> {
 
 impl<A> WeakAddr<A> {
     pub fn upgrade(&self) -> Option<Addr<A>> {
-        match self.tx.upgrade() {
-            Some(tx) => Some(Addr {
-                actor_id: self.actor_id,
-                tx,
-                rx_exit: self.rx_exit.clone(),
-            }),
-            None => None,
-        }
+        self.tx.upgrade().map(|tx| Addr {
+            actor_id: self.actor_id,
+            tx,
+            rx_exit: self.rx_exit.clone(),
+        })
     }
 }
 
diff --git a/src/broker.rs b/src/broker.rs
index 9b0b1ab..8298428 100644
--- a/src/broker.rs
+++ b/src/broker.rs
@@ -1,14 +1,9 @@
-use crate::{Actor, Addr, Context, Handler, Message, Result, Sender, Service};
+use crate::{Actor, ActorId, Addr, Context, Handler, Message, Result, Sender, Service};
 use fnv::FnvHasher;
-use std::any::Any;
-use std::collections::HashMap;
-use std::hash::BuildHasherDefault;
-use std::marker::PhantomData;
-
-type SubscriptionId = u64;
+use std::{collections::HashMap, hash::BuildHasherDefault};
 
 pub(crate) struct Subscribe<T: Message<Result = ()>> {
-    pub(crate) id: SubscriptionId,
+    pub(crate) actor_id: ActorId,
     pub(crate) sender: Sender<T>,
 }
 
@@ -17,19 +12,12 @@ impl<T: Message<Result = ()>> Message for Subscribe<T> {
 }
 
 pub(crate) struct Unsubscribe {
-    pub(crate) id: SubscriptionId,
+    pub(crate) actor_id: ActorId,
 }
 
 impl Message for Unsubscribe {
     type Result = ();
 }
-
-struct Publish<T: Message<Result = ()> + Clone>(T);
-
-impl<T: Message<Result = ()> + Clone> Message for Publish<T> {
-    type Result = ();
-}
-
 /// Message broker is used to support publishing and subscribing to messages.
 ///
 /// # Examples
@@ -86,15 +74,13 @@ impl<T: Message<Result = ()> + Clone> Message for Publish<T> {
 /// }
 /// ```
 pub struct Broker<T: Message<Result = ()>> {
-    subscribes: HashMap<SubscriptionId, Box<dyn Any + Send>, BuildHasherDefault<FnvHasher>>,
-    mark: PhantomData<T>,
+    subscribers: HashMap<ActorId, Sender<T>, BuildHasherDefault<FnvHasher>>,
 }
 
 impl<T: Message<Result = ()>> Default for Broker<T> {
     fn default() -> Self {
         Self {
-            subscribes: Default::default(),
-            mark: PhantomData,
+            subscribers: Default::default(),
         }
     }
 }
@@ -106,31 +92,29 @@ impl<T: Message<Result = ()>> Service for Broker<T> {}
 #[async_trait::async_trait]
 impl<T: Message<Result = ()>> Handler<Subscribe<T>> for Broker<T> {
     async fn handle(&mut self, _ctx: &mut Context<Self>, msg: Subscribe<T>) {
-        self.subscribes.insert(msg.id, Box::new(msg.sender));
+        self.subscribers.insert(msg.actor_id, msg.sender);
     }
 }
 
 #[async_trait::async_trait]
 impl<T: Message<Result = ()>> Handler<Unsubscribe> for Broker<T> {
     async fn handle(&mut self, _ctx: &mut Context<Self>, msg: Unsubscribe) {
-        self.subscribes.remove(&msg.id);
+        self.subscribers.remove(&msg.actor_id);
     }
 }
 
 #[async_trait::async_trait]
-impl<T: Message<Result = ()> + Clone> Handler<Publish<T>> for Broker<T> {
-    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: Publish<T>) {
-        for sender in self.subscribes.values_mut() {
-            if let Some(sender) = sender.downcast_mut::<Sender<T>>() {
-                sender.send(msg.0.clone()).ok();
-            }
-        }
+impl<T: Message<Result = ()> + Clone> Handler<T> for Broker<T> {
+    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: T) {
+        // Broadcast to all subscribers and remove any senders that return an error (most likely because reciever dropped because actor already stopped)
+        self.subscribers
+            .retain(|_actor_id, sender| sender.send(msg.clone()).is_ok())
     }
 }
 
 impl<T: Message<Result = ()> + Clone> Addr<Broker<T>> {
     /// Publishes a message of the specified type.
     pub fn publish(&mut self, msg: T) -> Result<()> {
-        self.send(Publish(msg))
+        self.send(msg)
     }
 }
diff --git a/src/caller.rs b/src/caller.rs
index b70a03a..5142e98 100644
--- a/src/caller.rs
+++ b/src/caller.rs
@@ -2,55 +2,57 @@ use crate::{ActorId, Message, Result};
 use std::future::Future;
 use std::hash::{Hash, Hasher};
 use std::pin::Pin;
-use std::sync::Mutex;
-
-pub(crate) type CallerFuture<T> =
-    Pin<Box<dyn Future<Output = Result<<T as Message>::Result>> + Send + 'static>>;
-
-pub(crate) type CallerFn<T> = Box<dyn Fn(T) -> CallerFuture<T> + Send + 'static>;
-
-pub(crate) type SenderFn<T> = Box<dyn Fn(T) -> Result<()> + 'static + Send>;
 
 /// Caller of a specific message type
 ///
-/// Like `Sender<T>, Caller has a weak reference to the recipient of the message type, and so will not prevent an actor from stopping if all Addr's have been dropped elsewhere.
+/// Like `Sender<T>`, Caller has a weak reference to the recipient of the message type, and so will not prevent an actor from stopping if all Addr's have been dropped elsewhere.
+/// This takes a boxed closure with the message as a parameter with the mpsc channel of the actor inside and the type of actor abstracted away.
 
 pub struct Caller<T: Message> {
     pub actor_id: ActorId,
-    pub(crate) caller_fn: Mutex<CallerFn<T>>,
+    pub(crate) caller_fn: Box<dyn CallerFn<T>>,
 }
 
 impl<T: Message> Caller<T> {
-    pub fn call(&self, msg: T) -> CallerFuture<T> {
-        (self.caller_fn.lock().unwrap())(msg)
+    pub async fn call(&self, msg: T) -> Result<T::Result> {
+        self.caller_fn.call(msg).await
     }
 }
 
-impl<T: Message<Result = ()>> PartialEq for Caller<T> {
+impl<T: Message> PartialEq for Caller<T> {
     fn eq(&self, other: &Self) -> bool {
         self.actor_id == other.actor_id
     }
 }
 
-impl<T: Message<Result = ()>> Hash for Caller<T> {
+impl<T: Message> Hash for Caller<T> {
     fn hash<H: Hasher>(&self, state: &mut H) {
         self.actor_id.hash(state)
     }
 }
 
+impl<T: Message> Clone for Caller<T> {
+    fn clone(&self) -> Caller<T> {
+        Caller {
+            actor_id: self.actor_id,
+            caller_fn: dyn_clone::clone_box(&*self.caller_fn),
+        }
+    }
+}
+
 /// Sender of a specific message type
 ///
-/// Like `Caller<T>, Sender has a weak reference to the recipient of the message type, and so will not prevent an actor from stopping if all Addr's have been dropped elsewhere.
+/// Like `Caller<T>`, Sender has a weak reference to the recipient of the message type, and so will not prevent an actor from stopping if all Addr's have been dropped elsewhere.
 /// This allows it to be used in `send_later` `send_interval` actor functions, and not keep the actor alive indefinitely even after all references to it have been dropped (unless `ctx.stop()` is called from within)
 
 pub struct Sender<T: Message> {
     pub actor_id: ActorId,
-    pub(crate) sender_fn: SenderFn<T>,
+    pub(crate) sender_fn: Box<dyn SenderFn<T>>,
 }
 
 impl<T: Message<Result = ()>> Sender<T> {
     pub fn send(&self, msg: T) -> Result<()> {
-        (self.sender_fn)(msg)
+        self.sender_fn.send(msg)
     }
 }
 
@@ -65,3 +67,50 @@ impl<T: Message<Result = ()>> Hash for Sender<T> {
         self.actor_id.hash(state)
     }
 }
+
+impl<T: Message<Result = ()>> Clone for Sender<T> {
+    fn clone(&self) -> Sender<T> {
+        Sender {
+            actor_id: self.actor_id,
+            sender_fn: dyn_clone::clone_box(&*self.sender_fn),
+        }
+    }
+}
+
+// https://stackoverflow.com/questions/63842261/how-to-derive-clone-for-structures-with-boxed-closure
+// https://users.rust-lang.org/t/expected-opaque-type-found-a-different-opaque-type-when-trying-futures-join-all/40596/3
+use dyn_clone::DynClone;
+
+pub trait SenderFn<T>: DynClone + 'static + Send + Sync
+where
+    T: Message,
+{
+    fn send(&self, msg: T) -> Result<()>;
+}
+
+impl<F, T> SenderFn<T> for F
+where
+    F: Fn(T) -> Result<()> + 'static + Send + Sync + Clone,
+    T: Message,
+{
+    fn send(&self, msg: T) -> Result<()> {
+        self(msg)
+    }
+}
+
+pub trait CallerFn<T>: DynClone + 'static + Send + Sync
+where
+    T: Message,
+{
+    fn call(&self, msg: T) -> Pin<Box<dyn Future<Output = Result<T::Result>>>>;
+}
+
+impl<F, T> CallerFn<T> for F
+where
+    F: Fn(T) -> Pin<Box<dyn Future<Output = Result<T::Result>>>> + 'static + Send + Sync + Clone,
+    T: Message,
+{
+    fn call(&self, msg: T) -> Pin<Box<dyn Future<Output = Result<T::Result>>>> {
+        self(msg)
+    }
+}
diff --git a/src/context.rs b/src/context.rs
index fa30be5..553353d 100644
--- a/src/context.rs
+++ b/src/context.rs
@@ -7,7 +7,9 @@ use futures::future::{AbortHandle, Abortable, Shared};
 use futures::{Stream, StreamExt};
 use once_cell::sync::OnceCell;
 use slab::Slab;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::fmt;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Mutex;
 use std::sync::{Arc, Weak};
 use std::time::Duration;
 
@@ -16,8 +18,16 @@ pub struct Context<A> {
     actor_id: ActorId,
     tx: Weak<mpsc::UnboundedSender<ActorEvent<A>>>,
     pub(crate) rx_exit: Option<Shared<oneshot::Receiver<()>>>,
-    pub(crate) streams: Slab<AbortHandle>,
-    pub(crate) intervals: Slab<AbortHandle>,
+    pub(crate) streams: Arc<Mutex<Slab<AbortHandle>>>,
+    pub(crate) intervals: Arc<Mutex<Slab<AbortHandle>>>,
+}
+
+impl<A> fmt::Debug for Context<A> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Context")
+            .field("actor_id", &self.actor_id)
+            .finish()
+    }
 }
 
 impl<A> Context<A> {
@@ -28,7 +38,7 @@ impl<A> Context<A> {
         mpsc::UnboundedReceiver<ActorEvent<A>>,
         Arc<mpsc::UnboundedSender<ActorEvent<A>>>,
     ) {
-        static ACTOR_ID: OnceCell<AtomicU64> = OnceCell::new();
+        static ACTOR_ID: OnceCell<AtomicUsize> = OnceCell::new();
 
         // Get an actor id
         let actor_id = ACTOR_ID
@@ -76,14 +86,18 @@ impl<A> Context<A> {
     }
 
     pub fn abort_intervals(&mut self) {
-        for handle in self.intervals.drain() {
-            handle.abort()
+        if let Ok(mut intervals) = self.intervals.lock() {
+            for handle in intervals.drain() {
+                handle.abort()
+            }
         }
     }
 
     pub fn abort_streams(&mut self) {
-        for handle in self.streams.drain() {
-            handle.abort();
+        if let Ok(mut intervals) = self.streams.lock() {
+            for handle in intervals.drain() {
+                handle.abort()
+            }
         }
     }
 
@@ -149,7 +163,9 @@ impl<A> Context<A> {
         A: StreamHandler<S::Item>,
     {
         let tx = self.tx.clone();
-        let entry = self.streams.vacant_entry();
+        let mut streams = self.streams.lock().unwrap();
+
+        let entry = streams.vacant_entry();
         let id = entry.key();
         let (handle, registration) = futures::future::AbortHandle::new_pair();
         entry.insert(handle);
@@ -209,59 +225,78 @@ impl<A> Context<A> {
     ///
     /// We use `Sender` instead of `Addr` so that the interval doesn't keep reference to address and prevent the actor from being dropped and stopped
 
-    pub fn send_later<T>(&mut self, msg: T, after: Duration)
+    pub fn send_later<T>(&mut self, msg: T, after: Duration) -> AbortHandle
     where
         A: Handler<T>,
         T: Message<Result = ()>,
     {
         let sender = self.address().sender();
-        let entry = self.intervals.vacant_entry();
+        let intervals_clone = self.intervals.clone();
+
+        let mut intervals = self.intervals.lock().unwrap();
+
+        let entry = intervals.vacant_entry();
+        let key = entry.key();
+
         let (handle, registration) = futures::future::AbortHandle::new_pair();
-        entry.insert(handle);
+        entry.insert(handle.clone());
 
         spawn(Abortable::new(
             async move {
                 sleep(after).await;
                 sender.send(msg).ok();
+                // We have to remove the entry after the send has been completed or the slab will grow indefinitely
+                let mut intervals = intervals_clone.lock().unwrap();
+                intervals.remove(key);
             },
             registration,
         ));
+        handle
     }
 
     /// Sends the message  to self, at a specified fixed interval.
     /// The message is created each time using a closure `f`.
-    pub fn send_interval_with<T, F>(&mut self, f: F, dur: Duration)
+    pub fn send_interval_with<T, F>(&mut self, f: F, dur: Duration) -> AbortHandle
     where
         A: Handler<T>,
         F: Fn() -> T + Sync + Send + 'static,
         T: Message<Result = ()>,
     {
         let sender = self.address().sender();
+        let intervals_clone = self.intervals.clone();
+
+        let mut intervals = self.intervals.lock().unwrap();
+
+        let entry = intervals.vacant_entry();
+        let key = entry.key();
 
-        let entry = self.intervals.vacant_entry();
         let (handle, registration) = futures::future::AbortHandle::new_pair();
-        entry.insert(handle);
+        entry.insert(handle.clone());
 
         spawn(Abortable::new(
             async move {
                 loop {
                     sleep(dur).await;
                     if sender.send(f()).is_err() {
+                        // Again, we have to remove the entry after the send has been completed or the slab will grow indefinitely
+                        let mut intervals = intervals_clone.lock().unwrap();
+                        intervals.remove(key);
                         break;
                     }
                 }
             },
             registration,
         ));
+        handle
     }
 
     /// Sends the message `msg` to self, at a specified fixed interval.
-    pub fn send_interval<T>(&mut self, msg: T, dur: Duration)
+    pub fn send_interval<T>(&mut self, msg: T, dur: Duration) -> AbortHandle
     where
         A: Handler<T>,
         T: Message<Result = ()> + Clone + Sync,
     {
-        self.send_interval_with(move || msg.clone(), dur);
+        self.send_interval_with(move || msg.clone(), dur)
     }
 
     /// Subscribes to a message of a specified type.
@@ -273,7 +308,7 @@ impl<A> Context<A> {
         let sender = self.address().sender();
         broker
             .send(Subscribe {
-                id: self.actor_id,
+                actor_id: self.actor_id,
                 sender,
             })
             .ok();
@@ -283,6 +318,8 @@ impl<A> Context<A> {
     /// Unsubscribe to a message of a specified type.
     pub async fn unsubscribe<T: Message<Result = ()>>(&self) -> Result<()> {
         let broker = Broker::<T>::from_registry().await?;
-        broker.send(Unsubscribe { id: self.actor_id })
+        broker.send(Unsubscribe {
+            actor_id: self.actor_id,
+        })
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 26343e7..baf7abc 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -72,15 +72,17 @@ mod service;
 mod supervisor;
 
 #[cfg(all(feature = "anyhow", feature = "eyre"))]
-compile_error!(r#"
+compile_error!(
+    r#"
     features `xactor/anyhow` and `xactor/eyre` are mutually exclusive.
     If you are trying to disable anyhow set `default-features = false`.
-"#);
+"#
+);
 
-#[cfg(feature="anyhow")]
+#[cfg(feature = "anyhow")]
 pub use anyhow as error;
 
-#[cfg(feature="eyre")]
+#[cfg(feature = "eyre")]
 pub use eyre as error;
 
 /// Alias of error::Result
@@ -89,7 +91,7 @@ pub type Result<T> = error::Result<T>;
 /// Alias of error::Error
 pub type Error = error::Error;
 
-pub type ActorId = u64;
+pub type ActorId = usize;
 
 pub use actor::{Actor, Handler, Message, StreamHandler};
 pub use addr::{Addr, WeakAddr};
diff --git a/src/service.rs b/src/service.rs
index bf2127d..fcac805 100644
--- a/src/service.rs
+++ b/src/service.rs
@@ -1,6 +1,6 @@
 use crate::actor::ActorManager;
-use crate::{Actor, Addr};
 use crate::error::Result;
+use crate::{Actor, Addr};
 use fnv::FnvHasher;
 use futures::lock::Mutex;
 use once_cell::sync::OnceCell;
diff --git a/src/supervisor.rs b/src/supervisor.rs
index d220e70..4fa0f7e 100644
--- a/src/supervisor.rs
+++ b/src/supervisor.rs
@@ -1,7 +1,7 @@
 use crate::addr::ActorEvent;
+use crate::error::Result;
 use crate::runtime::spawn;
 use crate::{Actor, Addr, Context};
-use crate::error::Result;
 use futures::StreamExt;
 
 /// Actor supervisor
@@ -97,8 +97,10 @@ impl Supervisor {
                             Some(ActorEvent::Stop(_err)) => break 'event_loop,
                             Some(ActorEvent::Exec(f)) => f(&mut actor, &mut ctx).await,
                             Some(ActorEvent::RemoveStream(id)) => {
-                                if ctx.streams.contains(id) {
-                                    ctx.streams.remove(id);
+                                let mut streams = ctx.streams.lock().unwrap();
+
+                                if streams.contains(id) {
+                                    streams.remove(id);
                                 }
                             }
                         }
diff --git a/xactor-derive/Cargo.toml b/xactor-derive/Cargo.toml
index 35e6ca9..8de5d11 100644
--- a/xactor-derive/Cargo.toml
+++ b/xactor-derive/Cargo.toml
@@ -16,6 +16,6 @@ categories = ["network-programming", "asynchronous"]
 proc-macro = true
 
 [dependencies]
-proc-macro2 = "1.0.6"
-syn = { version = "1.0.13", features = ["full"] }
-quote = "1.0.2"
+proc-macro2 = "1.0.36"
+syn = { version = "1.0.86", features = ["full"] }
+quote = "1.0.15"