fix(api): improve queue registration by using regex to extract queue names from processor outputs

Author: flarco

core/dbio/api/api.go

@@ -272,31 +272,41 @@ func (ac *APIConnection) ReadDataflow(endpointName string, sCfg APIStreamConfig)
 
 	// register queues being used by endpoint
 	{
-		for _, processor := range endpoint.Response.Processors {
-			if strings.HasPrefix(processor.Output, "queue.") {
-				queueName := strings.TrimPrefix(processor.Output, "queue.")
+		// Regex pattern to match queue references like "queue.name"
+		queuePattern := regexp.MustCompile(`queue\.([a-zA-Z0-9_]+)`)
+		foundQueues := make(map[string]bool) // Track unique queues
 
-				if !g.In(queueName, ac.Spec.Queues...) {
-					return nil, g.Error("did not declare queue %s in queues list", queueName)
+		// Extract queues from processor outputs
+		for _, processor := range endpoint.Response.Processors {
+			matches := queuePattern.FindAllStringSubmatch(processor.Output, -1)
+			for _, match := range matches {
+				if len(match) > 1 {
+					queueName := match[1]
+					foundQueues[queueName] = true
 				}
+			}
+		}
 
-				_, err = ac.RegisterQueue(queueName)
-				if err != nil {
-					return nil, g.Error(err, "could not register processor output queue: %s", queueName)
+		// Extract queues from iterate over expression
+		if overStr := cast.ToString(endpoint.Iterate.Over); overStr != "" {
+			matches := queuePattern.FindAllStringSubmatch(overStr, -1)
+			for _, match := range matches {
+				if len(match) > 1 {
+					queueName := match[1]
+					foundQueues[queueName] = true
 				}
 			}
 		}
 
-		if overStr := cast.ToString(endpoint.Iterate.Over); strings.HasPrefix(overStr, "queue.") {
-			queueName := strings.TrimPrefix(overStr, "queue.")
-
+		// Validate and register all found queues
+		for queueName := range foundQueues {
 			if !g.In(queueName, ac.Spec.Queues...) {
 				return nil, g.Error("did not declare queue %s in queues list", queueName)
 			}
 
 			_, err = ac.RegisterQueue(queueName)
 			if err != nil {
-				return nil, g.Error(err, "could not register iterate over queue: %s", queueName)
+				return nil, g.Error(err, "could not register queue: %s", queueName)
 			}
 		}
 	}