Add Segmentio implementation

Author: Hugo Rojas

cmd/consumer/main.go

@@ -8,6 +8,7 @@ import (
 	"github.com/hrojas2021/go-kafka-mongodb/pkg/iface"
 	"github.com/hrojas2021/go-kafka-mongodb/pkg/kafka/confluentic"
 	"github.com/hrojas2021/go-kafka-mongodb/pkg/kafka/sarama"
+	"github.com/hrojas2021/go-kafka-mongodb/pkg/kafka/segmentio"
 )
 
 func main() {
@@ -16,7 +17,7 @@ func main() {
 	if err != nil {
 		log.Fatal("unable to connecto to mongoDB ", err)
 	}
-	// handler, err := sarama.NewConsumerHandler(cf, db)
+
 	handler, err := getBroker(cf, db)
 	if err != nil {
 		log.Fatal("unable to create a kafka consumer handler ", err)
@@ -45,8 +46,10 @@ func getBroker(cf *config.Configuration, db *database.DB) (iface.ConsumerHandler
 	switch cf.BROKER {
 	case config.Sarama:
 		broker, err = sarama.NewConsumerHandler(cf, db)
-	default:
+	case config.Confluentic:
 		broker, err = confluentic.NewConsumerHandler(cf, db)
+	default:
+		broker, err = segmentio.NewConsumerHandler(cf, db)
 	}
 
 	return broker, err

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/Shopify/sarama v1.38.1
 	github.com/confluentinc/confluent-kafka-go v1.9.2
 	github.com/gorilla/mux v1.8.0
+	github.com/segmentio/kafka-go v0.4.39
 	github.com/spf13/viper v1.15.0
 	go.mongodb.org/mongo-driver v1.11.2
 )

go.sum

@@ -204,6 +204,7 @@ github.com/juju/qthttptest v0.1.1/go.mod h1:aTlAv8TYaflIiTDIQYzxnl1QdPjAg8Q8qJME
 github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
+github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/klauspost/compress v1.15.14 h1:i7WCKDToww0wA+9qrUZ1xOjp218vfFo3nTU6UHp+gOc=
 github.com/klauspost/compress v1.15.14/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
@@ -232,6 +233,7 @@ github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJ
 github.com/nrwiersma/avro-benchmarks v0.0.0-20210913175520-21aec48c8f76/go.mod h1:iKyFMidsk/sVYONJRE372sJuX/QTRPacU7imPqqsu7g=
 github.com/pelletier/go-toml/v2 v2.0.6 h1:nrzqCb7j9cDFj2coyLNLaZuJTLjWjlaz6nvTvIwycIU=
 github.com/pelletier/go-toml/v2 v2.0.6/go.mod h1:eumQOmlWiOPt5WriQQqoM5y18pDHwha2N+QD+EUNTek=
+github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pierrec/lz4/v4 v4.1.17 h1:kV4Ip+/hUBC+8T6+2EgburRtkE9ef4nbY3f4dFhGjMc=
 github.com/pierrec/lz4/v4 v4.1.17/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
@@ -250,6 +252,8 @@ github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTE
 github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8=
 github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
 github.com/santhosh-tekuri/jsonschema/v5 v5.0.0/go.mod h1:FKdcjfQW6rpZSnxxUvEA5H/cDPdvJ/SZJQLWWXWGrZ0=
+github.com/segmentio/kafka-go v0.4.39 h1:75smaomhvkYRwtuOwqLsdhgCG30B82NsbdkdDfFbvrw=
+github.com/segmentio/kafka-go v0.4.39/go.mod h1:T0MLgygYvmqmBvC+s8aCcbVNfJN4znVne5j0Pzowp/Q=
 github.com/spf13/afero v1.9.3 h1:41FoI0fD7OR7mGcKE/aOiLkGreyf8ifIOQmJANWogMk=
 github.com/spf13/afero v1.9.3/go.mod h1:iUV7ddyEEZPO5gA3zD4fJt6iStLlL+Lg4m2cihcDf8Y=
 github.com/spf13/cast v1.5.0 h1:rj3WzYc11XZaIZMPKmwP96zkFEnnAmV8s6XbB2aY32w=
@@ -286,6 +290,10 @@ github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3k
 github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgklLGvcBnW8=
 github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
 github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
+github.com/xdg/scram v1.0.5 h1:TuS0RFmt5Is5qm9Tm2SoD89OPqe4IRiFtyFY4iwWXsw=
+github.com/xdg/scram v1.0.5/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
+github.com/xdg/stringprep v1.0.3 h1:cmL5Enob4W83ti/ZHuZLuKD/xqJfus4fVPwE+/BDm+4=
+github.com/xdg/stringprep v1.0.3/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA=
 github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -381,6 +389,7 @@ golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20220706163947-c90051bbdb60/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220725212005-46097bf591d3/go.mod h1:AaygXjzTFtRAg2ttMY5RMuhpJ3cNnI0XpyFJD1iQRSM=
 golang.org/x/net v0.5.0 h1:GyT4nK/YDHSqa1c4753ouYCDajOYKTja9Xb/OHtgvSw=

pkg/kafka/confluentic/producer.go

@@ -66,6 +66,8 @@ func (k *kafkaP) saveJobToKafka(job model.Job) error {
 			Value:          []byte(word),
 		}, nil)
 	}
+
+	// k.Close()
 	log.Println("The job event has been created successfully")
 	return nil
 }

