write to connection once

README.md

···
 
 ### Sending data via a connection
 
-When sending a message representing an action (subscribe, publish etc) then a uint8 binary message is sent. 
+When sending a message representing an action (subscribe, publish etc) then a uint16 binary message is sent. 
 
 When sending any other data, the length of the data is to be sent first using a binary uint32 and then the actual data sent afterwards. 

example/main.go

···
 
 	for msg := range consumer.Messages() {
 		slog.Info("received message", "message", string(msg.Data))
+		msg.Ack(true)
 	}
 
 }

pubsub/publisher.go

···
 	op := func(conn net.Conn) error {
 		// send topic first
 		topic := fmt.Sprintf("topic:%s", message.Topic)
-		err := binary.Write(p.conn, binary.BigEndian, uint32(len(topic)))
-		if err != nil {
-			return fmt.Errorf("failed to write topic size to server")
-		}
+
+		topicLenB := make([]byte, 4)
+		binary.BigEndian.PutUint32(topicLenB, uint32(len(topic)))
 
-		_, err = p.conn.Write([]byte(topic))
-		if err != nil {
-			return fmt.Errorf("failed to write topic to server")
-		}
+		headers := append(topicLenB, []byte(topic)...)
 
-		err = binary.Write(p.conn, binary.BigEndian, uint32(len(message.Data)))
-		if err != nil {
-			return fmt.Errorf("failed to write message size to server")
-		}
+		messageLenB := make([]byte, 4)
+		binary.BigEndian.PutUint32(messageLenB, uint32(len(message.Data)))
+		headers = append(headers, messageLenB...)
 
-		_, err = p.conn.Write(message.Data)
+		_, err := conn.Write(append(headers, message.Data...))
 		if err != nil {
 			return fmt.Errorf("failed to publish data to server")
 		}

pubsub/subscriber.go

···
 // SubscribeToTopics will subscribe to the provided topics
 func (s *Subscriber) SubscribeToTopics(topicNames []string) error {
 	op := func(conn net.Conn) error {
-		err := binary.Write(conn, binary.BigEndian, server.Subscribe)
-		if err != nil {
-			return fmt.Errorf("failed to subscribe: %w", err)
-		}
+		actionB := make([]byte, 2)
+		binary.BigEndian.PutUint16(actionB, server.Subscribed)
+		headers := actionB
 
 		b, err := json.Marshal(topicNames)
 		if err != nil {
 			return fmt.Errorf("failed to marshal topic names: %w", err)
 		}
 
-		err = binary.Write(conn, binary.BigEndian, uint32(len(b)))
-		if err != nil {
-			return fmt.Errorf("failed to write topic data length: %w", err)
-		}
+		topicNamesB := make([]byte, 4)
+		binary.BigEndian.PutUint32(topicNamesB, uint32(len(b)))
+		headers = append(headers, topicNamesB...)
 
-		_, err = conn.Write(b)
+		_, err = conn.Write(append(headers, b...))
 		if err != nil {
 			return fmt.Errorf("failed to subscribe to topics: %w", err)
 		}
···
 // UnsubscribeToTopics will unsubscribe to the provided topics
 func (s *Subscriber) UnsubscribeToTopics(topicNames []string) error {
 	op := func(conn net.Conn) error {
-		err := binary.Write(conn, binary.BigEndian, server.Unsubscribe)
-		if err != nil {
-			return fmt.Errorf("failed to unsubscribe: %w", err)
-		}
+		actionB := make([]byte, 2)
+		binary.BigEndian.PutUint16(actionB, uint16(server.Unsubscribe))
+		headers := actionB
 
 		b, err := json.Marshal(topicNames)
 		if err != nil {
 			return fmt.Errorf("failed to marshal topic names: %w", err)
 		}
 
-		err = binary.Write(conn, binary.BigEndian, uint32(len(b)))
-		if err != nil {
-			return fmt.Errorf("failed to write topic data length: %w", err)
-		}
+		topicNamesB := make([]byte, 4)
+		binary.BigEndian.PutUint32(topicNamesB, uint32(len(b)))
+		headers = append(headers, topicNamesB...)
 
-		_, err = conn.Write(b)
+		_, err = conn.Write(append(headers, b...))
 		if err != nil {
 			return fmt.Errorf("failed to unsubscribe to topics: %w", err)
 		}
···
 			return ctx.Err()
 		case ack = <-msg.ack:
 		}
-		//ack := <-msg.ack
-
 		ackMessage := server.Nack
 		if ack {
 			ackMessage = server.Ack

server/server.go

···
 )
 
 // Action represents the type of action that a peer requests to do
-type Action uint8
+type Action uint16
 
 const (
 	Subscribe   Action = 1
···
 )
 
 // Status represents the status of a request
-type Status uint8
+type Status uint16
 
 const (
 	Subscribed   = 1
···
 }
 
 func writeStatus(status Status, message string, conn net.Conn) {
-	err := binary.Write(conn, binary.BigEndian, status)
-	if err != nil {
-		if !errors.Is(err, syscall.EPIPE) {
-			slog.Error("failed to write status to peers connection", "error", err, "peer", conn.RemoteAddr())
-		}
-		return
-	}
+	statusB := make([]byte, 2)
+	binary.BigEndian.PutUint16(statusB, uint16(status))
+
+	headers := statusB
 
-	if message == "" {
-		return
+	if len(message) > 0 {
+		sizeB := make([]byte, 4)
+		binary.BigEndian.PutUint32(sizeB, uint32(len(message)))
+		headers = append(headers, sizeB...)
 	}
 
 	msgBytes := []byte(message)
-	err = binary.Write(conn, binary.BigEndian, uint32(len(msgBytes)))
-	if err != nil {
-		slog.Error("failed to write message length to peers connection", "error", err, "peer", conn.RemoteAddr())
-		return
-	}
-
-	_, err = conn.Write(msgBytes)
+	_, err := conn.Write(append(headers, msgBytes...))
 	if err != nil {
-		slog.Error("failed to write message to peers connection", "error", err, "peer", conn.RemoteAddr())
+		if !errors.Is(err, syscall.EPIPE) {
+			slog.Error("failed to write status to peers connection", "error", err, "peer", conn.RemoteAddr())
+		}
 		return
 	}
 }

server/server_test.go

···
 	conn, err := net.Dial("tcp", fmt.Sprintf("localhost%s", serverAddr))
 	require.NoError(t, err)
 
-	err = binary.Write(conn, binary.BigEndian, Subscribe)
-	require.NoError(t, err)
-
-	rawTopics, err := json.Marshal(topics)
-	require.NoError(t, err)
-
-	err = binary.Write(conn, binary.BigEndian, uint32(len(rawTopics)))
-	require.NoError(t, err)
-
-	_, err = conn.Write(rawTopics)
-	require.NoError(t, err)
+	subscribeOrUnsubscribetoTopics(t, conn, topics, Subscribe)
 
 	expectedRes := Subscribed
 
···
 	return conn
 }
 
+func sendMessage(t *testing.T, conn net.Conn, topic string, message []byte) {
+	topicLenB := make([]byte, 4)
+	binary.BigEndian.PutUint32(topicLenB, uint32(len(topic)))
+
+	headers := topicLenB
+	headers = append(headers, []byte(topic)...)
+
+	messageLenB := make([]byte, 4)
+	binary.BigEndian.PutUint32(messageLenB, uint32(len(message)))
+	headers = append(headers, messageLenB...)
+
+	_, err := conn.Write(append(headers, message...))
+	require.NoError(t, err)
+}
+
+func subscribeOrUnsubscribetoTopics(t *testing.T, conn net.Conn, topics []string, action Action) {
+	actionB := make([]byte, 2)
+	binary.BigEndian.PutUint16(actionB, uint16(action))
+	headers := actionB
+
+	b, err := json.Marshal(topics)
+	require.NoError(t, err)
+
+	topicNamesB := make([]byte, 4)
+	binary.BigEndian.PutUint32(topicNamesB, uint32(len(b)))
+	headers = append(headers, topicNamesB...)
+
+	_, err = conn.Write(append(headers, b...))
+	require.NoError(t, err)
+}
+
 func TestSubscribeToTopics(t *testing.T) {
 	// create a server with an existing topic so we can test subscribing to a new and
 	// existing topic
···
 	assert.Len(t, srv.topics[topicB].subscriptions, 1)
 	assert.Len(t, srv.topics[topicC].subscriptions, 1)
 
-	err := binary.Write(conn, binary.BigEndian, Unsubscribe)
-	require.NoError(t, err)
-
 	topics := []string{topicA, topicB}
-	rawTopics, err := json.Marshal(topics)
-	require.NoError(t, err)
 
-	err = binary.Write(conn, binary.BigEndian, uint32(len(rawTopics)))
-	require.NoError(t, err)
-
-	_, err = conn.Write(rawTopics)
-	require.NoError(t, err)
+	subscribeOrUnsubscribetoTopics(t, conn, topics, Unsubscribe)
 
 	expectedRes := Unsubscribed
 
 	var resp Status
-	err = binary.Read(conn, binary.BigEndian, &resp)
+	err := binary.Read(conn, binary.BigEndian, &resp)
 	require.NoError(t, err)
 
 	assert.Equal(t, expectedRes, int(resp))
···
 	require.NoError(t, err)
 
 	data := []byte("hello world")
-	// send data length first
-	err = binary.Write(publisherConn, binary.BigEndian, uint32(len(data)))
-	require.NoError(t, err)
-	n, err := publisherConn.Write(data)
-	require.NoError(t, err)
-	require.Equal(t, len(data), n)
+
+	sendMessage(t, publisherConn, topicA, data)
 
 	assert.Len(t, srv.topics, 2)
 	assert.Len(t, srv.topics[topicA].subscriptions, 0)
···
 	conn, err := net.Dial("tcp", fmt.Sprintf("localhost%s", serverAddr))
 	require.NoError(t, err)
 
-	err = binary.Write(conn, binary.BigEndian, uint8(99))
+	err = binary.Write(conn, binary.BigEndian, uint16(99))
 	require.NoError(t, err)
 
 	expectedRes := Error
···
 	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)
+	sendMessage(t, publisherConn, topic, []byte(messageData))
 
 	// check the subsribers got the data
 	for _, conn := range subscribers {
···
 
 	// send multiple messages
 	for _, msg := range messages {
-		// 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(msg)))
-		require.NoError(t, err)
-		n, err := publisherConn.Write([]byte(msg))
-		require.NoError(t, err)
-		require.Equal(t, len(msg), n)
+		sendMessage(t, publisherConn, topic, []byte(msg))
 	}
 
 	select {
···
 	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)
+	sendMessage(t, publisherConn, topic, []byte(messageData))
 
 	// check the subsribers got the data
 	readMessage := func(conn net.Conn, ack Action) {
···
 	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)
+	sendMessage(t, publisherConn, topic, []byte(messageData))
 
 	// check the subsribers got the data
 	readMessage := func(conn net.Conn, ack bool) {
···
 	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)
+	sendMessage(t, publisherConn, topic, []byte(messageData))
 
 	// check the subsribers got the data
 	readMessage := func(conn net.Conn, ack bool) {

server/subscriber.go

···
 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)
-		}
+		// TODO: why did I chose uint64 for topic len?
+		topicB := make([]byte, 8)
+		binary.BigEndian.PutUint64(topicB, uint64(len(topic)))
 
-		dataLen := uint64(len(msg.data))
+		headers := topicB
+		headers = append(headers, []byte(topic)...)
 
-		err = binary.Write(conn, binary.BigEndian, dataLen)
-		if err != nil {
-			return fmt.Errorf("failed to send data length: %w", err)
-		}
+		// TODO: if message is empty, return error?
+		dataLenB := make([]byte, 8)
+		binary.BigEndian.PutUint64(dataLenB, uint64(len(msg.data)))
+		headers = append(headers, dataLenB...)
 
-		_, err = conn.Write(msg.data)
+		_, err := conn.Write(append(headers, msg.data...))
 		if err != nil {
 			return fmt.Errorf("failed to write to peer: %w", err)
 		}