Method generalProcessPipeline

ring.go:838–890

(
	ctx context.Context, cmds []Cmder, tx bool,
)

}

func (c *Ring) generalProcessPipeline(
	ctx context.Context, cmds []Cmder, tx bool,
) error {
	if tx {
		// Trim multi .. exec.
		cmds = cmds[1 : len(cmds)-1]
	}

	cmdsMap := make(map[string][]Cmder)

	for _, cmd := range cmds {
		hash := cmd.stringArg(cmdFirstKeyPosWithInfo(cmd, nil))
		if hash != "" {
			hash = c.sharding.Hash(hash)
		}
		cmdsMap[hash] = append(cmdsMap[hash], cmd)
	}

	var wg sync.WaitGroup
	errs := make(chan error, len(cmdsMap))

	for hash, cmds := range cmdsMap {
		wg.Add(1)
		go func(hash string, cmds []Cmder) {
			defer wg.Done()

			// TODO: retry?
			shard, err := c.sharding.GetByName(hash)
			if err != nil {
				setCmdsErr(cmds, err)
				return
			}

			hook := shard.Client.processPipelineHook
			if tx {
				cmds = wrapMultiExec(ctx, cmds)
				hook = shard.Client.processTxPipelineHook
			}

			if err = hook(ctx, cmds); err != nil {
				errs <- err
			}
		}(hash, cmds)
	}

	wg.Wait()
	close(errs)

	if err := <-errs; err != nil {
		return err
	}
	return cmdsFirstErr(cmds)
}

func (c *Ring) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
	if len(keys) == 0 {