feat: support kafka scram queue config

server/internal/library/queue/kafkamq.go

@@ -7,13 +7,20 @@ package queue
 
 import (
 	"context"
+	"crypto/sha256"
+	"crypto/sha512"
 	"fmt"
+	"hash"
+	"strings"
+	"time"
+
 	"github.com/IBM/sarama"
 	"github.com/gogf/gf/v2/errors/gerror"
 	"github.com/gogf/gf/v2/frame/g"
+	"github.com/xdg-go/scram"
+
 	"hotgo/internal/consts"
 	"hotgo/utility/simple"
-	"time"
 )
 
 type KafkaMq struct {
@@ -31,6 +38,29 @@ type KafkaConfig struct {
 	Version     string
 	UserName    string
 	Password    string
+	Mechanism   string
+}
+
+type xdgSCRAMClient struct {
+	*scram.ClientConversation
+	HashGeneratorFcn scram.HashGeneratorFcn
+}
+
+func (x *xdgSCRAMClient) Begin(userName, password, authzID string) error {
+	client, err := x.HashGeneratorFcn.NewClient(userName, password, authzID)
+	if err != nil {
+		return err
+	}
+	x.ClientConversation = client.NewConversation()
+	return nil
+}
+
+func (x *xdgSCRAMClient) Step(challenge string) (string, error) {
+	return x.ClientConversation.Step(challenge)
+}
+
+func (x *xdgSCRAMClient) Done() bool {
+	return x.ClientConversation.Done()
 }
 
 // SendMsg 按字符串类型生产数据
@@ -134,11 +164,7 @@ func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error)
 	conf := sarama.NewConfig()
 	conf.Consumer.Return.Errors = true
 	conf.Version = kfkVersion
-	if connOpt.UserName != "" {
-		conf.Net.SASL.Enable = true
-		conf.Net.SASL.User = connOpt.UserName
-		conf.Net.SASL.Password = connOpt.Password
-	}
+	configureKafkaSASL(conf, connOpt)
 
 	// 默认按随机方式消费
 	conf.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
@@ -191,11 +217,7 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
 	conf.ClientID = connOpt.ClientId
 
 	conf.Version = kfkVersion
-	if connOpt.UserName != "" {
-		conf.Net.SASL.Enable = true
-		conf.Net.SASL.User = connOpt.UserName
-		conf.Net.SASL.Password = connOpt.Password
-	}
+	configureKafkaSASL(conf, connOpt)
 
 	mqIns.producerIns, err = sarama.NewAsyncProducer(brokers, conf)
 	if err != nil {
@@ -209,6 +231,35 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
 	return
 }
 
+func configureKafkaSASL(conf *sarama.Config, connOpt KafkaConfig) {
+	if connOpt.UserName == "" {
+		return
+	}
+	conf.Net.SASL.Enable = true
+	conf.Net.SASL.User = connOpt.UserName
+	conf.Net.SASL.Password = connOpt.Password
+	mechanism := strings.ToUpper(strings.ReplaceAll(connOpt.Mechanism, "_", "-"))
+	if mechanism == "" {
+		mechanism = sarama.SASLTypeSCRAMSHA512
+	}
+	switch mechanism {
+	case sarama.SASLTypeSCRAMSHA512:
+		conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
+		conf.Net.SASL.SCRAMClientGeneratorFunc = scramClientGenerator(sha512.New)
+	case sarama.SASLTypeSCRAMSHA256:
+		conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA256
+		conf.Net.SASL.SCRAMClientGeneratorFunc = scramClientGenerator(sha256.New)
+	default:
+		conf.Net.SASL.Mechanism = sarama.SASLMechanism(mechanism)
+	}
+}
+
+func scramClientGenerator(hashFunc func() hash.Hash) func() sarama.SCRAMClient {
+	return func() sarama.SCRAMClient {
+		return &xdgSCRAMClient{HashGeneratorFcn: scram.HashGeneratorFcn(hashFunc)}
+	}
+}
+
 // validateVersion 验证版本是否有效
 func validateVersion(version sarama.KafkaVersion) bool {
 	for _, item := range sarama.SupportedVersions {

server/internal/library/queue/queue.go

@@ -57,6 +57,9 @@ type RocketmqConf struct {
 type KafkaConf struct {
 	Address       []string `json:"address"`
 	Version       string   `json:"version"`
+	Username      string   `json:"username"`
+	Password      string   `json:"password"`
+	Mechanism     string   `json:"mechanism"`
 	RandClient    bool     `json:"randClient"`
 	MultiConsumer bool     `json:"multiConsumer"`
 }
@@ -121,9 +124,12 @@ func NewProducer(groupName string) (mqClient MqProducer, err error) {
 			return
 		}
 		mqClient, err = RegisterKafkaProducer(KafkaConfig{
-			Brokers: config.Kafka.Address,
-			GroupID: groupName,
-			Version: config.Kafka.Version,
+			Brokers:   config.Kafka.Address,
+			GroupID:   groupName,
+			Version:   config.Kafka.Version,
+			UserName:  config.Kafka.Username,
+			Password:  config.Kafka.Password,
+			Mechanism: config.Kafka.Mechanism,
 		})
 	case "redis":
 		if _, err = g.Redis().Do(ctx, "ping"); err == nil {
@@ -185,10 +191,13 @@ func NewConsumer(groupName string) (mqClient MqConsumer, err error) {
 		}
 
 		mqClient, err = RegisterKafkaMqConsumer(KafkaConfig{
-			Brokers:  config.Kafka.Address,
-			GroupID:  groupName,
-			Version:  config.Kafka.Version,
-			ClientId: clientId,
+			Brokers:   config.Kafka.Address,
+			GroupID:   groupName,
+			Version:   config.Kafka.Version,
+			ClientId:  clientId,
+			UserName:  config.Kafka.Username,
+			Password:  config.Kafka.Password,
+			Mechanism: config.Kafka.Mechanism,
 		})
 	case "redis":
 		if _, err = g.Redis().Do(ctx, "ping"); err == nil {