57 lines
1.3 KiB
Go
57 lines
1.3 KiB
Go
package tariff
|
|
|
|
import (
|
|
"codeword/internal/models"
|
|
"codeword/internal/utils/transfer"
|
|
"context"
|
|
"github.com/twmb/franz-go/pkg/kgo"
|
|
"go.uber.org/zap"
|
|
"google.golang.org/protobuf/proto"
|
|
"log"
|
|
)
|
|
|
|
type ProducerDeps struct {
|
|
Logger *zap.Logger
|
|
Client *kgo.Client
|
|
Topic string
|
|
}
|
|
|
|
type Producer struct {
|
|
logger *zap.Logger
|
|
client *kgo.Client
|
|
topic string
|
|
}
|
|
|
|
func NewProducer(deps ProducerDeps) *Producer {
|
|
if deps.Logger == nil {
|
|
log.Panicln("logger is nil on <NewTariffProducer>")
|
|
}
|
|
|
|
if deps.Client == nil {
|
|
log.Panicln("Kafka client is nil on <NewTariffProducer>")
|
|
}
|
|
|
|
return &Producer{
|
|
logger: deps.Logger,
|
|
client: deps.Client,
|
|
topic: deps.Topic,
|
|
}
|
|
}
|
|
|
|
func (p *Producer) Send(ctx context.Context, userID string, tariff *models.Tariff) error {
|
|
bytes, err := proto.Marshal(transfer.TariffModelToProtoMessage(userID, tariff))
|
|
if err != nil {
|
|
p.logger.Error("failed to marshal tariff model", zap.Error(err))
|
|
return err
|
|
}
|
|
|
|
// упростил, возможно зря, но теперь возвращаем одну ошибку, просто прерываем цикл при первой встретившейся ошибке
|
|
err = p.client.ProduceSync(ctx, &kgo.Record{Topic: p.topic, Value: bytes}).FirstErr()
|
|
if err != nil {
|
|
p.logger.Error("failed to send tariff to Kafka", zap.Error(err))
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|