Method aggregateMultiSlotResults

osscluster_router.go:366–402

aggregateMultiSlotResults aggregates results from multi-slot execution

(ctx context.Context, cmd Cmder, results <-chan slotResult, keyOrder []string, policy *routing.CommandPolicy)

// aggregateMultiSlotResults aggregates results from multi-slot execution
func (c *ClusterClient) aggregateMultiSlotResults(ctx context.Context, cmd Cmder, results <-chan slotResult, keyOrder []string, policy *routing.CommandPolicy) error {
	keyedResults := make(map[string]routing.AggregatorResErr)
	var firstErr error

	for result := range results {
		if result.err != nil && firstErr == nil {
			firstErr = result.err
		}
		if result.cmd != nil && result.err == nil {
			value, err := ExtractCommandValue(result.cmd)

			// Check if the result is a slice (e.g., from MGET)
			if sliceValue, ok := value.([]interface{}); ok {
				// Map each element to its corresponding key
				for i, key := range result.keys {
					if i < len(sliceValue) {
						keyedResults[key] = routing.AggregatorResErr{Result: sliceValue[i], Err: err}
					} else {
						keyedResults[key] = routing.AggregatorResErr{Result: nil, Err: err}
					}
				}
			} else {
				// For non-slice results, map the entire result to each key
				for _, key := range result.keys {
					keyedResults[key] = routing.AggregatorResErr{Result: value, Err: err}
				}
			}
		}

		// TODO: return multiple errors by order when we will implement multiple errors returning
		if result.err != nil {
			firstErr = result.err
		}
	}

	return c.aggregateKeyedValues(cmd, keyedResults, keyOrder, policy)
}

// aggregateKeyedValues aggregates individual key-value pairs while preserving key order
func (c *ClusterClient) aggregateKeyedValues(cmd Cmder, keyedResults map[string]routing.AggregatorResErr, keyOrder []string, policy *routing.CommandPolicy) error {