diff --git a/server/server.go b/server/server.go
index fac9b8e..c5c7f2c 100644
--- a/server/server.go
+++ b/server/server.go
@@ -364,7 +364,11 @@ func (s *Server) removeSubsciberFromTopic(topicName string, peer *peer.Peer) {
 	if !ok {
 		return
 	}
-
+	sub, ok := t.subscriptions[peer.Addr()]
+	if !ok {
+		return
+	}
+	sub.unsubscribe()
 	delete(t.subscriptions, peer.Addr())
 }
 
@@ -373,6 +377,11 @@ func (s *Server) unsubscribePeerFromAllTopics(peer *peer.Peer) {
 	defer s.mu.Unlock()
 
 	for _, topic := range s.topics {
+		sub, ok := topic.subscriptions[peer.Addr()]
+		if !ok {
+			continue
+		}
+		sub.unsubscribe()
 		delete(topic.subscriptions, peer.Addr())
 	}
 }
diff --git a/server/subscriber.go b/server/subscriber.go
index a182070..344baea 100644
--- a/server/subscriber.go
+++ b/server/subscriber.go
@@ -11,9 +11,10 @@ import (
 )
 
 type subscriber struct {
-	peer     *peer.Peer
-	topic    string
-	messages chan message
+	peer          *peer.Peer
+	topic         string
+	messages      chan message
+	unsubscribeCh chan struct{}
 
 	ackDelay   time.Duration
 	ackTimeout time.Duration
@@ -30,11 +31,12 @@ func newMessage(data []byte) message {
 
 func newSubscriber(peer *peer.Peer, topic string, ackDelay, ackTimeout time.Duration) *subscriber {
 	s := &subscriber{
-		peer:       peer,
-		topic:      topic,
-		messages:   make(chan message),
-		ackDelay:   ackDelay,
-		ackTimeout: ackTimeout,
+		peer:          peer,
+		topic:         topic,
+		messages:      make(chan message),
+		ackDelay:      ackDelay,
+		ackTimeout:    ackTimeout,
+		unsubscribeCh: make(chan struct{}),
 	}
 
 	go s.sendMessages()
@@ -43,32 +45,42 @@ func newSubscriber(peer *peer.Peer, topic string, ackDelay, ackTimeout time.Dura
 }
 
 func (s *subscriber) sendMessages() {
-	// TODO: should think about how to break out of this if the subsciber closes its connection etc
-	for msg := range s.messages {
-		ack, err := s.sendMessage(s.topic, msg)
-		if err != nil {
-			slog.Error("failed to send to message", "error", err, "peer", s.peer.Addr())
-		}
+	for {
+		select {
+		case <-s.unsubscribeCh:
+			return
+		case msg := <-s.messages:
+			ack, err := s.sendMessage(s.topic, msg)
+			if err != nil {
+				slog.Error("failed to send to message", "error", err, "peer", s.peer.Addr())
+			}
 
-		if ack {
-			continue
-		}
+			if ack {
+				continue
+			}
 
-		if msg.deliveryCount >= 5 {
-			slog.Error("max delivery count for message. Dropping", "peer", s.peer.Addr())
-			continue
-		}
+			if msg.deliveryCount >= 5 {
+				slog.Error("max delivery count for message. Dropping", "peer", s.peer.Addr())
+				continue
+			}
 
-		msg.deliveryCount++
-		s.addMessage(msg, s.ackDelay)
+			msg.deliveryCount++
+			s.addMessage(msg, s.ackDelay)
+		}
 	}
 }
 
 func (s *subscriber) addMessage(msg message, delay time.Duration) {
 	go func() {
-		time.Sleep(delay)
-		// TODO: should think about how to break out of this if the subsciber closes its connection etc
-		s.messages <- msg
+		timer := time.NewTimer(delay)
+		defer timer.Stop()
+
+		select {
+		case <-s.unsubscribeCh:
+			return
+		case <-timer.C:
+			s.messages <- msg
+		}
 	}()
 }
 
@@ -122,3 +134,7 @@ func (s *subscriber) sendMessage(topic string, msg message) (bool, error) {
 
 	return ack, err
 }
+
+func (s *subscriber) unsubscribe() {
+	close(s.unsubscribeCh)
+}