func (d *downstreamResponseMap) getAggregatedKeys(m *mapper.Mapper) (map[string]bool, error) {
	d.mu.RLock()
	defer d.mu.RUnlock()
	keys := make(map[string]bool)
	for request := range d.watches {
		key, err := mapper.Mapper.GetKey(*m, request)
		if err != nil {
			return nil, err
		}
		keys[key] = true
	}
	return keys, nil
}
getAggregatedKeys returns a list of aggregated keys for all requests in the downstream response map.
package orchestrator
import (
	"sync"
	"github.com/envoyproxy/xds-relay/internal/app/cache"
	"github.com/envoyproxy/xds-relay/internal/app/mapper"
	"github.com/envoyproxy/xds-relay/internal/app/transport"
	"github.com/uber-go/tally"
)
type downstreamResponseMap struct {
	mu      sync.RWMutex
	watches map[transport.Request]transport.Watch
}
func newDownstreamResponseMap() downstreamResponseMap {
	return downstreamResponseMap{
		watches: make(map[transport.Request]transport.Watch),
	}
}
func (d *downstreamResponseMap) createWatch(req transport.Request, scope tally.Scope) transport.Watch {
	d.mu.Lock()
	defer d.mu.Unlock()
	if _, ok := d.watches[req]; !ok {
		d.watches[req] = req.CreateWatch(scope)
	}
	return d.watches[req]
}
func (d *downstreamResponseMap) get(req transport.Request) (transport.Watch, bool) {
	d.mu.RLock()
	defer d.mu.RUnlock()
	channel, ok := d.watches[req]
	return channel, ok
}
func (d *downstreamResponseMap) delete(req transport.Request) transport.Watch {
	d.mu.Lock()
	defer d.mu.Unlock()
	if watch, ok := d.watches[req]; ok {
		watch.Close()
		delete(d.watches, req)
		return watch
	}
	return nil
}
func (d *downstreamResponseMap) deleteAll(watchers *cache.RequestsStore) {
	d.mu.Lock()
	defer d.mu.Unlock()
	watchers.ForEach(func(watch transport.Request) {
		if w, ok := d.watches[watch]; ok {
			w.Close()
			delete(d.watches, watch)
		}
	})
}