package authPostman import ( "context" "encoding/json" "fmt" "log" "net" "strconv" "git.slaventius.ru/test3k/authPostman/internal/config" kafka "git.slaventius.ru/test3k/authPostman/internal/transport/kafka" // telegram "git.slaventius.ru/test3k/authPostman/internal/transport/telegram" smtp "git.slaventius.ru/test3k/authPostman/internal/transport/smtp" api "git.slaventius.ru/test3k/umate/pkg/kafka" ) type AuthPostmanServer struct { ctx context.Context kafkaReader *kafka.KafkaReader config *config.Config } func NewServer(ctx context.Context, config *config.Config) *AuthPostmanServer { address := []string{net.JoinHostPort(config.Kafka.Host, strconv.Itoa(config.Kafka.Port))} return &AuthPostmanServer{ ctx: ctx, kafkaReader: kafka.NewReader(ctx, config, api.TopicRegistrations, address...), config: config, } } func (s *AuthPostmanServer) GracefulStop() error { return s.kafkaReader.Close() } func (s *AuthPostmanServer) ReadMessage(offset int64) error { // ... // s.kafkaReader.SetOffset(offset) // for { m, err := s.kafkaReader.ReadMessage() if err != nil { return err } // Декодируем сообщение msg := api.MessageRegistration{} erk := json.Unmarshal(m.Value, &msg) if erk != nil { return erk } // log.Printf("send code %s to %s ...", msg.Code, msg.Email) // text := fmt.Sprintf("Confirmation code %v", msg.Code) // postman := telegram.NewService(s.config.Telegram.ChatToken) // message := postman.NewMessage(s.config.Telegram.ChatID, text) // ers := postman.SendMessage(message) // if ers != nil { // log.Print(ers) // } else { // log.Printf("send code %s to %s completed", amsg.Code, amsg.Email) // } // message := smtp.NewMessage("Confirmation code", text) message.AppendRecipient(msg.Email) // smtpSender := smtp.NewService(s.config.Smtp.Host, s.config.Smtp.Port, s.config.Smtp.Sender, s.config.Smtp.Password) ers := smtpSender.Send(message) if ers != nil { log.Print(ers) } else { log.Printf("send code %s to %s completed", msg.Code, msg.Email) } } }