Consuming Messages with the CLI

1. The Consumer's Role

A consumer subscribes to one or more topics and processes the stream of messages published to them. It's the counterpart to the producer. The broker manages the delivery of messages from the topic's partitions to the connected consumers.

Our goal is to create a command that looks like this:
minikafka-cli consume --topic <topic-name> --offset <offset>

2. Defining the consume Command

As before, we'll start by defining the new command and its flags using Cobra. This process should feel familiar after building the topic and produce commands.

Create a new file in your minikafka-cli/cmd/ directory named consume.go.

We'll add flags for the topic, the starting offset, and the broker address. The RunE function will contain the logic to connect and read messages. For now, it will just print the parsed flags.

To guide the structure of this command, you can refer to the "List Command" example in the following article. While it lists static data, the pattern of defining a command, adding flags, and then formatting output in the RunE function is directly applicable here.

Now, add the following code to cmd/consume.go.

// cmd/consume.go
package cmd

import (
	"fmt"
	"os"
	"os/signal"
	"syscall"

	"github.com/spf13/cobra"
)

var (
	consumeTopic  string
	consumeOffset int64
	consumeBroker string
)

// consumeCmd represents the consume command
var consumeCmd = &cobra.Command{
	Use:   "consume",
	Short: "Consume messages from a topic",
	Long: `Connects to the broker and streams messages from a specified topic
starting at a given offset. The command runs continuously until interrupted (Ctrl+C).`,
	RunE: func(cmd *cobra.Command, args []string) error {
		// Set up a channel to listen for OS signals (like Ctrl+C)
		sigchan := make(chan os.Signal, 1)
		signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

		// For now, let's just confirm we've parsed everything correctly.
		fmt.Printf("Consuming from topic: '%s'\n", consumeTopic)
		fmt.Printf("Starting at offset: %d\n", consumeOffset)
		fmt.Printf("Connecting to broker: '%s'\n", consumeBroker)
		fmt.Println("Press Ctrl+C to stop consuming.")

		// In the next step, we'll add the network logic here.
		// The `<-sigchan` will block until a signal is received.
		<-sigchan
		fmt.Println("\nReceived interrupt signal, shutting down consumer.")

		return nil
	},
}

func init() {
	rootCmd.AddCommand(consumeCmd)

	consumeCmd.Flags().StringVarP(&consumeTopic, "topic", "t", "", "Topic to consume messages from (required)")
	consumeCmd.Flags().Int64VarP(&consumeOffset, "offset", "o", 0, "Offset to start consuming from (0 for earliest)")
	consumeCmd.Flags().StringVar(&consumeBroker, "broker", "localhost:9092", "Address of the broker")

	consumeCmd.MarkFlagRequired("topic")
}

Notice a few key details:

• Flags: We use StringVarP for the topic and Int64VarP for the offset, since offsets are numeric.
• Signal Handling: We've preemptively added signal handling. signal.Notify captures SIGINT (Ctrl+C) and SIGTERM, sending them to sigchan. The <-sigchan at the end of RunE makes the command block and wait, keeping the consumer alive until you decide to stop it. This is essential for a long-running streaming command.

Test the new command to ensure it's set up correctly:

# Run from your minikafka-cli directory
go run main.go consume --topic my-first-topic --offset 0

The program should print the flag values and then wait. You can exit by pressing Ctrl+C.

3. Requesting and Reading the Message Stream

This is where our consume command comes to life. The process involves three steps:

1. Serialize a "consume" request according to our binary protocol.
2. Send the request to the broker.
3. Enter a read loop to continuously decode and display the length-prefixed messages sent back by the broker.

First, let's define the binary protocol for a consume request. This should mirror what the broker expects.

Now, replace the RunE function in cmd/consume.go with the complete implementation below. You will need to add some imports as well.

// cmd/consume.go

// Add these to your imports
import (
	"bufio"
	"bytes"
	"context"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"os"
	"os/signal"
	"syscall"

	"github.com/spf13/cobra"
)

// ... (keep your variable and command definitions)

