Method _process

redis.go:958–1013

(ctx context.Context, cmd Cmder, attempt int)

}

func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool, *pool.Conn, error) {
	if attempt > 0 {
		if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
			return false, nil, err
		}
	}

	var usedConn *pool.Conn
	retryTimeout := uint32(0)
	if err := c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
		usedConn = cn
		// Process any pending push notifications before executing the command
		if err := c.processPushNotifications(ctx, cn); err != nil {
			internal.Logger.Printf(ctx, "push: error processing pending notifications before command: %v", err)
		}

		if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
			return writeCmd(wr, cmd)
		}); err != nil {
			atomic.StoreUint32(&retryTimeout, 1)
			return err
		}
		readReplyFunc := cmd.readReply
		// Apply unstable RESP3 search module.
		if c.opt.Protocol != 2 {
			useRawReply, err := c.assertUnstableCommand(cmd)
			if err != nil {
				return err
			}
			if useRawReply {
				readReplyFunc = cmd.readRawReply
			}
		}
		if err := cn.WithReader(c.context(ctx), c.cmdTimeout(cmd), func(rd *proto.Reader) error {
			// To be sure there are no buffered push notifications, we process them before reading the reply
			if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
				internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
			}
			return readReplyFunc(rd)
		}); err != nil {
			if cmd.readTimeout() == nil {
				atomic.StoreUint32(&retryTimeout, 1)
			} else {
				atomic.StoreUint32(&retryTimeout, 0)
			}
			return err
		}

		return nil
	}); err != nil {
		retry := shouldRetry(err, atomic.LoadUint32(&retryTimeout) == 1)
		return retry, usedConn, err
	}

	return false, usedConn, nil
}

func (c *baseClient) retryBackoff(attempt int) time.Duration {