use channels to pass (error) results to and receive results from prox…

…ies/requests
coupergateway · Jan 31, 2023 · 9f93643 · 9f93643
1 parent 36a6b83
commit 9f93643
Show file tree

Hide file tree

Showing 12 changed files with 248 additions and 1,029 deletions.
diff --git a/config/runtime/backend.go b/config/runtime/backend.go
@@ -180,12 +180,12 @@ func newRequestAuthorizer(evalCtx *hcl.EvalContext, block *hclsyntax.Block,
 	case *config.OAuth2ReqAuth:
 		return transport.NewOAuth2ReqAuth(evalCtx, impl, memStore, authorizerBackend)
 	case *config.TokenRequest:
-		reqs := producer.Requests{&producer.Request{
+		req := &producer.Request{
 			Backend: authorizerBackend,
 			Context: impl.HCLBody(),
 			Name:    impl.Name,
-		}}
-		return transport.NewTokenRequest(impl, memStore, reqs)
+		}
+		return transport.NewTokenRequest(impl, memStore, req)
 	default:
 		return nil, errors.Configuration.Message("unknown authorizer type")
 	}

diff --git a/config/runtime/endpoint.go b/config/runtime/endpoint.go
@@ -112,7 +112,7 @@ func NewEndpointOptions(confCtx *hcl.EvalContext, endpointConf *config.Endpoint,
 		blockBodies = append(blockBodies, response.Context)
 	}
 
