Refactor v2 Paths to parse YAML using TranslatePipeline.

datamodel/high/v2/path_item.go

@@ -4,6 +4,8 @@
 package v2
 
 import (
+	"sync"
+
 	"github.com/pb33f/libopenapi/datamodel/high"
 	low "github.com/pb33f/libopenapi/datamodel/low/v2"
 )
@@ -40,71 +42,61 @@ func NewPathItem(pathItem *low.PathItem) *PathItem {
 		}
 		p.Parameters = params
 	}
-	var buildOperation = func(method string, op *low.Operation, resChan chan<- asyncResult[*Operation]) {
-		resChan <- asyncResult[*Operation]{
-			key:    method,
-			result: NewOperation(op),
-		}
+	var buildOperation = func(method string, op *low.Operation) *Operation {
+		return NewOperation(op)
 	}
-	totalOperations := 0
-	resChan := make(chan asyncResult[*Operation])
+
+	var wg sync.WaitGroup
 	if !pathItem.Get.IsEmpty() {
-		totalOperations++
-		go buildOperation(low.GetLabel, pathItem.Get.Value, resChan)
+		wg.Add(1)
+		go func() {
+			p.Get = buildOperation(low.GetLabel, pathItem.Get.Value)
+			wg.Done()
+		}()
 	}
 	if !pathItem.Put.IsEmpty() {
-		totalOperations++
-		go buildOperation(low.PutLabel, pathItem.Put.Value, resChan)
+		wg.Add(1)
+		go func() {
+			p.Put = buildOperation(low.PutLabel, pathItem.Put.Value)
+			wg.Done()
+		}()
 	}
 	if !pathItem.Post.IsEmpty() {
-		totalOperations++
-		go buildOperation(low.PostLabel, pathItem.Post.Value, resChan)
+		wg.Add(1)
+		go func() {
+			p.Post = buildOperation(low.PostLabel, pathItem.Post.Value)
+			wg.Done()
+		}()
 	}
 	if !pathItem.Patch.IsEmpty() {
-		totalOperations++
-		go buildOperation(low.PatchLabel, pathItem.Patch.Value, resChan)
+		wg.Add(1)
+		go func() {
+			p.Patch = buildOperation(low.PatchLabel, pathItem.Patch.Value)
+			wg.Done()
+		}()
 	}
 	if !pathItem.Delete.IsEmpty() {
-		totalOperations++
-		go buildOperation(low.DeleteLabel, pathItem.Delete.Value, resChan)
+		wg.Add(1)
+		go func() {
+			p.Delete = buildOperation(low.DeleteLabel, pathItem.Delete.Value)
+			wg.Done()
+		}()
 	}
 	if !pathItem.Head.IsEmpty() {
-		totalOperations++
-		go buildOperation(low.HeadLabel, pathItem.Head.Value, resChan)
+		wg.Add(1)
+		go func() {
+			p.Head = buildOperation(low.HeadLabel, pathItem.Head.Value)
+			wg.Done()
+		}()
 	}
 	if !pathItem.Options.IsEmpty() {
-		totalOperations++
-		go buildOperation(low.OptionsLabel, pathItem.Options.Value, resChan)
-	}
-	completedOperations := 0
-	for completedOperations < totalOperations {
-		select {
-		case r := <-resChan:
-			switch r.key {
-			case low.GetLabel:
-				completedOperations++
-				p.Get = r.result
-			case low.PutLabel:
-				completedOperations++
-				p.Put = r.result
-			case low.PostLabel:
-				completedOperations++
-				p.Post = r.result
-			case low.PatchLabel:
-				completedOperations++
-				p.Patch = r.result
-			case low.DeleteLabel:
-				completedOperations++
-				p.Delete = r.result
-			case low.HeadLabel:
-				completedOperations++
-				p.Head = r.result
-			case low.OptionsLabel:
-				completedOperations++
-				p.Options = r.result
-			}
-		}
+		wg.Add(1)
+		go func() {
+			p.Options = buildOperation(low.OptionsLabel, pathItem.Options.Value)
+			wg.Done()
+		}()
 	}
+	wg.Wait()
 	return p
 }
 

datamodel/high/v2/paths.go

