Merge pull request #56 from ShaPoHun/SASL_SSL

support kafka connection type: SASL_SSL

kq/config.go

@@ -12,6 +12,7 @@ type KqConf struct {
 	Brokers     []string
 	Group       string
 	Topic       string
+	CaFile      string `json:",optional"`
 	Offset      string `json:",options=first|last,default=last"`
 	Conns       int    `json:",default=1"`
 	Consumers   int    `json:",default=8"`

kq/queue.go

@@ -2,8 +2,11 @@ package kq
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"io"
 	"log"
+	"os"
 	"time"
 
 	"github.com/segmentio/kafka-go"
@@ -121,6 +124,23 @@ func newKafkaQueue(c KqConf, handler ConsumeHandler, options queueOptions) queue
 			},
 		}
 	}
+	if len(c.CaFile) > 0 {
+		caCert, err := os.ReadFile(c.CaFile)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		caCertPool := x509.NewCertPool()
+		ok := caCertPool.AppendCertsFromPEM(caCert)
+		if !ok {
+			log.Fatal(err)
+		}
+
+		readerConfig.Dialer.TLS = &tls.Config{
+			RootCAs:            caCertPool,
+			InsecureSkipVerify: true,
+		}
+	}
 	consumer := kafka.NewReader(readerConfig)
 
 	return &kafkaQueue{