update

.github/workflows/tests.yml

@@ -6,7 +6,7 @@ on:
 
 jobs:
   test:
-    name: Event-Stream
+    name: event-shark
     runs-on: ubuntu-latest
     env:
       DOCKER_BUILDKIT: 1
@@ -22,4 +22,4 @@ jobs:
       - name: Compose down
         run: docker-compose down --rmi all
       - name: check logs
-        run: docker compose logs -t -f event-stream
+        run: docker compose logs -t -f event-shark

cmd/main.go

@@ -8,9 +8,9 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/dipjyotimetia/event-stream/pkg/config"
-	"github.com/dipjyotimetia/event-stream/pkg/events"
-	"github.com/dipjyotimetia/event-stream/pkg/router"
+	"github.com/dipjyotimetia/event-shark/pkg/config"
+	"github.com/dipjyotimetia/event-shark/pkg/events"
+	"github.com/dipjyotimetia/event-shark/pkg/router"
 	"github.com/goccy/go-json"
 	"github.com/gofiber/fiber/v2"
 	"github.com/gofiber/fiber/v2/middleware/cors"

go.mod

@@ -1,4 +1,4 @@
-module github.com/dipjyotimetia/event-stream
+module github.com/dipjyotimetia/event-shark
 
 go 1.22
 

go.sum

@@ -59,8 +59,6 @@ golang.org/x/crypto v0.19.0 h1:ENy+Az/9Y1vSrlrvBSyna3PITt4tiZLf7sgCjZBX7Wo=
 golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
-golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
 golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

pkg/events/Produce.go

@@ -6,7 +6,7 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/dipjyotimetia/event-stream/pkg/config"
+	"github.com/dipjyotimetia/event-shark/pkg/config"
 	"github.com/hamba/avro/v2"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"github.com/twmb/franz-go/pkg/sr"

pkg/handler/expenseHandler.go

@@ -7,9 +7,9 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/dipjyotimetia/event-stream/gen"
-	"github.com/dipjyotimetia/event-stream/pkg/config"
-	"github.com/dipjyotimetia/event-stream/pkg/events"
+	"github.com/dipjyotimetia/event-shark/gen"
+	"github.com/dipjyotimetia/event-shark/pkg/config"
+	"github.com/dipjyotimetia/event-shark/pkg/events"
 	"github.com/gofiber/fiber/v2"
 )
 

pkg/handler/paymentHandler.go

@@ -5,9 +5,9 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/dipjyotimetia/event-stream/gen"
-	"github.com/dipjyotimetia/event-stream/pkg/config"
-	"github.com/dipjyotimetia/event-stream/pkg/events"
+	"github.com/dipjyotimetia/event-shark/gen"
+	"github.com/dipjyotimetia/event-shark/pkg/config"
+	"github.com/dipjyotimetia/event-shark/pkg/events"
 	"github.com/gofiber/fiber/v2"
 )
 

pkg/router/expenseRouter.go

@@ -3,9 +3,9 @@ package router
 import (
 	"context"
 
-	"github.com/dipjyotimetia/event-stream/pkg/config"
-	"github.com/dipjyotimetia/event-stream/pkg/events"
-	"github.com/dipjyotimetia/event-stream/pkg/handler"
+	"github.com/dipjyotimetia/event-shark/pkg/config"
+	"github.com/dipjyotimetia/event-shark/pkg/events"
+	"github.com/dipjyotimetia/event-shark/pkg/handler"
 	"github.com/gofiber/fiber/v2"
 )
 

script/consumer/main.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+)
+
+func main() {
+	topic := "expense-topic"
+	ctx := context.Background()
+
+	seeds := []string{"localhost:9092"}
+	opts := []kgo.Opt{}
+	opts = append(opts,
+		kgo.SeedBrokers(seeds...),
+		kgo.ConsumeTopics(topic),
+		kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
+	)
+
+	client, err := kgo.NewClient(opts...)
+	if err != nil {
+		panic(err)
+	}
+	defer client.Close()
+
+	for {
+		fetches := client.PollFetches(ctx)
+		if errs := fetches.Errors(); len(errs) > 0 {
+			panic(fmt.Sprint(errs))
+		}
+		iter := fetches.RecordIter()
+		for !iter.Done() {
+			record := iter.Next()
+			topicInfo := fmt.Sprintf("topic: %s (%d|%d)",
+				record.Topic, record.Partition, record.Offset)
+			messageInfo := fmt.Sprintf("key: %s, Value: %s",
+				record.Key, record.Value)
+			fmt.Printf("Message consumed: %s, %s \n", topicInfo, messageInfo)
+		}
+	}
+}

tests/expense_test.go

@@ -10,7 +10,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/dipjyotimetia/event-stream/gen"
+	"github.com/dipjyotimetia/event-shark/gen"
 )
 
 func TestExpenseAPI(t *testing.T) {

tests/payments_test.go

@@ -10,7 +10,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/dipjyotimetia/event-stream/gen"
+	"github.com/dipjyotimetia/event-shark/gen"
 )
 
 func TestPaymentsAPI(t *testing.T) {