-	allProxies := make(map[string]*producer.Proxy)
+	allProducers := make(map[string]producer.Roundtrip)
 	for _, proxyConf := range endpointConf.Proxies {
 		backend, berr := NewBackend(confCtx, proxyConf.Backend, log, conf, memStore)
 		if berr != nil {
@@ -137,11 +137,10 @@ func NewEndpointOptions(confCtx *hcl.EvalContext, endpointConf *config.Endpoint,
 			RoundTrip: proxyHandler,
 		}
 
-		allProxies[proxyConf.Name] = p
+		allProducers[proxyConf.Name] = p
 		blockBodies = append(blockBodies, proxyConf.Backend, proxyBody)
 	}
 
-	allRequests := make(map[string]*producer.Request)
 	for _, requestConf := range endpointConf.Requests {
 		backend, berr := NewBackend(confCtx, requestConf.Backend, log, conf, memStore)
 		if berr != nil {
@@ -154,14 +153,30 @@ func NewEndpointOptions(confCtx *hcl.EvalContext, endpointConf *config.Endpoint,
 			Name:    requestConf.Name,
 		}
 
-		allRequests[requestConf.Name] = pr
+		allProducers[requestConf.Name] = pr
 		blockBodies = append(blockBodies, requestConf.Backend, requestConf.HCLBody())
 	}
 
-	sequences, requests, proxies := resolveDependencies(allProxies, allRequests, endpointConf.Sequences...)
+	markDepencencies(allProducers, endpointConf.Sequences)
+	// add independent producers
+	// TODO simplify
+	allDeps := sequence.Dependencies(endpointConf.Sequences)
+	for name := range allProducers {
+		inSeq := false
+		for _, deps := range allDeps {
+			for _, dep := range deps {
+				if name == dep {
+					inSeq = true
+				}
+			}
+		}
+		if !inSeq {
+			endpointConf.Sequences = append(endpointConf.Sequences, &sequence.Item{Name: name})
+		}
+	}
 
 	// TODO: redirect
-	if endpointConf.Response == nil && len(proxies)+len(requests)+len(sequences) == 0 { // && redirect == nil
+	if endpointConf.Response == nil && len(allProducers) == 0 { // && redirect == nil
 		r := endpointConf.HCLBody().SrcRange
 		m := fmt.Sprintf("configuration error: endpoint: %q requires at least one proxy, request or response block", endpointConf.Pattern)
 		return nil, hcl.Diagnostics{&hcl.Diagnostic{
@@ -192,132 +207,26 @@ func NewEndpointOptions(confCtx *hcl.EvalContext, endpointConf *config.Endpoint,
 		APIName:           apiName,
 		Context:           endpointConf.HCLBody(),
 		ErrorTemplate:     errTpl,
+		Items:             endpointConf.Sequences,
 		LogPattern:        endpointConf.Pattern,
-		Proxies:           proxies,
+		Producers:         allProducers,
 		ReqBodyLimit:      bodyLimit,
 		BufferOpts:        bufferOpts,
-		Requests:          requests,
 		SendServerTimings: conf.Settings.SendServerTimings,
-		Sequences:         sequences,
 		Response:          response,
 		ServerOpts:        serverOptions,
 	}, nil
 }
 
-// resolveDependencies lookups any request related dependency and sort them into a sequence.
-// Also return left-overs for parallel usage.
-func resolveDependencies(proxies map[string]*producer.Proxy, requests map[string]*producer.Request,
-	items ...*sequence.Item) (producer.Parallel, producer.Requests, producer.Proxies) {
-
-	allDeps := sequence.Dependencies(items)
-
-	var reqs producer.Requests
-	var ps producer.Proxies
-	var seqs producer.Parallel
-	roundtrips := map[string]producer.Roundtrip{}
-
-	// read from prepared config sequences
-	for _, seq := range items {
-		seqs = append(seqs, newRoundtrip(seq, roundtrips, proxies, requests))
-	}
-
-proxyLeftovers:
-	for name, p := range proxies {
-		for _, deps := range allDeps {
-			for _, dep := range deps {
-				if name == dep {
-					continue proxyLeftovers
-				}
-			}
+func markDepencencies(allProducers map[string]producer.Roundtrip, items sequence.List) {
+	for _, item := range items {
+		pr := allProducers[item.Name]
+		var prevs []string
+		deps := item.Deps()
+		for _, dep := range deps {
+			prevs = append(prevs, dep.Name)
 		}
-		ps = append(ps, p)
-	}
-
-reqLeftovers:
-	for name, r := range requests {
-		for _, deps := range allDeps {
-			for _, dep := range deps {
-				if name == dep {
-					continue reqLeftovers
-				}
-			}
-		}
-		reqs = append(reqs, r)
-	}
-
-	return seqs, reqs, ps
-}
-
-func newRoundtrip(seq *sequence.Item,
-	roundtrips map[string]producer.Roundtrip,
-	proxies map[string]*producer.Proxy,
-	requests map[string]*producer.Request) producer.Roundtrip {
-
-	deps := seq.Deps()
-	var rt producer.Roundtrip
-
-	var previous []string
-	if len(deps) > 1 { // more deps per item can be parallelized
-		var names []string
-		for _, d := range deps {
-			names = append(names, d.Name)
-		}
-		k := fmt.Sprintf("%v", names)
-		for _, d := range deps {
-			previous = append(previous, d.Name)
-		}
-		if np, ok := roundtrips[k]; ok {
-			rt = np
-		} else {
-			var pl producer.Parallel
-			for _, d := range deps {
-				pl = append(pl, newRoundtrip(d, roundtrips, proxies, requests))
-			}
-			rt = &pl
-			roundtrips[k] = &pl
-		}
-	} else if len(deps) == 1 {
-		rt = newRoundtrip(deps[0], roundtrips, proxies, requests)
-		previous = append(previous, deps[0].Name)
-	}
-
-	leaf := newLeafRoundtrip(seq.Name, strings.Join(previous, ","), roundtrips, proxies, requests)
-	if rt != nil {
-		var names []string
-		names = append(names, rt.Names()...)
-		names = append(names, leaf.Names()...)
-		k := fmt.Sprintf("%v", names)
-		if ns, ok := roundtrips[k]; ok {
-			return ns
-		}
-		s := &producer.Sequence{rt, leaf}
-		roundtrips[k] = s
-		return s
-	}
-	return leaf
-}
-
-// newLeafRoundtrip creates a "leaf" Roundtrip, i.e. one of
-// producer.Proxies or producer.Requests,
-// no producer.Parallel or producer.Sequence
-func newLeafRoundtrip(name, previous string,
-	roundtrips map[string]producer.Roundtrip,
-	proxies map[string]*producer.Proxy,
-	requests map[string]*producer.Request) producer.Roundtrip {
-	if rt, ok := roundtrips[name]; ok {
-		return rt
-	}
-	if p, ok := proxies[name]; ok {
-		p.PreviousSequence = previous
-		ps := &producer.Proxies{p}
-		roundtrips[name] = ps
-		return ps
-	}
-	if r, ok := requests[name]; ok {
-		r.PreviousSequence = previous
-		rs := &producer.Requests{r}
-		roundtrips[name] = rs
-		return rs
+		pr.SetPreviousSequence(strings.Join(prevs, ","))
+		markDepencencies(allProducers, deps)
 	}
-	return nil
 }