Exec all the commands queued in client
(ctx *Context)
| 20 | |
| 21 | // Exec all the commands queued in client |
| 22 | func Exec(ctx *Context) { |
| 23 | ctx.Client.Multi = false |
| 24 | commands := ctx.Client.Commands |
| 25 | if len(commands) == 0 { |
| 26 | resp.ReplyArray(ctx.Out, 0) |
| 27 | return |
| 28 | } |
| 29 | ctx.Client.Commands = nil |
| 30 | |
| 31 | // Has watch command been issued |
| 32 | watching := ctx.Client.Txn != nil |
| 33 | txn := ctx.Client.Txn |
| 34 | ctx.Client.Txn = nil |
| 35 | |
| 36 | size := len(commands) |
| 37 | var err error |
| 38 | var outputs []*bytes.Buffer |
| 39 | var onCommits []OnCommit |
| 40 | err = retry.Ensure(ctx, func() error { |
| 41 | if !watching { |
| 42 | txn, err = ctx.Client.DB.Begin() |
| 43 | if err != nil { |
| 44 | zap.L().Error("begin txn failed", |
| 45 | zap.Int64("clientid", ctx.Client.ID), |
| 46 | zap.String("command", ctx.Name), |
| 47 | zap.String("traceid", ctx.TraceID), |
| 48 | zap.Error(err)) |
| 49 | resp.ReplyArray(ctx.Out, 0) |
| 50 | return err |
| 51 | } |
| 52 | } |
| 53 | outputs = make([]*bytes.Buffer, size) |
| 54 | onCommits = make([]OnCommit, size) |
| 55 | commandCount := 0 |
| 56 | for i, cmd := range commands { |
| 57 | var onCommit OnCommit |
| 58 | out := bytes.NewBuffer(nil) |
| 59 | subCtx := &Context{ |
| 60 | Name: cmd.Name, |
| 61 | Args: cmd.Args, |
| 62 | In: ctx.In, |
| 63 | Out: out, |
| 64 | Context: ctx.Context, |
| 65 | } |
| 66 | name := strings.ToLower(cmd.Name) |
| 67 | if _, ok := txnCommands[name]; ok { |
| 68 | start := time.Now() |
| 69 | onCommit, err = TxnCall(subCtx, txn) |
| 70 | zap.L().Debug("execute", zap.String("command", subCtx.Name), zap.Int64("cost(us)", time.Since(start).Nanoseconds()/1000)) |
| 71 | if err != nil { |
| 72 | resp.ReplyError(out, err.Error()) |
| 73 | } |
| 74 | } else { |
| 75 | Call(subCtx) |
| 76 | } |
| 77 | onCommits[i] = onCommit |
| 78 | outputs[i] = out |
| 79 | commandCount++ |
no test coverage detected