Expose pipeline command results via Cmds() method

[RasankRam]

Motivation

We are building a transaction package manager on top of go-redis that orchestrates pipelines and Postgres transactions together.
For this integration, we need a way to retrieve the list of commands that were queued in a pipeline.

At the moment, pipeline.go stores the commands in a private field. Because it is unexported, we cannot access or inspect the collected commands outside the package.

Proposal

Introduce a method like:

func (p *Pipeline) Cmds() []Cmder

[cxljs]

@ndyakov Do you think this feature is necessary to add? If so, I'd be happy to implement it.

[htemelski-redis]

Currently, it's a bit inconvenient to keep track of the commands executed using a pipeline, so I think such a feature will be beneficial for the library. But let's hear @ndyakov's opinion too.

[ndyakov]

@RasankRam do you need access to the underlying []Cmder or a clone of it? My concerns is that if we grant access to the underlying []Cmder they may be altered and the pipeline may fails.

[RasankRam]

@ndyakov I need access to cmds so that when a savepoint is released, the commands can be moved to the parent pipeline. I understand your concern that someone might “mess with” the underlying array, e.g. cmds[0] = "someCmd".

I’d be fine with something like (if there are something problems with granting cmds):

func (p *Pipeline) Cmds() []redis.Cmder {
    out := append([]redis.Cmder(nil), p.cmds...)
    return out[:len(out):len(out)]
}

If the internal logic of the package makes returning cmds undesirable, I’d also be fine with something like:

func (p *Pipeline) StealCmds() []redis.Cmder {
    cmds := p.cmds
    p.cmds = p.cmds[:0] // or nil
    return cmds
}

All I need is to move commands from a child savepoint into the parent savepoint (if it exists, otherwise into the transaction).

[ndyakov]

@htemelski-redis will have the final say, after a brief discussion I don't think there is an issue to exposing the Cmder slice. At the end of the day, reordering it or altering it can be considered a feature and any issues that may arise are the users responsibility. We should, however, be careful to not allow data races on this slice.

@cxljs once @htemelski-redis decides (he will look more indepth at the implementation at the moment), feel free to work on this, thank you.

[htemelski-redis]

Hello @RasankRam,
The pipeline is already marked as not being thread-safe, and I don't think it is going to be an issue to create a way to expose the underlying queue of commands in the pipeline.
Have you thought of a way to append the commands from one pipeline to another?

Would something like that satisfy your use case?

func (c *Pipeline) AppendTo(dst *Pipeline) {
	dst.cmds = append(dst.cmds, c.cmds...)
	c.Discard()
}

@ndyakov
Also, if we are adding mechanisms to modify the underlying queue, what do you think about adding a Pop method to remove the last command?

[RasankRam]

@htemelski-redis Thank you! It will be quite enough for us. Though it seems a bit narrow for our case. But I can’t think of another situation where commands inside TxPipeline would be needed.

[RasankRam]

@htemelski-redis After giving it some thought, we do need the commands. Since we’ll be bringing the cache into a consistent state by flushing Redis commands, and we’ll be retrieving these commands from the pipeline. The architecture in this case will look like this: commit -> redisPipe.Exec() -failed-> Kafka producer (Kafka receives the set of commands from the pipeline that just need to be flushed).

Then, when the service tries to fetch something from the cache, it will get it from the database and then put the actual data into the cache.

[cxljs]

func (c *Pipeline) AppendTo(dst *Pipeline) {
dst.cmds = append(dst.cmds, c.cmds...)
c.Discard()
}

Pipeline has a similar API:

// Process queues the cmd for later execution.
func (c *Pipeline) Process(ctx context.Context, cmd Cmder) error {
	c.cmds = append(c.cmds, cmd)
	return nil
}

[cxljs]

In the scenario "commit → redisPipe.Exec() failed → Kafka producer", @RasankRam you don't need a new Cmds API, you can just like:

cmds, err := redisPipe.Exec()
if err != nil {
    // send cmds to kafka.
}

Is there any scenario where Cmds need to be obtained before executing the pipeline?

[RasankRam]

@cxljs Regarding redisPipe.Exec, Thank you! I didn’t consider first argument when .Exec(). Then I’m left with one problem: transferring commands when releasing a savepoint into the parent savepoint/transaction