*

cmd/main.go

@@ -18,7 +18,7 @@ import (
 func main() {
 	config := config.NewConfig()
 	ctx, _ := context.WithCancel(context.Background())
-	srv := server.NewServer()
+	srv := server.NewServer(config)
 	s := grpc.NewServer()
 
 	//

internal/config/config.go

@@ -6,19 +6,19 @@ import (
 	"github.com/kelseyhightower/envconfig"
 )
 
-// type DbConfig struct {
+type appConfig struct {
-// 	Host string `envconfig:"DB_HOST"`
-// 	Port int    `envconfig:"DB_PORT"`
-// }
-
-type AppConfig struct {
 	Port int `envconfig:"APP_PORT"`
 }
 
+type kafkaConfig struct {
+	Host string `envconfig:"KAFKA_HOST"`
+	Port int    `envconfig:"KAFKA_PORT"`
+}
+
 // ...
 type Config struct {
-	// Db  DbConfig
+	App   appConfig
-	App AppConfig
+	Kafka kafkaConfig
 }
 
 func NewConfig() *Config {

internal/transport/grpc/grpc.go

@@ -7,11 +7,16 @@ import (
 	"net"
 	"strconv"
 	"sync"
+	"test3k/authDB/internal/config"
 	kafka "test3k/authDB/internal/transport/kafka"
 	api "test3k/authDB/pkg/api"
 	"time"
 )
 
+const (
+	topicRegistrations string = "registrations" // Топик для регистраций
+)
+
 type user struct {
 	ID        int32
 	Code      string
@@ -29,11 +34,11 @@ type AuthDBServer struct {
 	id int32
 }
 
-func NewServer() *AuthDBServer {
+func NewServer(config *config.Config) *AuthDBServer {
 	return &AuthDBServer{
 		mu:          sync.Mutex{},
 		users:       make(map[string]*user),
-		kafkaWriter: kafka.NewWriter("registrations", net.JoinHostPort("localhost", "9094")),
+		kafkaWriter: kafka.NewWriter(topicRegistrations, net.JoinHostPort(config.Kafka.Host, strconv.Itoa(config.Kafka.Port))),
 		id:          0,
 	}
 }

internal/transport/kafka/kafka_writer.go

@@ -101,7 +101,14 @@ func (s *KafkaWriter) checkTopic() error {
 
 	// Если топика нет, то создадим
 	if _, ok := topics[s.topic]; !ok {
-		return s.createTopic()
+		era := s.createTopic()
+		if era != nil {
+			return era
+		}
+
+		log.Printf("create topic %q\n", s.topic)
+
+		return era
 	}
 
 	return nil