[SERF-1593] Add PubSub configuration

pkg/configuration/configuration.go

@@ -5,6 +5,7 @@ import (
 	database "github.com/scribd/go-sdk/pkg/database"
 	instrumentation "github.com/scribd/go-sdk/pkg/instrumentation"
 	logger "github.com/scribd/go-sdk/pkg/logger"
+	"github.com/scribd/go-sdk/pkg/pubsub"
 	server "github.com/scribd/go-sdk/pkg/server"
 	tracking "github.com/scribd/go-sdk/pkg/tracking"
 )
@@ -17,6 +18,7 @@ type Config struct {
 	Logger          *logger.Config
 	Server          *server.Config
 	Tracking        *tracking.Config
+	PubSub          *pubsub.Config
 }
 
 // NewConfig returns a new Config instance
@@ -53,12 +55,18 @@ func NewConfig() (*Config, error) {
 		return config, err
 	}
 
+	pubsubConfig, err := pubsub.NewConfig()
+	if err != nil {
+		return config, err
+	}
+
 	config.App = appConfig
 	config.Database = dbConfig
 	config.Instrumentation = instrumentationConfig
 	config.Logger = loggerConfig
 	config.Server = serverConfig
 	config.Tracking = trackingConfig
+	config.PubSub = pubsubConfig
 
 	return config, nil
 }

pkg/pubsub/config.go

@@ -0,0 +1,55 @@
+package pubsub
+
+import (
+	"fmt"
+	"time"
+
+	cbuilder "github.com/scribd/go-sdk/internal/pkg/configuration/builder"
+)
+
+type (
+	Config struct {
+		Kafka Kafka `mapstructure:"kafka"`
+	}
+
+	Publisher struct {
+		MaxAttempts  int           `mapstructure:"max_attempts"`
+		WriteTimeout time.Duration `mapstructure:"write_timeout"`
+		Topic        string        `mapstructure:"topic"`
+		Enabled      bool          `mapstructure:"enabled"`
+	}
+
+	Subscriber struct {
+		Topic   string `mapstructure:"topic"`
+		GroupId string `mapstructure:"group_id"`
+		Enabled bool   `mapstructure:"enabled"`
+	}
+
+	Kafka struct {
+		BrokerUrls             []string   `mapstructure:"broker_urls"`
+		ClientId               string     `mapstructure:"client_id"`
+		Cert                   string     `mapstructure:"cert_pem"`
+		CertKey                string     `mapstructure:"cert_pem_key"`
+		SecurityProtocol       string     `mapstructure:"security_protocol"`
+		Publisher              Publisher  `mapstructure:"publisher"`
+		Subscriber             Subscriber `mapstructure:"subscriber"`
+		SSLVerificationEnabled bool       `mapstructure:"ssl_verification_enabled"`
+	}
+)
+
+// NewConfig returns a new Config instance.
+func NewConfig() (*Config, error) {
+	config := &Config{}
+	viperBuilder := cbuilder.New("pubsub")
+
+	vConf, err := viperBuilder.Build()
+	if err != nil {
+		return config, err
+	}
+
+	if err = vConf.Unmarshal(config); err != nil {
+		return config, fmt.Errorf("unable to decode into struct: %s", err.Error())
+	}
+
+	return config, nil
+}

pkg/pubsub/config_test.go

@@ -0,0 +1,78 @@
+package pubsub
+
+import (
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewConfig(t *testing.T) {
+	testCases := []struct {
+		name      string
+		wantError bool
+	}{
+		{
+			name:      "NewWithoutConfigFileFails",
+			wantError: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := NewConfig()
+
+			gotError := err != nil
+			assert.Equal(t, gotError, tc.wantError)
+		})
+	}
+}
+
+func TestNewConfigWithAppRoot(t *testing.T) {
+	testCases := []struct {
+		name  string
+		env   string
+		kafka Kafka
+	}{
+		{
+			name: "NewWithConfigFileWorks",
+			env:  "test",
+			kafka: Kafka{
+				BrokerUrls:       []string{"localhost:9092"},
+				ClientId:         "test-app",
+				Cert:             "pem string",
+				CertKey:          "pem key",
+				SecurityProtocol: "ssl",
+				Publisher: Publisher{
+					MaxAttempts:  3,
+					WriteTimeout: 10 * time.Second,
+					Topic:        "test-topic",
+				},
+				Subscriber: Subscriber{
+					Topic: "test-topic",
+				},
+				SSLVerificationEnabled: true,
+			},
+		},
+	}
+
+	currentAppRoot := os.Getenv("APP_ROOT")
+	defer os.Setenv("APP_ROOT", currentAppRoot)
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, filename, _, _ := runtime.Caller(0)
+			tmpRootParent := filepath.Dir(filename)
+			os.Setenv("APP_ROOT", filepath.Join(tmpRootParent, "testdata"))
+
+			c, err := NewConfig()
+			require.Nil(t, err)
+
+			assert.Equal(t, c.Kafka, tc.kafka)
+		})
+	}
+}

pkg/pubsub/testdata/config/pubsub.yml

@@ -0,0 +1,28 @@
+common: &common
+  kafka:
+
+test: &test
+  <<: *common
+  kafka:
+    # Set by APP_PUBSUB_KAFKA_BROKER_URLS env variable
+    broker_urls:
+      - "localhost:9092"
+    # Set by APP_PUBSUB_KAFKA_CLIENT_ID env variable
+    client_id: "test-app"
+    # Set by APP_PUBSUB_KAFKA_CERT_PEM env variable
+    cert_pem: "pem string"
+    # Set by APP_PUBSUB_KAFKA_CERT_PEM_KEY env variable
+    cert_pem_key: "pem key"
+    security_protocol: "ssl"
+    ssl_verification_enabled: true
+    publisher:
+      # Set by APP_PUBSUB_KAFKA_PUBLISHER_MAX_ATTEMPTS env variable
+      max_attempts: 3
+      write_timeout: "10s"
+      topic: "test-topic"
+    subscriber:
+      topic: "test-topic"
+      group_id: ""
+
+development:
+  <<: *test