Method aggregateResponses

osscluster_router.go:430–463

aggregateResponses aggregates multiple shard responses

(cmd Cmder, cmds []Cmder, policy *routing.CommandPolicy)

// aggregateResponses aggregates multiple shard responses
func (c *ClusterClient) aggregateResponses(cmd Cmder, cmds []Cmder, policy *routing.CommandPolicy) error {
	if len(cmds) == 0 {
		return errNoCmdsToAggregate
	}

	if len(cmds) == 1 {
		shardCmd := cmds[0]
		if err := shardCmd.Err(); err != nil {
			cmd.SetErr(err)
			return err
		}
		value, _ := ExtractCommandValue(shardCmd)
		return c.setCommandValue(cmd, value)
	}

	aggregator := c.createAggregator(policy, cmd, false)

	batchWithErrs := []routing.AggregatorResErr{}
	// Add all results to aggregator
	for _, shardCmd := range cmds {
		value, err := ExtractCommandValue(shardCmd)
		batchWithErrs = append(batchWithErrs, routing.AggregatorResErr{
			Result: value,
			Err:    err,
		})
	}

	err := aggregator.BatchSlice(batchWithErrs)
	if err != nil {
		return err
	}

	return c.finishAggregation(cmd, aggregator)
}

// createAggregator creates the appropriate response aggregator
func (c *ClusterClient) createAggregator(policy *routing.CommandPolicy, cmd Cmder, isKeyed bool) routing.ResponseAggregator {