@@ -4,50 +4,44 @@
 package v2
 
 import (
+	"github.com/pb33f/libopenapi/datamodel"
 	"github.com/pb33f/libopenapi/datamodel/high"
-	low "github.com/pb33f/libopenapi/datamodel/low/v2"
+	"github.com/pb33f/libopenapi/datamodel/low"
+	v2low "github.com/pb33f/libopenapi/datamodel/low/v2"
 )
 
 // Paths represents a high-level Swagger / OpenAPI Paths object, backed by a low-level one.
 type Paths struct {
 	PathItems  map[string]*PathItem
 	Extensions map[string]any
-	low        *low.Paths
+	low        *v2low.Paths
 }
 
 // NewPaths creates a new high-level instance of Paths from a low-level one.
-func NewPaths(paths *low.Paths) *Paths {
+func NewPaths(paths *v2low.Paths) *Paths {
 	p := new(Paths)
 	p.low = paths
 	p.Extensions = high.ExtractExtensions(paths.Extensions)
+	pathItems := make(map[string]*PathItem)
 
-	resultChan := make(chan asyncResult[*PathItem])
-	var buildPath = func(path string, pi *low.PathItem, rChan chan<- asyncResult[*PathItem]) {
-		rChan <- asyncResult[*PathItem]{
-			key:    path,
-			result: NewPathItem(pi),
-		}
+	translateFunc := func(key low.KeyReference[string], value low.ValueReference[*v2low.PathItem]) (asyncResult[*PathItem], error) {
+		return asyncResult[*PathItem]{
+			key:    key.Value,
+			result: NewPathItem(value.Value),
+		}, nil
 	}
-	if len(paths.PathItems) > 0 {
-		pathItems := make(map[string]*PathItem)
-		totalPaths := len(paths.PathItems)
-		for k := range paths.PathItems {
-			go buildPath(k.Value, paths.PathItems[k].Value, resultChan)
-		}
-		completedPaths := 0
-		for completedPaths < totalPaths {
-			select {
-			case res := <-resultChan:
-				completedPaths++
-				pathItems[res.key] = res.result
-			}
-		}
-		p.PathItems = pathItems
+	resultFunc := func(result asyncResult[*PathItem]) error {
+		pathItems[result.key] = result.result
+		return nil
 	}
+	_ = datamodel.TranslateMapParallel[low.KeyReference[string], low.ValueReference[*v2low.PathItem], asyncResult[*PathItem]](
+		paths.PathItems, translateFunc, resultFunc,
+	)
+	p.PathItems = pathItems
 	return p
 }
 
 // GoLow returns the low-level Paths instance that backs the high level one.
-func (p *Paths) GoLow() *low.Paths {
+func (p *Paths) GoLow() *v2low.Paths {
 	return p.low
 }

datamodel/high/v3/paths.go

@@ -33,19 +33,19 @@ func NewPaths(paths *v3low.Paths) *Paths {
 	p.Extensions = high.ExtractExtensions(paths.Extensions)
 	items := make(map[string]*PathItem)
 
-	type pRes struct {
+	type pathItemResult struct {
 		key   string
 		value *PathItem
 	}
 
-	translateFunc := func(key low.KeyReference[string], value low.ValueReference[*v3low.PathItem]) (pRes, error) {
-		return pRes{key: key.Value, value: NewPathItem(value.Value)}, nil
+	translateFunc := func(key low.KeyReference[string], value low.ValueReference[*v3low.PathItem]) (pathItemResult, error) {
+		return pathItemResult{key: key.Value, value: NewPathItem(value.Value)}, nil
 	}
-	resultFunc := func(value pRes) error {
+	resultFunc := func(value pathItemResult) error {
 		items[value.key] = value.value
 		return nil
 	}
-	_ = datamodel.TranslateMapParallel[low.KeyReference[string], low.ValueReference[*v3low.PathItem], pRes](
+	_ = datamodel.TranslateMapParallel[low.KeyReference[string], low.ValueReference[*v3low.PathItem], pathItemResult](
 		paths.PathItems, translateFunc, resultFunc,
 	)
 	p.PathItems = items

datamodel/low/v2/path_item.go

@@ -6,13 +6,14 @@ package v2
 import (
 	"crypto/sha256"
 	"fmt"
+	"sort"
+	"strings"
+	"sync"
+
 	"github.com/pb33f/libopenapi/datamodel/low"
 	"github.com/pb33f/libopenapi/index"
 	"github.com/pb33f/libopenapi/utils"
 	"gopkg.in/yaml.v3"
-	"sort"
-	"strings"
-	"sync"
 )
 
 // PathItem represents a low-level Swagger / OpenAPI 2 PathItem object.

datamodel/low/v2/paths.go

@@ -4,14 +4,18 @@
 package v2
 
 import (
+	"context"
 	"crypto/sha256"
 	"fmt"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/pb33f/libopenapi/datamodel"
 	"github.com/pb33f/libopenapi/datamodel/low"
 	"github.com/pb33f/libopenapi/index"
 	"github.com/pb33f/libopenapi/utils"
 	"gopkg.in/yaml.v3"
-	"sort"
-	"strings"
 )
 
 // Paths represents a low-level Swagger / OpenAPI Paths object.
@@ -55,65 +59,104 @@ func (p *Paths) Build(_, root *yaml.Node, idx *index.SpecIndex) error {
 	root = utils.NodeAlias(root)
 	utils.CheckForMergeNodes(root)
 	p.Extensions = low.ExtractExtensions(root)
-	skip := false
-	var currentNode *yaml.Node
+	// skip := false
+	// var currentNode *yaml.Node
 
-	pathsMap := make(map[low.KeyReference[string]]low.ValueReference[*PathItem])
-
-	// build each new path, in a new thread.
+	// Translate YAML nodes to pathsMap using `TranslatePipeline`.
 	type pathBuildResult struct {
-		k low.KeyReference[string]
-		v low.ValueReference[*PathItem]
+		key   low.KeyReference[string]
+		value low.ValueReference[*PathItem]
 	}
+	type nodeItem struct {
+		currentNode *yaml.Node
+		pathNode    *yaml.Node
+	}
+	pathsMap := make(map[low.KeyReference[string]]low.ValueReference[*PathItem])
+	in := make(chan nodeItem)
+	out := make(chan pathBuildResult)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var wg sync.WaitGroup
+	wg.Add(2) // input and output goroutines.
 
-	bChan := make(chan pathBuildResult)
-	eChan := make(chan error)
-	var buildPathItem = func(cNode, pNode *yaml.Node, b chan<- pathBuildResult, e chan<- error) {
+	// TranslatePipeline input.
+	go func() {
+		defer func() {
+			close(in)
+			wg.Done()
+		}()
+		skip := false
+		var currentNode *yaml.Node
+		for i, pathNode := range root.Content {
+			if strings.HasPrefix(strings.ToLower(pathNode.Value), "x-") {
+				skip = true
+				continue
+			}
+			if skip {
+				skip = false
+				continue
+			}
+			if i%2 == 0 {
+				currentNode = pathNode
+				continue
+			}
+
+			select {
+			case in <- nodeItem{
+				currentNode: currentNode,
+				pathNode:    pathNode,
+			}:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	// TranslatePipeline output.
+	go func() {
+		defer func() {
+			cancel()
+			wg.Done()
+		}()
+		for {
+			select {
+			case result, ok := <-out:
+				if !ok {
+					return
+				}
+				pathsMap[result.key] = result.value
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	translateFunc := func(value nodeItem) (retval pathBuildResult, _ error) {
+		pNode := value.pathNode
+		cNode := value.currentNode
 		path := new(PathItem)
 		_ = low.BuildModel(pNode, path)
 		err := path.Build(cNode, pNode, idx)
 		if err != nil {
-			e <- err
-			return
+			return retval, err
 		}
-		b <- pathBuildResult{
-			k: low.KeyReference[string]{
+		return pathBuildResult{
+			key: low.KeyReference[string]{
 				Value:   cNode.Value,
 				KeyNode: cNode,
 			},
-			v: low.ValueReference[*PathItem]{
+			value: low.ValueReference[*PathItem]{
 				Value:     path,
 				ValueNode: pNode,
 			},
-		}
+		}, nil
 	}
-	pathCount := 0
-	for i, pathNode := range root.Content {
-		if strings.HasPrefix(strings.ToLower(pathNode.Value), "x-") {
-			skip = true
-			continue
-		}
-		if skip {
-			skip = false
-			continue
-		}
-		if i%2 == 0 {
-			currentNode = pathNode
-			continue
-		}
-		pathCount++
-		go buildPathItem(currentNode, pathNode, bChan, eChan)
-	}
-	completedItems := 0
-	for completedItems < pathCount {
-		select {
-		case err := <-eChan:
-			return err
-		case res := <-bChan:
-			completedItems++
-			pathsMap[res.k] = res.v
-		}
+	err := datamodel.TranslatePipeline[nodeItem, pathBuildResult](in, out, translateFunc)
+	wg.Wait()
+	if err != nil {
+		return err
 	}
+
 	p.PathItems = pathsMap
 	return nil
 }

datamodel/low/v3/path_item.go

@@ -4,13 +4,13 @@
 package v3
 
 import (
-	"context"
 	"crypto/sha256"
 	"fmt"
 	"sort"
 	"strings"
 	"sync"
 
+	"github.com/pb33f/libopenapi/datamodel"
 	"github.com/pb33f/libopenapi/datamodel/low"
 	"github.com/pb33f/libopenapi/index"
 	"github.com/pb33f/libopenapi/utils"
@@ -271,58 +271,24 @@ func (p *PathItem) Build(_, root *yaml.Node, idx *index.SpecIndex) error {
 
 	// all operations have been superficially built,
 	// now we need to build out the operation, we will do this asynchronously for speed.
-	opBuildChan := make(chan bool)
-	opErrorChan := make(chan error)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	buildOpFunc := func(op low.NodeReference[*Operation], ch chan<- bool, errCh chan<- error, ref string) {
-		er := op.Value.Build(op.KeyNode, op.ValueNode, idx)
-		if ref != "" {
-			op.Value.Reference.Reference = ref
-		}
-		if er != nil {
-			select {
-			case errCh <- er:
-			case <-ctx.Done():
-			}
-			return
-		}
-		select {
-		case ch <- true:
-		case <-ctx.Done():
-		}
-	}
-
-	if len(ops) <= 0 {
-		return nil // nothing to do.
-	}
-
-	for _, op := range ops {
+	translateFunc := func(_ int, op low.NodeReference[*Operation]) (any, error) {
 		ref := ""
 		if op.ReferenceNode {
 			ref = op.Reference
 		}
-		go buildOpFunc(op, opBuildChan, opErrorChan, ref)
-	}
 
-	n := 0
-	total := len(ops)
-FORLOOP1:
-	for n < total {
-		select {
-		case buildError := <-opErrorChan:
-			return buildError
-		case <-opBuildChan:
-			n++
-		case <-ctx.Done():
-			break FORLOOP1
+		err := op.Value.Build(op.KeyNode, op.ValueNode, idx)
+		if ref != "" {
+			op.Value.Reference.Reference = ref
 		}
+		if err != nil {
+			return nil, err
+		}
+		return nil, nil
 	}
-
-	// make sure we don't exit before the path is finished building.
-	if len(ops) > 0 {
-		wg.Wait()
+	err := datamodel.TranslateSliceParallel[low.NodeReference[*Operation], any](ops, translateFunc, nil)
+	if err != nil {
+		return err
 	}
 	return nil
 }

datamodel/translate.go

@@ -25,7 +25,7 @@ type jobStatus[OUT any] struct {
 	result OUT
 }
 
-type tpJobStatus[IN any, OUT any] struct {
+type pipelineJobStatus[IN any, OUT any] struct {
 	done   chan struct{}
 	cont   bool
 	eof    bool
@@ -201,8 +201,8 @@ func TranslatePipeline[IN any, OUT any](in <-chan IN, out chan<- OUT, translate
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	concurrency := runtime.NumCPU()
-	workChan := make(chan *tpJobStatus[IN, OUT])
-	resultChan := make(chan *tpJobStatus[IN, OUT])
+	workChan := make(chan *pipelineJobStatus[IN, OUT])
+	resultChan := make(chan *pipelineJobStatus[IN, OUT])
 	var reterr error
 	var mu sync.Mutex
 	var wg sync.WaitGroup
@@ -257,7 +257,7 @@ func TranslatePipeline[IN any, OUT any](in <-chan IN, out chan<- OUT, translate
 				if !ok {
 					return
 				}
-				j := &tpJobStatus[IN, OUT]{
+				j := &pipelineJobStatus[IN, OUT]{
 					done:  make(chan struct{}),
 					input: value,
 				}