nack/ack

example/main.go

···
 	i := 0
 	for {
 		i++
-		msg := pubsub.Message{
-			Topic: "topic a",
-			Data:  []byte(fmt.Sprintf("message %d", i)),
-		}
+		msg := pubsub.NewMessage("topic a", []byte(fmt.Sprintf("message %d", i)))
 
 		err = publisher.PublishMessage(msg)
 		if err != nil {

example/server/main.go

···
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/willdot/messagebroker/server"
 )
 
 func main() {
-	srv, err := server.New(":3000")
+	srv, err := server.New(":3000", time.Second, time.Second*2)
 	if err != nil {
 		log.Fatal(err)
 	}

pubsub/message.go

···
 type Message struct {
 	Topic string `json:"topic"`
 	Data  []byte `json:"data"`
+
+	ack chan bool
+}
+
+// NewMessage creates a new message
+func NewMessage(topic string, data []byte) *Message {
+	return &Message{
+		Topic: topic,
+		Data:  data,
+		ack:   make(chan bool),
+	}
+}
+
+// Ack will send the provided value of the ack to the server
+func (m *Message) Ack(ack bool) {
+	m.ack <- ack
 }

pubsub/publisher.go

···
 }
 
 // Publish will publish the given message to the server
-func (p *Publisher) PublishMessage(message Message) error {
+func (p *Publisher) PublishMessage(message *Message) error {
 	op := func(conn net.Conn) error {
 		// send topic first
 		topic := fmt.Sprintf("topic:%s", message.Topic)

pubsub/subscriber.go

···
 // Consumer allows the consumption of messages. If during the consumer receiving messages from the
 // server an error occurs, it will be stored in Err
 type Consumer struct {
-	msgs chan Message
+	msgs chan *Message
 	// TODO: better error handling? Maybe a channel of errors?
 	Err error
 }
 
 // Messages returns a channel in which this consumer will put messages onto. It is safe to range over the channel since it will be closed once
 // the consumer has finished either due to an error or from being cancelled.
-func (c *Consumer) Messages() <-chan Message {
+func (c *Consumer) Messages() <-chan *Message {
 	return c.msgs
 }
 
···
 // to read the messages
 func (s *Subscriber) Consume(ctx context.Context) *Consumer {
 	consumer := &Consumer{
-		msgs: make(chan Message),
+		msgs: make(chan *Message),
 	}
 
 	go s.consume(ctx, consumer)
···
 			return
 		}
 
-		msg, err := s.readMessage()
+		err := s.readMessage(consumer.msgs)
 		if err != nil {
 			consumer.Err = err
 			return
 		}
-
-		if msg != nil {
-			consumer.msgs <- *msg
-		}
 	}
 }
 
-func (s *Subscriber) readMessage() (*Message, error) {
-	var msg *Message
+func (s *Subscriber) readMessage(msgChan chan *Message) error {
+	// var msg *Message
 	op := func(conn net.Conn) error {
 		err := s.conn.SetReadDeadline(time.Now().Add(time.Second))
 		if err != nil {
···
 			return err
 		}
 
-		msg = &Message{
-			Data:  dataBuf,
-			Topic: string(topicBuf),
+		msg := NewMessage(string(topicBuf), dataBuf)
+
+		msgChan <- msg
+
+		ack := <-msg.ack
+
+		ackMessage := server.Nack
+		if ack {
+			ackMessage = server.Ack
 		}
 
-		return nil
+		err = binary.Write(s.conn, binary.BigEndian, ackMessage)
+		if err != nil {
+			return fmt.Errorf("failed to ack/nack message: %w", err)
+		}
 
+		return nil
 	}
 
 	err := s.connOperation(op)
 	if err != nil {
 		var neterr net.Error
 		if errors.As(err, &neterr) && neterr.Timeout() {
-			return nil, nil
+			return nil
 		}
-		return nil, err
+		return err
 	}
 
-	return msg, err
+	return err
 }
 
 func (s *Subscriber) connOperation(op connOpp) error {

pubsub/subscriber_test.go

···
 )
 
 func createServer(t *testing.T) {
-	server, err := server.New(serverAddr)
+	server, err := server.New(serverAddr, time.Millisecond*100, time.Millisecond*100)
 	require.NoError(t, err)
 
 	t.Cleanup(func() {
···
 	consumer := sub.Consume(ctx)
 	require.NoError(t, err)
 
-	var receivedMessages []Message
+	var receivedMessages []*Message
 	consumerFinCh := make(chan struct{})
 	go func() {
 		for msg := range consumer.Messages() {
+			msg.Ack(true)
 			receivedMessages = append(receivedMessages, msg)
 		}
 
-		require.NoError(t, err)
 		consumerFinCh <- struct{}{}
 	}()
 
···
 		publisher.Close()
 	})
 
-	msg := Message{
-		Topic: topicA,
-		Data:  []byte("hello world"),
-	}
+	msg := NewMessage(topicA, []byte("hello world"))
 
 	err = publisher.PublishMessage(msg)
 	require.NoError(t, err)
···
 }
 
 func TestPublishAndSubscribe(t *testing.T) {
-	createServer(t)
-
-	sub, err := NewSubscriber(serverAddr)
-	require.NoError(t, err)
-
-	t.Cleanup(func() {
-		sub.Close()
-	})
-
-	topics := []string{topicA, topicB}
-
-	err = sub.SubscribeToTopics(topics)
-	require.NoError(t, err)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	t.Cleanup(func() {
-		cancel()
-	})
-
-	consumer := sub.Consume(ctx)
-	require.NoError(t, err)
+	consumer, cancel := setupConsumer(t)
 
-	var receivedMessages []Message
+	var receivedMessages []*Message
 
 	consumerFinCh := make(chan struct{})
 	go func() {
 		for msg := range consumer.Messages() {
+			msg.Ack(true)
 			receivedMessages = append(receivedMessages, msg)
 		}
 
-		require.NoError(t, err)
 		consumerFinCh <- struct{}{}
 	}()
 
···
 	})
 
 	// send some messages
-	sentMessages := make([]Message, 0, 10)
+	sentMessages := make([]*Message, 0, 10)
 	for i := 0; i < 10; i++ {
-		msg := Message{
-			Topic: topicA,
-			Data:  []byte(fmt.Sprintf("message %d", i)),
-		}
+		msg := NewMessage(topicA, []byte(fmt.Sprintf("message %d", i)))
 
 		sentMessages = append(sentMessages, msg)
 
···
 	select {
 	case <-consumerFinCh:
 		break
-	case <-time.After(time.Second):
+	case <-time.After(time.Second * 5):
 		t.Fatal("timed out waiting for consumer to read messages")
 	}
 
+	// THIS IS SO HACKY
+	for _, msg := range receivedMessages {
+		msg.ack = nil
+	}
+
+	for _, msg := range sentMessages {
+		msg.ack = nil
+	}
+
 	assert.ElementsMatch(t, receivedMessages, sentMessages)
 }
+
+func TestPublishAndSubscribeNackMessage(t *testing.T) {
+	consumer, cancel := setupConsumer(t)
+
+	var receivedMessages []*Message
+
+	consumerFinCh := make(chan struct{})
+	timesMsgWasReceived := 0
+	go func() {
+		for msg := range consumer.Messages() {
+			msg.Ack(false)
+			timesMsgWasReceived++
+		}
+
+		consumerFinCh <- struct{}{}
+	}()
+
+	publisher, err := NewPublisher("localhost:9999")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		publisher.Close()
+	})
+
+	// send a message
+	msg := NewMessage(topicA, []byte("hello world"))
+
+	err = publisher.PublishMessage(msg)
+	require.NoError(t, err)
+
+	// give the consumer some time to read the messages -- TODO: make better!
+	time.Sleep(time.Millisecond * 500)
+	cancel()
+
+	select {
+	case <-consumerFinCh:
+		break
+	case <-time.After(time.Second * 5):
+		t.Fatal("timed out waiting for consumer to read messages")
+	}
+
+	assert.Empty(t, receivedMessages)
+	assert.Equal(t, 5, timesMsgWasReceived)
+}
+
+func setupConsumer(t *testing.T) (*Consumer, context.CancelFunc) {
+	createServer(t)
+
+	sub, err := NewSubscriber(serverAddr)
+	require.NoError(t, err)
+
+	t.Cleanup(func() {
+		sub.Close()
+	})
+
+	topics := []string{topicA, topicB}
+
+	err = sub.SubscribeToTopics(topics)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(func() {
+		cancel()
+	})
+
+	consumer := sub.Consume(ctx)
+	require.NoError(t, err)
+
+	return consumer, cancel
+}

server/server.go

···
 	Subscribe   Action = 1
 	Unsubscribe Action = 2
 	Publish     Action = 3
+	Ack         Action = 4
+	Nack        Action = 5
 )
 
 // Status represents the status of a request
···
 
 	mu     sync.Mutex
 	topics map[string]*topic
+
+	ackDelay   time.Duration
+	ackTimeout time.Duration
 }
 
 // New creates and starts a new server
