Optional use QueueSubscribe for NATS events source (#3131)

Signed-off-by: sysr9 <sysr999@gmail.com>

api/jsonschema/schema.json

@@ -2288,6 +2288,10 @@
           "description": "Metadata holds the user defined metadata which will passed along the event payload.",
           "type": "object"
         },
+        "queue": {
+          "description": "Queue is the name of the queue group to subscribe as if specified. Uses QueueSubscribe logic to subscribe as queue group. If the queue is empty, uses default Subscribe logic.",
+          "type": "string"
+        },
         "subject": {
           "description": "Subject holds the name of the subject onto which messages are published",
           "type": "string"

eventsources/sources/nats/start.go

@@ -129,8 +129,7 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 		log.Info("assuming all events have a json body...")
 	}
 
-	log.Info("subscribing to messages on the queue...")
-	_, err := conn.Subscribe(natsEventSource.Subject, func(msg *natslib.Msg) {
+	handler := func(msg *natslib.Msg) {
 		defer func(start time.Time) {
 			el.Metrics.EventProcessingDuration(el.GetEventSourceName(), el.GetEventName(), float64(time.Since(start)/time.Millisecond))
 		}(time.Now())
@@ -157,7 +156,17 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 			log.Errorw("failed to dispatch a NATS event", zap.Error(err))
 			el.Metrics.EventProcessingFailed(el.GetEventSourceName(), el.GetEventName())
 		}
-	})
+	}
+
+	var err error
+	if natsEventSource.Queue != nil {
+		log.Infof("subscribing to messages on the subject %s queue %s", natsEventSource.Subject, *natsEventSource.Queue)
+		_, err = conn.QueueSubscribe(natsEventSource.Subject, *natsEventSource.Queue, handler)
+	} else {
+		log.Infof("subscribing to messages on the subject %s", natsEventSource.Subject)
+		_, err = conn.Subscribe(natsEventSource.Subject, handler)
+	}
+
 	if err != nil {
 		return fmt.Errorf("failed to subscribe to the subject %s for event source %s, %w", natsEventSource.Subject, el.GetEventName(), err)
 	}

eventsources/sources/nats/validate.go

@@ -43,5 +43,8 @@ func validate(eventSource *v1alpha1.NATSEventsSource) error {
 	if eventSource.TLS != nil {
 		return apicommon.ValidateTLSConfig(eventSource.TLS)
 	}
+	if eventSource.Queue != nil && *eventSource.Queue == "" {
+		return fmt.Errorf("queue group cannot be empty if specified")
+	}
 	return nil
 }

examples/event-sources/nats.yaml

@@ -79,3 +79,12 @@ spec:
 #          name: my-secret
 #          key: my-credential
 
+#    example-queued-subscription:
+#      url: nats://nats.argo-events.svc:4222
+#      jsonBody: true
+#      subject: "foo"
+#      queue: "my-queue"
+#      auth:
+#        credential:
+#          name: my-secret
+#          key: my-credential

pkg/apis/eventsource/v1alpha1/types.go

@@ -631,6 +631,10 @@ type NATSEventsSource struct {
 	// Filter
 	// +optional
 	Filter *EventSourceFilter `json:"filter,omitempty" protobuf:"bytes,8,opt,name=filter"`
+	// Queue is the name of the queue group to subscribe as if specified. Uses QueueSubscribe
+	// logic to subscribe as queue group. If the queue is empty, uses default Subscribe logic.
+	// +optional
+	Queue *string `json:"queue" protobuf:"bytes,9,opt,name=queue"`
 }
 
 // NATSAuth refers to the auth info for NATS EventSource