slaventius 2 years ago
parent cb6a45544b
commit 0bb0a4864e
  1. 23
      internal/transport/grpc/grpc.go
  2. 6
      internal/transport/kafka/kafka_writer.go

@ -2,6 +2,7 @@ package grpc
import (
"context"
"encoding/json"
"errors"
"log"
"net"
@ -17,13 +18,17 @@ const (
topicRegistrations string = "registrations" // Топик для регистраций
)
type msg struct {
Code string
Email string
}
type user struct {
ID int32
Code string
Login string
Password string
Email string
Confirmed bool
msg
}
type AuthDBServer struct {
@ -91,16 +96,24 @@ func (s *AuthDBServer) Registration(ctx context.Context, req *api.RegistrationRe
//
user := &user{
ID: s.id,
Code: code,
Login: req.GetLogin(),
Password: code, // TODO
Email: req.GetEmail(),
Confirmed: false,
msg: msg{
Code: code,
Email: req.GetEmail(),
},
}
s.users[req.Login] = user
// TODO
err := s.kafkaWriter.WriteMessage(user.Login, user.Email)
value, eru := json.Marshal(user.msg)
if eru != nil {
return nil, eru
}
//
err := s.kafkaWriter.WriteMessage([]byte(user.Login), value)
if err != nil {
log.Print(err)

@ -116,9 +116,9 @@ func (s *KafkaWriter) checkTopic() error {
return nil
}
func (s *KafkaWriter) WriteMessage(key string, value string) error {
func (s *KafkaWriter) WriteMessage(key []byte, value []byte) error {
return s.writer.WriteMessages(s.ctx, kafka.Message{
Key: []byte(key),
Value: []byte(value),
Key: key,
Value: value,
})
}

Loading…
Cancel
Save