-func New(Addr string) (*Server, error) {
+func New(Addr string, ackDelay, ackTimeout time.Duration) (*Server, error) {
 	lis, err := net.Listen("tcp", Addr)
 	if err != nil {
 		return nil, fmt.Errorf("failed to listen: %w", err)
 	}
 
 	srv := &Server{
-		lis:    lis,
-		topics: map[string]*topic{},
+		lis:        lis,
+		topics:     map[string]*topic{},
+		ackDelay:   ackDelay,
+		ackTimeout: ackTimeout,
 	}
 
 	go srv.start()
···
 		t = newTopic(topicName)
 	}
 
-	t.subscriptions[peer.Addr()] = subscriber{
-		peer:          peer,
-		currentOffset: 0,
-	}
+	t.subscriptions[peer.Addr()] = newSubscriber(peer, topicName, s.ackDelay, s.ackTimeout)
 
 	s.topics[topicName] = t
 }
···
 	var action Action
 	op := func(conn net.Conn) error {
 		if timeout > 0 {
-			err := conn.SetReadDeadline(time.Now().Add(timeout))
-			if err != nil {
+			if err := conn.SetReadDeadline(time.Now().Add(timeout)); err != nil {
 				slog.Error("failed to set connection read deadline", "error", err, "peer", peer.Addr())
 			}
+			defer func() {
+				if err := conn.SetReadDeadline(time.Time{}); err != nil {
+					slog.Error("failed to reset connection read deadline", "error", err, "peer", peer.Addr())
+				}
+			}()
 		}
 
 		err := binary.Read(conn, binary.BigEndian, &action)

server/server_test.go

···
 	topicC = "topic c"
 
 	serverAddr = ":6666"
+
+	ackDelay   = time.Millisecond * 100
+	ackTimeout = time.Millisecond * 100
 )
 
 func createServer(t *testing.T) *Server {
-	srv, err := New(serverAddr)
+	srv, err := New(serverAddr, ackDelay, ackTimeout)
 	require.NoError(t, err)
 
 	t.Cleanup(func() {
···
 	srv := createServer(t)
 	srv.topics[topicName] = &topic{
 		name:          topicName,
-		subscriptions: make(map[net.Addr]subscriber),
+		subscriptions: make(map[net.Addr]*subscriber),
 	}
 
 	return srv
···
 		buf := make([]byte, dataLen)
 		n, err := conn.Read(buf)
 		require.NoError(t, err)
+
 		require.Equal(t, int(dataLen), n)
 
 		assert.Equal(t, messageData, string(buf))
+
+		err = binary.Write(conn, binary.BigEndian, Ack)
+		require.NoError(t, err)
 	}
 }
 
···
 			require.Equal(t, int(dataLen), n)
 
 			results = append(results, string(buf))
+
+			err = binary.Write(subscriberConn, binary.BigEndian, Ack)
+			require.NoError(t, err)
 		}
 
 		assert.ElementsMatch(t, results, messages)
···
 		t.Fatal(fmt.Errorf("timed out waiting for subscriber to read messages"))
 	}
 }
