Merge pull request #22 from usedatabrew/nats-rabbit-sinks-improvements

feat(sinks): add auth mechanics for nats, exchange name as optional f…

internal/sinks/nats/config.go

@@ -1,6 +1,8 @@
 package nats
 
 type Config struct {
-	Url     string `json:"url"`
-	Subject string `json:"subject"`
+	Url      string `json:"url" yaml:"url"`
+	Subject  string `json:"subject" yaml:"subject"`
+	Username string `json:"username" yaml:"username"`
+	Password string `json:"password" yaml:"password"`
 }

internal/sinks/nats/plugin.go

@@ -28,7 +28,13 @@ func NewNatsSinkPlugin(config Config, schema []schema.StreamSchema, appCtx *stre
 }
 
 func (s *SinkPlugin) Connect(ctx context.Context) error {
-	client, err := nats.Connect(s.config.Url)
+	var opt nats.Option
+
+	if s.config.Username != "" {
+		opt = nats.UserInfo(s.config.Username, s.config.Password)
+	}
+
+	client, err := nats.Connect(s.config.Url, opt)
 
 	if err != nil {
 		return err

internal/sinks/rabbit_mq/plugin.go

@@ -52,11 +52,17 @@ func (s *SinkPlugin) Connect(ctx context.Context) error {
 }
 
 func (s *SinkPlugin) Write(message *message.Message) error {
+	var opts func(*rabbitmq.PublishOptions)
+
+	if s.config.ExchangeName != "" {
+		opts = rabbitmq.WithPublishOptionsExchange(s.config.ExchangeName)
+	}
+
 	return s.publisher.Publish(
 		[]byte(message.AsJSONString()),
 		[]string{s.config.RoutingKey},
 		rabbitmq.WithPublishOptionsContentType("application/json"),
-		rabbitmq.WithPublishOptionsExchange(s.config.ExchangeName),
+		opts,
 	)
 }