( ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap, )
| 1705 | } |
| 1706 | |
| 1707 | func (c *ClusterClient) processPipelineNode( |
| 1708 | ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap, |
| 1709 | ) { |
| 1710 | _ = node.Client.withProcessPipelineHook(ctx, cmds, func(ctx context.Context, cmds []Cmder) error { |
| 1711 | cn, err := node.Client.getConn(ctx) |
| 1712 | if err != nil { |
| 1713 | if !isContextError(err) { |
| 1714 | node.MarkAsFailing() |
| 1715 | } |
| 1716 | _ = c.mapCmdsByNode(ctx, failedCmds, cmds) |
| 1717 | setCmdsErr(cmds, err) |
| 1718 | return err |
| 1719 | } |
| 1720 | |
| 1721 | var processErr error |
| 1722 | defer func() { |
| 1723 | node.Client.releaseConn(ctx, cn, processErr) |
| 1724 | }() |
| 1725 | processErr = c.processPipelineNodeConn(ctx, node, cn, cmds, failedCmds) |
| 1726 | |
| 1727 | return processErr |
| 1728 | }) |
| 1729 | } |
| 1730 | |
| 1731 | func (c *ClusterClient) processPipelineNodeConn( |
| 1732 | ctx context.Context, node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds *cmdsMap, |
no test coverage detected