+
+func TestSendsDataToTopicSubscriberNacksThenAcks(t *testing.T) {
+	_ = createServer(t)
+
+	subscriberConn := createConnectionAndSubscribe(t, []string{topicA, topicB})
+
+	publisherConn, err := net.Dial("tcp", fmt.Sprintf("localhost%s", serverAddr))
+	require.NoError(t, err)
+
+	err = binary.Write(publisherConn, binary.BigEndian, Publish)
+	require.NoError(t, err)
+
+	topic := fmt.Sprintf("topic:%s", topicA)
+	messageData := "hello world"
+
+	// send topic first
+	err = binary.Write(publisherConn, binary.BigEndian, uint32(len(topic)))
+	require.NoError(t, err)
+	_, err = publisherConn.Write([]byte(topic))
+	require.NoError(t, err)
+
+	// now send the data
+	err = binary.Write(publisherConn, binary.BigEndian, uint32(len(messageData)))
+	require.NoError(t, err)
+	n, err := publisherConn.Write([]byte(messageData))
+	require.NoError(t, err)
+	require.Equal(t, len(messageData), n)
+
+	// check the subsribers got the data
+	readMessage := func(conn net.Conn, ack Action) {
+		var topicLen uint64
+		err = binary.Read(conn, binary.BigEndian, &topicLen)
+		require.NoError(t, err)
+
+		topicBuf := make([]byte, topicLen)
+		_, err = conn.Read(topicBuf)
+		require.NoError(t, err)
+		assert.Equal(t, topicA, string(topicBuf))
+
+		var dataLen uint64
+		err = binary.Read(conn, binary.BigEndian, &dataLen)
+		require.NoError(t, err)
+
+		buf := make([]byte, dataLen)
+		n, err := conn.Read(buf)
+		require.NoError(t, err)
+
+		require.Equal(t, int(dataLen), n)
+
+		assert.Equal(t, messageData, string(buf))
+
+		err = binary.Write(conn, binary.BigEndian, ack)
+		require.NoError(t, err)
+	}
+
+	// NACK the  message and then ack it
+	readMessage(subscriberConn, Nack)
+	readMessage(subscriberConn, Ack)
+	// reading for another message should now timeout but give enough time for the ack delay to kick in
+	// should the second read of the message not have been ack'd properly
+	var topicLen uint64
+	_ = subscriberConn.SetReadDeadline(time.Now().Add(ackDelay + time.Millisecond*100))
+	err = binary.Read(subscriberConn, binary.BigEndian, &topicLen)
+	require.Error(t, err)
+}
+
+func TestSendsDataToTopicSubscriberDoesntAckMessage(t *testing.T) {
+	_ = createServer(t)
+
+	subscriberConn := createConnectionAndSubscribe(t, []string{topicA, topicB})
+
+	publisherConn, err := net.Dial("tcp", fmt.Sprintf("localhost%s", serverAddr))
+	require.NoError(t, err)
+
+	err = binary.Write(publisherConn, binary.BigEndian, Publish)
+	require.NoError(t, err)
+
+	topic := fmt.Sprintf("topic:%s", topicA)
+	messageData := "hello world"
+
+	// send topic first
+	err = binary.Write(publisherConn, binary.BigEndian, uint32(len(topic)))
+	require.NoError(t, err)
+	_, err = publisherConn.Write([]byte(topic))
+	require.NoError(t, err)
+
+	// now send the data
+	err = binary.Write(publisherConn, binary.BigEndian, uint32(len(messageData)))
+	require.NoError(t, err)
+	n, err := publisherConn.Write([]byte(messageData))
+	require.NoError(t, err)
+	require.Equal(t, len(messageData), n)
+
+	// check the subsribers got the data
+	readMessage := func(conn net.Conn, ack bool) {
+		var topicLen uint64
+		err = binary.Read(conn, binary.BigEndian, &topicLen)
+		require.NoError(t, err)
+
+		topicBuf := make([]byte, topicLen)
+		_, err = conn.Read(topicBuf)
+		require.NoError(t, err)
+		assert.Equal(t, topicA, string(topicBuf))
+
+		var dataLen uint64
+		err = binary.Read(conn, binary.BigEndian, &dataLen)
+		require.NoError(t, err)
+
+		buf := make([]byte, dataLen)
+		n, err := conn.Read(buf)
+		require.NoError(t, err)
+
+		require.Equal(t, int(dataLen), n)
+
+		assert.Equal(t, messageData, string(buf))
+
+		if ack {
+			err = binary.Write(conn, binary.BigEndian, Ack)
+			require.NoError(t, err)
+			return
+		}
+	}
+
+	// don't send ack or nack and then ack on the second attempt
+	readMessage(subscriberConn, false)
+	readMessage(subscriberConn, true)
+
+	// reading for another message should now timeout but give enough time for the ack delay to kick in
+	// should the second read of the message not have been ack'd properly
+	var topicLen uint64
+	_ = subscriberConn.SetReadDeadline(time.Now().Add(ackDelay + time.Millisecond*100))
+	err = binary.Read(subscriberConn, binary.BigEndian, &topicLen)
+	require.Error(t, err)
+}
+
+func TestSendsDataToTopicSubscriberDeliveryCountTooHighWithNoAck(t *testing.T) {
+	_ = createServer(t)
+
+	subscriberConn := createConnectionAndSubscribe(t, []string{topicA, topicB})
+
+	publisherConn, err := net.Dial("tcp", fmt.Sprintf("localhost%s", serverAddr))
+	require.NoError(t, err)
+
+	err = binary.Write(publisherConn, binary.BigEndian, Publish)
+	require.NoError(t, err)
+
+	topic := fmt.Sprintf("topic:%s", topicA)
+	messageData := "hello world"
+
+	// send topic first
+	err = binary.Write(publisherConn, binary.BigEndian, uint32(len(topic)))
+	require.NoError(t, err)
+	_, err = publisherConn.Write([]byte(topic))
+	require.NoError(t, err)
+
+	// now send the data
+	err = binary.Write(publisherConn, binary.BigEndian, uint32(len(messageData)))
+	require.NoError(t, err)
+	n, err := publisherConn.Write([]byte(messageData))
+	require.NoError(t, err)
+	require.Equal(t, len(messageData), n)
+
+	// check the subsribers got the data
+	readMessage := func(conn net.Conn, ack bool) {
+		var topicLen uint64
+		err = binary.Read(conn, binary.BigEndian, &topicLen)
+		require.NoError(t, err)
+
+		topicBuf := make([]byte, topicLen)
+		_, err = conn.Read(topicBuf)
+		require.NoError(t, err)
+		assert.Equal(t, topicA, string(topicBuf))
+
+		var dataLen uint64
+		err = binary.Read(conn, binary.BigEndian, &dataLen)
+		require.NoError(t, err)
+
+		buf := make([]byte, dataLen)
+		n, err := conn.Read(buf)
+		require.NoError(t, err)
+
+		require.Equal(t, int(dataLen), n)
+
+		assert.Equal(t, messageData, string(buf))
+
+		if ack {
+			err = binary.Write(conn, binary.BigEndian, Ack)
+			require.NoError(t, err)
+			return
+		}
+	}
+
+	// nack the message 5 times
+	readMessage(subscriberConn, false)
+	readMessage(subscriberConn, false)
+	readMessage(subscriberConn, false)
+	readMessage(subscriberConn, false)
+	readMessage(subscriberConn, false)
+
+	// reading for the message should now timeout as we have nack'd the message too many times
+	var topicLen uint64
+	_ = subscriberConn.SetReadDeadline(time.Now().Add(ackDelay + time.Millisecond*100))
+	err = binary.Read(subscriberConn, binary.BigEndian, &topicLen)
+	require.Error(t, err)
+}

