refactor: replace Kafka topic configuration with mqbuild constants for improved readability

cmd/main.go

@@ -498,6 +498,7 @@ func startRedisServerRegister(ctx context.Context, cfg *serverConfig, client dis
 			log.ZWarn(ctx, "gateway unregister failed", err, "address", selfAddr)
 		}
 	}()
+	register()
 	for {
 		select {
 		case <-timer.C:

internal/msgtransfer/init.go

@@ -73,11 +73,11 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	mongoProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToMongoTopic)
+	mongoProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToMongo)
 	if err != nil {
 		return err
 	}
-	pushProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToPushTopic)
+	pushProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToPush)
 	if err != nil {
 		return err
 	}
@@ -100,11 +100,11 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	historyConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToRedisTopic)
+	historyConsumer, err := builder.GetTopicConsumer(ctx, mqbuild.TopicToRedis)
 	if err != nil {
 		return err
 	}
-	historyMongoConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToMongoTopic)
+	historyMongoConsumer, err := builder.GetTopicConsumer(ctx, mqbuild.TopicToMongo)
 	if err != nil {
 		return err
 	}

internal/push/push.go

@@ -65,17 +65,17 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	offlinePushProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToOfflinePushTopic)
+	offlinePushProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToOfflinePush)
 	if err != nil {
 		return err
 	}
 	database := controller.NewPushDatabase(cacheModel, offlinePushProducer)
 
-	pushConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToPushTopic)
+	pushConsumer, err := builder.GetTopicConsumer(ctx, mqbuild.TopicToPush)
 	if err != nil {
 		return err
 	}
-	offlinePushConsumer, err := builder.GetTopicConsumer(ctx, config.KafkaConfig.ToOfflinePushTopic)
+	offlinePushConsumer, err := builder.GetTopicConsumer(ctx, mqbuild.TopicToOfflinePush)
 	if err != nil {
 		return err
 	}

internal/rpc/msg/server.go

@@ -92,7 +92,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	redisProducer, err := builder.GetTopicProducer(ctx, config.KafkaConfig.ToRedisTopic)
+	redisProducer, err := builder.GetTopicProducer(ctx, mqbuild.TopicToRedis)
 	if err != nil {
 		return err
 	}

pkg/common/config/mq.go

@@ -14,7 +14,7 @@ const (
 
 func NormalizeQueueEngine(engine string) string {
 	switch strings.ToLower(strings.TrimSpace(engine)) {
-	case "kafka":
+	case "", "kafka":
 		return QueueEngineKafka
 	case "redis":
 		return QueueEngineRedis