Get non-docker-compose commands working (#13)

* Get non-docker-compose commands working

* Slight cleanups

Makefile

@@ -1,30 +1,39 @@
 BUFSTREAM_VERSION := 0.1.0
 
-.DEFAULT_GOAL := run-docker-compose
+.DEFAULT_GOAL := docker-compose-run
 
 ## Demo run commands
 
-.PHONY: run-docker-compose
-run-docker-compose: # Run the demo within docker compose.
+.PHONY: docker-compose-run
+docker-compose-run: # Run the demo within docker compose.
 	docker compose up --build
 
-.PHONY: run-bufstream
-run-bufstream: # Run Bufstream within Docker.
-	docker run --rm -p "9092:9092" -v "./config/bufstream.yaml:/bufstream.yaml" \
-		us-docker.pkg.dev/buf-images-1/bufstream-public/images/bufstream:$(BUFSTREAM_VERSION) -c "/bufstream.yaml"
+.PHONY: docker-bufstream-run
+docker-bufstream-run: # Run Bufstream within Docker.
+	docker run --rm -p 9092:9092 -v ./config/bufstream.yaml:/bufstream.yaml \
+		"us-docker.pkg.dev/buf-images-1/bufstream-public/images/bufstream:$(BUFSTREAM_VERSION)" \
+			--config /bufstream.yaml
 
-.PHONY: run-consume
-run-consume: # Run the demo consumer within Docker.
+.PHONY: docker-produce-run
+docker-produce-run: # Run the demo producer within Docker. If you have Go installed, you can call produce-run.
+	docker build -t bufstream/demo-produce -f Dockerfile.produce .
+	docker run --rm --network=host bufstream/demo-produce
+
+.PHONY: docker-consume-run
+docker-consume-run: # Run the demo consumer within Docker. If you have Go installed, you can call consume-run.
 	docker build -t bufstream/demo-consume -f Dockerfile.consume .
-	docker run --rm bufstream/demo-consume
+	docker run --rm --network=host bufstream/demo-consume
 
-.PHONY: run-produce
-run-produce: # Run the demo producer within Docker.
-	docker build -t bufstream/demo-produce -f Dockerfile.produce .
-	docker run --rm bufstream/demo-produce
+.PHONY: produce-run
+produce-run: # Run the demo producer. Go must be installed.
+	go run ./cmd/bufstream-demo-produce
+
+.PHONY: consume-run
+consume-run: # Run the demo consumer. Go must be installed.
+	go run ./cmd/bufstream-demo-consume
 
-.PHONY: clean-docker-compose
-clean-docker-compose: # Cleanup docker compose assets.
+.PHONY: docker-compose-clean
+docker-compose-clean: # Cleanup docker compose assets.
 	docker compose down --rmi all
 
 ## Development commands

cmd/bufstream-demo-consume/main.go

@@ -47,6 +47,7 @@ func run(ctx context.Context, config app.Config) error {
 		consume.WithMessageHandler(handleEmailUpdated),
 	)
 
+	slog.Info("starting consume")
 	for {
 		// Read as many messages as we can.
 		//

cmd/bufstream-demo-produce/main.go

@@ -13,6 +13,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"time"
 
@@ -52,11 +53,15 @@ func run(ctx context.Context, config app.Config) error {
 		config.Kafka.Topic,
 	)
 
+	slog.Info("starting produce")
 	for {
 		id := newID()
 		// Produces semantically-valid EmailUpdated message, where both email
 		// fields are valid email addresses.
 		if err := producer.ProduceProtobufMessage(ctx, id, newSemanticallyValidEmailUpdated(id)); err != nil {
+			if errors.Is(err, context.Canceled) {
+				return err
+			}
 			slog.Error("error on produce or semantically valid protobuf message", "error", err)
 		} else {
 			slog.Info("produced semantically valid protobuf message", "id", id)
@@ -65,13 +70,19 @@ func run(ctx context.Context, config app.Config) error {
 		// Produces a semantically-invalid EmailUpdated message, where the new email field
 		// is not a valid email address.
 		if err := producer.ProduceProtobufMessage(ctx, id, newSemanticallyInvalidEmailUpdated(id)); err != nil {
+			if errors.Is(err, context.Canceled) {
+				return err
+			}
 			slog.Error("error on produce of semantically invalid protobuf message", "error", err)
 		} else {
 			slog.Info("produced semantically invalid protobuf message", "id", id)
 		}
 		id = newID()
 		// Produces record containing a payload that is not valid Protobuf.
 		if err := producer.ProduceInvalid(ctx, id); err != nil {
+			if errors.Is(err, context.Canceled) {
+				return err
+			}
 			slog.Error("error on produce of invalid data", "error", err)
 		} else {
 			slog.Info("produced invalid data", "id", id)

config/bufstream.yaml

@@ -7,7 +7,7 @@ kafka:
     host: 0.0.0.0
     port: 9092
   public_address:
-    host: bufstream
+    host: 0.0.0.0
     port: 9092
 observability:
   trace_exporter: NONE

docker-compose.yaml

@@ -19,7 +19,10 @@ services:
     # Edit config/bufstream.yaml within this repository to change configuation.
     volumes:
       - "./config/bufstream.yaml:/bufstream.yaml"
-    command: [ "-c", "/bufstream.yaml" ]
+    command: [
+      "--config", "/bufstream.yaml",
+      "--config.kafka.public_address.host", "bufstream",
+    ]
   # The demo consumer.
   #
   # This is a Docker image that just runs the binary created from cmd/bufstream-demo-consume.