server/subscriber.go

···
+package server
+
+import (
+	"encoding/binary"
+	"fmt"
+	"log/slog"
+	"net"
+	"time"
+
+	"github.com/willdot/messagebroker/server/peer"
+)
+
+type subscriber struct {
+	peer     *peer.Peer
+	topic    string
+	messages chan message
+
+	ackDelay   time.Duration
+	ackTimeout time.Duration
+}
+
+type message struct {
+	data          []byte
+	deliveryCount int
+}
+
+func newMessage(data []byte) message {
+	return message{data: data, deliveryCount: 1}
+}
+
+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,
+	}
+
+	go s.sendMessages()
+
+	return s
+}
+
+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())
+		}
+
+		if ack {
+			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)
+	}
+}
+
+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
+	}()
+}
+
+func (s *subscriber) sendMessage(topic string, msg message) (bool, error) {
+	var ack bool
+	op := func(conn net.Conn) error {
+		topicLen := uint64(len(topic))
+		err := binary.Write(conn, binary.BigEndian, topicLen)
+		if err != nil {
+			return fmt.Errorf("failed to send topic length: %w", err)
+		}
+		_, err = conn.Write([]byte(topic))
+		if err != nil {
+			return fmt.Errorf("failed to send topic: %w", err)
+		}
+
+		dataLen := uint64(len(msg.data))
+
+		err = binary.Write(conn, binary.BigEndian, dataLen)
+		if err != nil {
+			return fmt.Errorf("failed to send data length: %w", err)
+		}
+
+		_, err = conn.Write(msg.data)
+		if err != nil {
+			return fmt.Errorf("failed to write to peer: %w", err)
+		}
+
+		var ackRes Action
+		if err := conn.SetReadDeadline(time.Now().Add(s.ackTimeout)); err != nil {
+			slog.Error("failed to set connection read deadline", "error", err, "peer", s.peer.Addr())
+		}
+		defer func() {
+			if err := conn.SetReadDeadline(time.Time{}); err != nil {
+				slog.Error("failed to reset connection read deadline", "error", err, "peer", s.peer.Addr())
+			}
+		}()
+		err = binary.Read(conn, binary.BigEndian, &ackRes)
+		if err != nil {
+			return fmt.Errorf("failed to read ack from peer: %w", err)
+		}
+
+		if ackRes == Ack {
+			ack = true
+		}
+
+		return nil
+	}
+
+	err := s.peer.RunConnOperation(op)
+
+	return ack, err
+}

