Update for project structure

constants/constants.go

@@ -0,0 +1,3 @@
+package constants
+
+const GroupProtocol = "roundrobin"

consumer.go → consumer/consumer.go

@@ -1,26 +1,26 @@
-package kafka
+package consumer
 
 import (
 	"context"
 	"os"
 	"time"
 
+	"github.com/teamseodo/kafka-do/constants"
+	"github.com/teamseodo/kafka-do/model"
 	"github.com/twmb/franz-go/pkg/kgo"
 )
 
-type Message []byte
-
 type Consumer struct {
 	client *kgo.Client
 }
 
-func NewConsumer(groupName string, topics []string, brokers []string, logger bool) (*Consumer, error) {
+func New(groupName string, topics []string, brokers []string, logger bool) (*Consumer, error) {
 	opts := []kgo.Opt{
 		kgo.SeedBrokers(brokers...),
 		kgo.ConsumerGroup(groupName),
 		kgo.ConsumeTopics(topics...),
 		kgo.DisableAutoCommit(),
-		kgo.GroupProtocol("roundrobin"),
+		kgo.GroupProtocol(constants.GroupProtocol),
 	}
 
 	if logger {
@@ -37,8 +37,8 @@ func NewConsumer(groupName string, topics []string, brokers []string, logger boo
 	}, nil
 }
 
-func (c *Consumer) ConsumeBatch(ctx context.Context, batchSize int) []Message {
-	var messages []Message
+func (c *Consumer) ConsumeBatch(ctx context.Context, batchSize int) []model.Message {
+	var messages []model.Message
 
 	for batchSize > 0 {
 		timeout, cancel := context.WithTimeout(ctx, time.Minute*1)

consumer_test.go → consumer/consumer_test.go

@@ -1,39 +1,42 @@
-package kafka
+package consumer
 
 import (
 	"context"
 	"testing"
+
+	"github.com/teamseodo/kafka-do/model"
+	"github.com/teamseodo/kafka-do/producer"
 )
 
 func TestConsumeBatch(t *testing.T) {
 	tests := []struct {
 		name     string
-		messages []Message
+		messages []model.Message
 	}{
 		{
 			name: "should-got-message",
-			messages: []Message{
-				Message("message 1"),
+			messages: []model.Message{
+				model.Message("message 1"),
 			},
 		},
 		{
 			name: "should-got-all-messages",
-			messages: []Message{
-				Message("message 1"), Message("message 2"), Message("message 3"), Message("message 4"), Message("message 5"),
+			messages: []model.Message{
+				model.Message("message 1"), model.Message("message 2"), model.Message("message 3"), model.Message("message 4"), model.Message("message 5"),
 			},
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			producer, err := NewProducer("127.0.0.1:9092")
+			producer, err := producer.New("127.0.0.1:9092")
 			if err != nil {
 				t.Fatal(err)
 			}
 			defer producer.Close()
 			producer.Produce(context.Background(), test.messages, "kafka_do_test")
 
-			consumer, err := NewConsumer("kafka_do", []string{"kafka_do_test"}, []string{"127.0.0.1:9092"}, false)
+			consumer, err := New("kafka_do", []string{"kafka_do_test"}, []string{"127.0.0.1:9092"}, false)
 			if err != nil {
 				t.Fatal(err)
 			}

model/message.go

@@ -0,0 +1,3 @@
+package model
+
+type Message []byte

producer.go → producer/producer.go

@@ -1,16 +1,17 @@
-package kafka
+package producer
 
 import (
 	"context"
 
+	"github.com/teamseodo/kafka-do/model"
 	"github.com/twmb/franz-go/pkg/kgo"
 )
 
 type Producer struct {
 	client *kgo.Client
 }
 
-func NewProducer(brokers ...string) (*Producer, error) {
+func New(brokers ...string) (*Producer, error) {
 	cl, err := kgo.NewClient(
 		kgo.SeedBrokers(brokers...),
 	)
@@ -23,7 +24,7 @@ func NewProducer(brokers ...string) (*Producer, error) {
 	}, nil
 }
 
-func (p *Producer) Produce(ctx context.Context, messages []Message, topic string) kgo.ProduceResults {
+func (p *Producer) Produce(ctx context.Context, messages []model.Message, topic string) kgo.ProduceResults {
 	var records []*kgo.Record
 
 	for _, message := range messages {

producer_test.go → producer/producer_test.go

@@ -1,40 +1,43 @@
-package kafka
+package producer
 
 import (
 	"context"
 	"testing"
+
+	"github.com/teamseodo/kafka-do/consumer"
+	"github.com/teamseodo/kafka-do/model"
 )
 
 func TestProduce(t *testing.T) {
 	tests := []struct {
 		name     string
-		messages []Message
+		messages []model.Message
 	}{
 		{
 			name: "should-got-message",
-			messages: []Message{
-				Message("message 1"),
+			messages: []model.Message{
+				model.Message("message 1"),
 			},
 		},
 		{
 			name: "should-got-all-messages",
-			messages: []Message{
-				Message("message 1"), Message("message 2"), Message("message 3"), Message("message 4"), Message("message 5"),
+			messages: []model.Message{
+				model.Message("message 1"), model.Message("message 2"), model.Message("message 3"), model.Message("message 4"), model.Message("message 5"),
 			},
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			producer, err := NewProducer("127.0.0.1:9092")
+			producer, err := New("127.0.0.1:9092")
 			if err != nil {
 				t.Fatal(err)
 			}
 			defer producer.Close()
 
 			producer.Produce(context.Background(), test.messages, "kafka_do_test")
 
-			consumer, err := NewConsumer("kafka_do", []string{"kafka_do_test"}, []string{"127.0.0.1:9092"}, false)
+			consumer, err := consumer.New("kafka_do", []string{"kafka_do_test"}, []string{"127.0.0.1:9092"}, false)
 			if err != nil {
 				t.Fatal(err)
 			}