facade done

Author: bobturanov

cmd/srv-verification-api-facade/main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/ozonmp/srv-verification-api/internal/app/reader"
+	"github.com/ozonmp/srv-verification-api/internal/config"
+	"github.com/ozonmp/srv-verification-api/internal/pkg/logger"
+	"github.com/ozonmp/srv-verification-api/internal/tracer"
+)
+
+func main() {
+	sigs := make(chan os.Signal, 1)
+	ctx := context.Background()
+
+	if err := config.ReadConfigYML("config.yml"); err != nil {
+		logger.FatalKV(ctx, "Failed init configuration", "err", err)
+	}
+
+	cfgInst := config.GetConfigInstance()
+
+	tracing, err := tracer.NewTracer(&cfgInst)
+	if err != nil {
+		logger.ErrorKV(ctx, "Failed init tracing", "err", err)
+		return
+	}
+	defer tracing.Close()
+
+	initCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
+	defer cancel()
+
+	kafka, err := reader.NewEventConsumer(cfgInst.Kafka.Brokers, cfgInst.Kafka.Topic, cfgInst.Kafka.GroupID)
+	if err != nil {
+		logger.FatalKV(ctx, "Failed init reader", "err", err)
+	}
+
+	kafka.Read(initCtx)
+
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+
+	<-sigs
+
+}

internal/app/reader/reader.go

@@ -0,0 +1,84 @@
+package reader
+
+import (
+	"context"
+	"log"
+	"sync"
+
+	"github.com/Shopify/sarama"
+)
+
+type EventReader interface {
+	Read() error
+}
+
+type kafkaReader struct {
+	consumer sarama.ConsumerGroup
+	topic    []string
+	groupId  string
+}
+
+type Consumer struct {
+	ready chan bool
+}
+
+func NewEventConsumer(brokers []string, topic string, groupId string) (*kafkaReader, error) {
+	config := sarama.NewConfig()
+
+	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
+	config.Consumer.Offsets.Initial = sarama.OffsetOldest
+	config.Consumer.Return.Errors = true
+
+	consumer, err := sarama.NewConsumerGroup(brokers, "consumer_group_1", config)
+
+	return &kafkaReader{
+		consumer: consumer,
+		topic:    []string{topic},
+		groupId:  groupId,
+	}, err
+}
+
+func (kr *kafkaReader) Read(ctx context.Context) {
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	consumer := Consumer{
+		ready: make(chan bool),
+	}
+
+	go func() {
+		defer wg.Done()
+		for {
+
+			if err := kr.consumer.Consume(ctx, kr.topic, &consumer); err != nil {
+				log.Panicf("Error from consumer: %v", err)
+			}
+
+			consumer.ready = make(chan bool)
+		}
+	}()
+
+}
+
+func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
+	// Mark the consumer as ready
+	close(consumer.ready)
+	return nil
+}
+
+// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
+func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
+	return nil
+}
+
+// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
+func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+
+	for message := range claim.Messages() {
+		log.Printf(string(message.Value))
+		log.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
+		session.MarkMessage(message, "")
+	}
+
+	return nil
+}

internal/app/repo/method_lock.go

@@ -33,21 +33,21 @@ func (r repo) Lock(ctx context.Context, n uint64) ([]model.VerificationEvent, er
 }
 
 func (r repo) convertToVerificationEventModel(n uint64, eventsData []struct {
-	EventId          uint64 `db:"event_id"`
-	EventType        model.EventType `db:"type"`
+	EventId          uint64            `db:"event_id"`
+	EventType        model.EventType   `db:"type"`
 	EventStatus      model.EventStatus `db:"status"`
-	VerificationId   uint64 `db:"id"`
-	VerificationName string `db:"name"`
+	VerificationId   uint64            `db:"id"`
+	VerificationName string            `db:"name"`
 }) []model.VerificationEvent {
 
 	events := make([]model.VerificationEvent, 0, n)
 
 	for _, event := range eventsData {
 		events = append(events, model.VerificationEvent{
-			ID:     event.EventId,
+			ID:             event.EventId,
 			VerificationID: event.VerificationId,
-			Type:   event.EventType,
-			Status: event.EventStatus,
+			Type:           event.EventType,
+			Status:         event.EventStatus,
 			Entity: &model.Verification{
 				ID:   event.VerificationId,
 				Name: event.VerificationName,
@@ -58,11 +58,11 @@ func (r repo) convertToVerificationEventModel(n uint64, eventsData []struct {
 }
 
 func (r repo) getEventsDataFromDB(ctx context.Context, err error, eventIds []uint64) ([]struct {
-	EventId          uint64 `db:"event_id"`
-	EventType        model.EventType `db:"type"`
+	EventId          uint64            `db:"event_id"`
+	EventType        model.EventType   `db:"type"`
 	EventStatus      model.EventStatus `db:"status"`
-	VerificationId   uint64 `db:"id"`
-	VerificationName string `db:"name"`
+	VerificationId   uint64            `db:"id"`
+	VerificationName string            `db:"name"`
 }, error) {
 
 	query, args, err := squirrel.Select("verification_events.event_id",
@@ -81,11 +81,11 @@ func (r repo) getEventsDataFromDB(ctx context.Context, err error, eventIds []uin
 	}
 
 	var eventsData []struct {
-		EventId          uint64 `db:"event_id"`
-		EventType        model.EventType `db:"type"`
+		EventId          uint64            `db:"event_id"`
+		EventType        model.EventType   `db:"type"`
 		EventStatus      model.EventStatus `db:"status"`
-		VerificationId   uint64 `db:"id"`
-		VerificationName string `db:"name"`
+		VerificationId   uint64            `db:"id"`
+		VerificationName string            `db:"name"`
 	}
 
 	err = r.db.SelectContext(ctx, &eventsData, query, args...)