server/topic.go

···
 package server
 
 import (
-	"encoding/binary"
-	"fmt"
-	"log/slog"
 	"net"
 	"sync"
-
-	"github.com/willdot/messagebroker/server/peer"
 )
 
 type topic struct {
 	name          string
-	subscriptions map[net.Addr]subscriber
+	subscriptions map[net.Addr]*subscriber
 	mu            sync.Mutex
 }
 
-type subscriber struct {
-	peer          *peer.Peer
-	currentOffset int
-}
-
 func newTopic(name string) *topic {
 	return &topic{
 		name:          name,
-		subscriptions: make(map[net.Addr]subscriber),
+		subscriptions: make(map[net.Addr]*subscriber),
 	}
 }
 
···
 	subscribers := t.subscriptions
 	t.mu.Unlock()
 
-	var wg sync.WaitGroup
-
 	for _, subscriber := range subscribers {
-		wg.Add(1)
-		sub := subscriber
-		go func() {
-			defer wg.Done()
-			sendMessage(sub, t.name, msgData)
-		}()
-	}
-
-	wg.Wait()
-}
-
-func sendMessage(sub subscriber, topicName string, message []byte) {
-	err := sub.peer.RunConnOperation(sendMessageOp(topicName, message))
-	if err != nil {
-		slog.Error("failed to send to message", "error", err, "peer", sub.peer.Addr())
-		return
-	}
-}
-
-func sendMessageOp(topic string, data []byte) peer.ConnOpp {
-	return func(conn net.Conn) error {
-		topicLen := uint64(len(topic))
-		err := binary.Write(conn, binary.BigEndian, topicLen)
-		if err != nil {
-			return fmt.Errorf("failed to send topic length: %w", err)
-		}
-		_, err = conn.Write([]byte(topic))
-		if err != nil {
-			return fmt.Errorf("failed to send topic: %w", err)
-		}
-
-		dataLen := uint64(len(data))
-
-		err = binary.Write(conn, binary.BigEndian, dataLen)
-		if err != nil {
-			return fmt.Errorf("failed to send data length: %w", err)
-		}
-
-		_, err = conn.Write(data)
-		if err != nil {
-			return fmt.Errorf("failed to write to peer: %w", err)
-		}
-		return nil
+		subscriber.addMessage(newMessage(msgData), 0)
 	}
 }