diff --git a/README.md b/README.md index c7591c5..13365a2 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,6 @@ Once a subscription has declared itself as a publisher, it will enter a loop whe ### 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. \ No newline at end of file diff --git a/example/main.go b/example/main.go index 87773a3..97e817e 100644 --- a/example/main.go +++ b/example/main.go @@ -41,6 +41,7 @@ func main() { for msg := range consumer.Messages() { slog.Info("received message", "message", string(msg.Data)) + msg.Ack(true) } } diff --git a/pubsub/publisher.go b/pubsub/publisher.go index 883baf3..4ba6614 100644 --- a/pubsub/publisher.go +++ b/pubsub/publisher.go @@ -43,22 +43,17 @@ func (p *Publisher) PublishMessage(message *Message) error { 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") - } - _, err = p.conn.Write([]byte(topic)) - if err != nil { - return fmt.Errorf("failed to write topic to server") - } + topicLenB := make([]byte, 4) + binary.BigEndian.PutUint32(topicLenB, uint32(len(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") - } + headers := append(topicLenB, []byte(topic)...) + + 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") } diff --git a/pubsub/subscriber.go b/pubsub/subscriber.go index b316a17..d146d49 100644 --- a/pubsub/subscriber.go +++ b/pubsub/subscriber.go @@ -41,22 +41,20 @@ func (s *Subscriber) Close() error { // 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) } @@ -92,22 +90,20 @@ func (s *Subscriber) SubscribeToTopics(topicNames []string) error { // 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) } @@ -230,8 +226,6 @@ func (s *Subscriber) readMessage(ctx context.Context, msgChan chan *Message) err return ctx.Err() case ack = <-msg.ack: } - //ack := <-msg.ack - ackMessage := server.Nack if ack { ackMessage = server.Ack diff --git a/server/server.go b/server/server.go index c5c7f2c..420b8a1 100644 --- a/server/server.go +++ b/server/server.go @@ -17,7 +17,7 @@ import ( ) // Action represents the type of action that a peer requests to do -type Action uint8 +type Action uint16 const ( Subscribe Action = 1 @@ -28,7 +28,7 @@ const ( ) // Status represents the status of a request -type Status uint8 +type Status uint16 const ( Subscribed = 1 @@ -445,28 +445,23 @@ func dataLength(conn net.Conn) (uint32, error) { } 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)) - if message == "" { - return - } + headers := statusB - 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 + if len(message) > 0 { + sizeB := make([]byte, 4) + binary.BigEndian.PutUint32(sizeB, uint32(len(message))) + headers = append(headers, sizeB...) } - _, err = conn.Write(msgBytes) + msgBytes := []byte(message) + _, 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 } } diff --git a/server/server_test.go b/server/server_test.go index 9660cc6..c12798b 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -48,17 +48,7 @@ func createConnectionAndSubscribe(t *testing.T, topics []string) net.Conn { 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 @@ -71,6 +61,37 @@ func createConnectionAndSubscribe(t *testing.T, topics []string) net.Conn { 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 @@ -93,23 +114,14 @@ func TestUnsubscribesFromTopic(t *testing.T) { 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)) @@ -140,12 +152,8 @@ func TestSubscriberClosesWithoutUnsubscribing(t *testing.T) { 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) @@ -158,7 +166,7 @@ func TestInvalidAction(t *testing.T) { 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 @@ -240,18 +248,7 @@ func TestSendsDataToTopicSubscribers(t *testing.T) { 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 { @@ -335,18 +332,7 @@ func TestPublishMultipleTimes(t *testing.T) { // 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 { @@ -371,18 +357,7 @@ func TestSendsDataToTopicSubscriberNacksThenAcks(t *testing.T) { 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) { @@ -436,18 +411,7 @@ func TestSendsDataToTopicSubscriberDoesntAckMessage(t *testing.T) { 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) { @@ -505,18 +469,7 @@ func TestSendsDataToTopicSubscriberDeliveryCountTooHighWithNoAck(t *testing.T) { 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) { diff --git a/server/subscriber.go b/server/subscriber.go index 344baea..8fe8981 100644 --- a/server/subscriber.go +++ b/server/subscriber.go @@ -87,24 +87,19 @@ func (s *subscriber) addMessage(msg message, delay time.Duration) { 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) }