refactor: moving DSN parsing from factory to transport for kafka

Author: Gerfey

transport/kafka/config.go

@@ -16,5 +16,4 @@ type OptionsConfig struct {
 	Offset           string        `yaml:"offset"             default:"latest"`
 	Group            string        `yaml:"group"              default:"group"`
 	Topic            string        `yaml:"topic"              default:"topic"`
-	Brokers          []string      `yaml:"brokers"`
 }

transport/kafka/factory.go

@@ -3,7 +3,6 @@ package kafka
 import (
 	"fmt"
 	"log/slog"
-	"net/url"
 	"strings"
 
 	"github.com/creasty/defaults"
@@ -38,13 +37,6 @@ func (t *TransportFactory) Create(name string, dsn string, options []byte) (api.
 		return nil, fmt.Errorf("kafka: unmarshal options: %w", err)
 	}
 
-	u, err := url.Parse(dsn)
-	if err != nil {
-		return nil, fmt.Errorf("kafka: failed to parse dsn: %w", err)
-	}
-
-	optsConfig.Brokers = strings.Split(u.Host, ",")
-
 	tCfg := TransportConfig{
 		Name:    name,
 		DSN:     dsn,

transport/kafka/transport.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"net/url"
+	"strings"
 
 	"github.com/gerfey/messenger/api"
 	"github.com/gerfey/messenger/serializer"
@@ -19,7 +21,14 @@ type Transport struct {
 }
 
 func NewTransport(cfg TransportConfig, resolver api.TypeResolver, logger *slog.Logger) (api.Transport, error) {
-	conn, err := NewConnection(cfg.Options.Brokers)
+	u, err := url.Parse(cfg.DSN)
+	if err != nil {
+		return nil, fmt.Errorf("kafka: failed to parse dsn: %w", err)
+	}
+
+	brokers := strings.Split(u.Host, ",")
+
+	conn, err := NewConnection(brokers)
 	if err != nil {
 		logger.Error("failed to connect to Kafka brokers", "error", err)