fix: nats connection

Author: moshloop

pkg/apis/batch/v1/types.go

@@ -105,7 +105,7 @@ type ExecAction struct {
 }
 
 type Retry struct {
-	Attempts int `json:"attempts"`
+	Attempts int `json:"attempts,omitempty"`
 	// Delay is the time in seconds to wait between retries
 	Delay int `json:"delay"`
 }
@@ -164,13 +164,13 @@ func (p PubSubConfig) String() string {
 
 // +kubebuilder:object:generate=true
 type NATSConfig struct {
-	URL     `json:",inline"`
+	URL     string `json:"url,omitempty"`
 	Subject string `json:"subject"`
-	Queue   string `json:"queue"`
+	Queue   string `json:"queue,omitempty"`
 }
 
 func (n NATSConfig) String() string {
-	return fmt.Sprintf("nats://%s", n.Queue)
+	return fmt.Sprintf("nats://%s", n.Subject)
 }
 
 // +kubebuilder:object:generate=true

pkg/types.go

@@ -7,9 +7,11 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/flanksource/commons/logger"
 	"github.com/flanksource/duty/context"
+	"github.com/nats-io/nats.go"
 	"gocloud.dev/pubsub"
 	"gocloud.dev/pubsub/awssnssqs"
 	"gocloud.dev/pubsub/kafkapubsub"
+	"gocloud.dev/pubsub/natspubsub"
 
 	batchv1alpha1 "github.com/flanksource/batch-runner/pkg/apis/batch/v1"
 )
@@ -63,7 +65,14 @@ func Subscribe(ctx context.Context, c *batchv1alpha1.Config) (*pubsub.Subscripti
 	}
 
 	if c.NATS != nil {
-		return pubsub.OpenSubscription(ctx, fmt.Sprintf("nats://%s", c.NATS.Queue))
+		conn, err := nats.Connect(c.NATS.URL)
+		if err != nil {
+			return nil, err
+		}
+
+		return natspubsub.OpenSubscriptionV2(conn, c.NATS.Subject, &natspubsub.SubscriptionOptions{
+			Queue: c.NATS.Queue,
+		})
 	}
 
 	if c.Memory != nil {