// Replace the old RunE with this new one
var consumeCmd = &cobra.Command{
    // ... (Use, Short, Long are the same)
	RunE: func(cmd *cobra.Command, args []string) error {
		// Use a context for graceful shutdown
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()

		// Set up signal handling to cancel the context
		sigchan := make(chan os.Signal, 1)
		signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
		go func() {
			<-sigchan
			fmt.Println("\nReceived interrupt signal, shutting down consumer...")
			cancel()
		}()

		// 1. Serialize the consume request
		reqBuf := new(bytes.Buffer)
		// Message Type (0x02 for Consume)
		binary.Write(reqBuf, binary.BigEndian, byte(0x02))
		// Topic
		binary.Write(reqBuf, binary.BigEndian, uint16(len(consumeTopic)))
		reqBuf.WriteString(consumeTopic)
		// Offset
		binary.Write(reqBuf, binary.BigEndian, uint64(consumeOffset))

		finalPayload := new(bytes.Buffer)
		binary.Write(finalPayload, binary.BigEndian, uint32(reqBuf.Len()))
		finalPayload.Write(reqBuf.Bytes())

		// 2. Connect to the broker
		conn, err := net.Dial("tcp", consumeBroker)
		if err != nil {
			return fmt.Errorf("failed to connect to broker: %w", err)
		}
		defer conn.Close()

		// 3. Send the request
		if _, err := conn.Write(finalPayload.Bytes()); err != nil {
			return fmt.Errorf("failed to send consume request: %w", err)
		}
		fmt.Printf("Consuming messages from topic '%s'...\n", consumeTopic)

		// 4. Enter the read loop
		go readMessages(ctx, conn)

		// Block until the context is canceled (e.g., by Ctrl+C)
		<-ctx.Done()
		return nil
	},
}

// readMessages runs in a goroutine to continuously read from the connection.
func readMessages(ctx context.Context, conn net.Conn) {
	reader := bufio.NewReader(conn)
	for {
		select {
		case <-ctx.Done():
			// Context was canceled, exit the loop
			return
		default:
			// Read message length (4 bytes)
			lenBuf := make([]byte, 4)
			if _, err := io.ReadFull(reader, lenBuf); err != nil {
				if err != io.EOF {
					fmt.Fprintf(os.Stderr, "Error reading message length: %v\n", err)
				}
				return
			}
			msgLen := binary.BigEndian.Uint32(lenBuf)

			// Read message payload
			payload := make([]byte, msgLen)
			if _, err := io.ReadFull(reader, payload); err != nil {
				if err != io.EOF {
					fmt.Fprintf(os.Stderr, "Error reading message payload: %v\n", err)
				}
				return
			}

			// In a real scenario, you'd deserialize this payload
			// back into a message struct. For now, we'll print it.
			fmt.Printf("Received message: %s\n", string(payload))
		}
	}
}

// ... (init function remains the same)

Dissecting the Read Loop

The readMessages function is the core of this lesson. Let's break it down:

• Graceful Shutdown: The function now accepts a context.Context. The select block allows the goroutine to listen for context cancellation (<-ctx.Done()) while also attempting to perform its primary work. This is a robust pattern for managing goroutine lifecycles. For a deeper look at this pattern, the following resource is excellent.

• Reading Framed Data: The broker sends length-prefixed messages. We can't just read arbitrarily.
  1. io.ReadFull(reader, lenBuf): We first read exactly 4 bytes to get the length of the upcoming payload. Using io.ReadFull is critical as it guarantees a full read or returns an error, preventing partial reads that would corrupt our framing logic.
  2. binary.BigEndian.Uint32(lenBuf): We decode those 4 bytes into an integer.
  3. io.ReadFull(reader, payload): We then read exactly that many bytes to get the full message payload.
• Error Handling: We check for io.EOF (End of File), which is the expected error when the server closes the connection. Other errors indicate a problem with the network stream.

4. Putting It All Together

You're now ready for a full end-to-end test.

1. Make sure your broker server is running.
2. In one terminal, start the consumer:

   go run main.go consume --topic my-logs
   

3. In a second terminal, produce some messages to that same topic:

   go run main.go produce "server started successfully" --topic my-logs
   go run main.go produce "user logged in" --topic my-logs
   go run main.go produce "database query failed" --topic my-logs
   

As you run each produce command, you should see the message appear almost instantly in your consumer's terminal. When you're finished, press Ctrl+C in the consumer terminal to shut it down gracefully.

Conclusion

Excellent work. You have now implemented the consume command, completing the primary produce-and-consume functionality of our distributed messaging system. You can now send a message from one client and see it received by another, all orchestrated by your custom broker.

Key Takeaways:

• Long-Running CLI Commands: You built a command that runs continuously, using signal handling and context for a graceful shutdown.
• Streaming Data Processing: You implemented a read loop to process a continuous stream of data from a TCP connection, a common task in network programming.
• Binary Deserialization: You successfully implemented the client-side logic for a length-prefix framing protocol, correctly decoding messages from a byte stream.
• System Integration: You have witnessed the full, end-to-end message flow of the system you've been building, a major project milestone.

Next Up

With the core client-server functionality in place, our next lesson will shift focus to production-readiness. We will integrate the Prometheus client library to expose a /metrics HTTP endpoint on the broker, allowing us to monitor its internal state and performance.