pkg/kafka/sarama/producer.go

@@ -70,7 +70,7 @@ func (k *kafkaP) saveJobToKafka(job model.Job) error {
 	if err != nil {
 		return err
 	}
-
+	// k.Close()
 	log.Printf("The job event has been created in partition %d and offset %d \n", partition, offset)
 	return nil
 }

pkg/kafka/segmentio/consumer.go

@@ -0,0 +1,71 @@
+package segmentio
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/hrojas2021/go-kafka-mongodb/pkg/config"
+	"github.com/hrojas2021/go-kafka-mongodb/pkg/database"
+	"github.com/hrojas2021/go-kafka-mongodb/pkg/iface"
+	"github.com/hrojas2021/go-kafka-mongodb/pkg/model"
+	"github.com/segmentio/kafka-go"
+)
+
+type kafkaC struct {
+	*kafka.Reader
+}
+
+type consumerHandler struct {
+	config *config.Configuration
+	db     *database.DB
+	kafkaC kafkaC
+}
+
+func NewConsumerHandler(cf *config.Configuration, db *database.DB) (iface.ConsumerHandler, error) {
+	r := kafka.NewReader(kafka.ReaderConfig{
+		Brokers: []string{cf.KAFKAURL},
+		Topic:   cf.KAFKATOPIC,
+	})
+
+	return &consumerHandler{
+		db:     db,
+		config: cf,
+		kafkaC: kafkaC{r},
+	}, nil
+}
+
+func (h *consumerHandler) ReadMessagesFromKafka() error {
+	return h.kafkaC.readMessagesFromKafka(h)
+}
+
+func (h *consumerHandler) Close() error {
+	return h.kafkaC.Close()
+}
+
+func (h *consumerHandler) Subscribe() error {
+	return nil
+}
+
+func (k *kafkaC) readMessagesFromKafka(h *consumerHandler) error {
+	fmt.Println("Start reading Kafka messages in SEGMENTIO")
+	var err error
+	for {
+		msg, err := k.ReadMessage(context.Background())
+		if err != nil {
+			break
+		}
+
+		fmt.Printf("\nReceived msg from Kafka partition with key %d: %s\n", msg.Partition, string(msg.Key))
+		var job model.Job
+		err = json.Unmarshal(msg.Value, &job)
+		if err != nil {
+			break
+		}
+		err = h.db.SaveJob(&job)
+		if err != nil {
+			break
+		}
+	}
+	return err
+}

pkg/kafka/segmentio/producer.go

@@ -0,0 +1,80 @@
+package segmentio
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+
+	"github.com/hrojas2021/go-kafka-mongodb/pkg/config"
+	"github.com/hrojas2021/go-kafka-mongodb/pkg/model"
+	"github.com/segmentio/kafka-go"
+)
+
+type producerHandler struct {
+	kafkaP kafkaP
+}
+
+type kafkaP struct {
+	*kafka.Writer
+	*config.Configuration
+	counter int
+}
+
+func NewProducerHandler(cf *config.Configuration) (*producerHandler, error) {
+	w := kafka.NewWriter(kafka.WriterConfig{
+		Brokers: []string{cf.KAFKAURL},
+		Topic:   cf.KAFKATOPIC,
+	})
+
+	return &producerHandler{
+		kafkaP: kafkaP{w, cf, 1},
+	}, nil
+}
+
+// TODO:  move this to handlers folder and call with interface  saveJobToKafka
+func (h *producerHandler) JobsPostHandler(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		log.Fatal("unable to read the payload ", err)
+	}
+	defer r.Body.Close()
+
+	var job model.Job
+	err = json.Unmarshal(body, &job)
+	if err != nil {
+		log.Fatal("unable to unmarshall the body ", err)
+	}
+
+	err = h.kafkaP.saveJobToKafka(job)
+	if err != nil {
+		log.Fatal("unable to produce the event  ", err)
+	}
+
+	w.Header().Set("content-type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	w.Write(body)
+}
+
+func (k *kafkaP) saveJobToKafka(job model.Job) error {
+	d, err := json.Marshal(job)
+	if err != nil {
+		return err
+	}
+
+	msg := kafka.Message{
+		Key:   []byte(fmt.Sprintf("key-%d", k.counter)),
+		Value: d,
+	}
+
+	err = k.WriteMessages(context.Background(), msg)
+	if err != nil {
+		return err
+	}
+
+	// k.Close()
+
+	return nil
+}