Repository navigation
Expand file tree
/
Copy pathprogress.go
More file actions
112 lines (98 loc) · 3.5 KB
/
Copy pathprogress.go
File metadata and controls
112 lines (98 loc) · 3.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
package cli
import (
"fmt"
"io"
"sync"
"time"
"github.com/ucloud/ucloud-sdk-go/ucloud/request"
"github.com/ucloud/ucloud-cli/pkg/ui"
)
// Block is the platform alias for ui.Block, so product packages get the
// concurrent-progress block type without importing platform internals.
type Block = ui.Block
// Progress is a per-invocation concurrent-progress session bound to the ctx
// progress writer (stdout in table mode, stderr in json/yaml).
type Progress struct {
out io.Writer
doc *ui.Document
}
// NewProgress builds a Progress bound to the ctx progress writer. Non-TTY
// writers suppress animation (handled inside ui.NewDocument).
func (c *Context) NewProgress() *Progress {
w := c.ProgressWriter()
return &Progress{out: w, doc: ui.NewDocument(w)}
}
// Disable switches off per-block animation (the count>5 aggregate path).
func (p *Progress) Disable() { p.doc.Disable() }
// Animated reports whether the progress document renders live frames. It is
// false when the bound writer is not a TTY (pipe/file/json mode) or Disable()
// was called for the aggregate count>5 path. When false, block content is never
// shown, so callers must surface errors to stderr (ctx.Err()) themselves.
func (p *Progress) Animated() bool { return !p.doc.Disabled() }
// NewBlock appends a fresh block to the document and returns it.
func (p *Progress) NewBlock() *Block {
b := ui.NewBlock()
p.doc.Append(b)
return b
}
// Refresh prints an aggregate counter line to the progress writer.
func (p *Progress) Refresh(text string) { ui.NewRefresh(p.out).Do(text) }
// Sspoll runs the concurrent poller into block, bound to the progress writer.
func (p *Progress) Sspoll(describe func(string, *request.CommonBase) (interface{}, error),
resourceID, text string, targetStates []string, block *Block, common *request.CommonBase, opts ...PollerOption) {
NewPoller(describe, p.out, opts...).Sspoll(resourceID, text, targetStates, block, common)
}
// ConcurrentAction runs actionFunc over reqs with bounded concurrency (limit),
// aggregating a refresh counter when count>5. It is a verbatim port of
// cmd/util.go concurrentAction, rebound to the ctx progress writer. Products
// call this instead of touching platform internals.
func (c *Context) ConcurrentAction(reqs []request.Common, limit int, actionFunc func(request.Common) (bool, []string)) {
if limit <= 0 {
limit = 10
}
w := c.ProgressWriter()
refresh := ui.NewRefresh(w)
count := len(reqs)
var wg sync.WaitGroup
result := make(chan bool)
tokens := make(chan bool, limit) // 控制并发量，最多 limit 个并发
success, fail := 0, 0
// 同时执行任务数量大于 5 时，不再单独显示每个任务，而是聚合显示。
if count > 5 {
refresh.Do(fmt.Sprintf("total:%d, doing:%d, success:%d, fail:%d", count, len(tokens), success, fail))
}
go func() {
for {
select {
case ret := <-result:
if ret {
success++
} else {
fail++
}
case <-time.Tick(time.Second / 30):
if count == (success+fail) && fail > 0 {
fmt.Fprintf(w, "Check logs in %s\n", c.LogFilePath())
return
}
if count > 5 {
refresh.Do(fmt.Sprintf("total:%d, doing:%d, success:%d, fail:%d", count, len(tokens), success, fail))
}
}
}
}()
for _, req := range reqs {
wg.Add(1)
go func(req request.Common) {
tokens <- true
ok, logs := actionFunc(req)
result <- ok
logs = append([]string{"========================================"}, logs...)
c.LogInfo(logs...)
<-tokens
time.Sleep(time.Second / 5)
wg.Done()
}(req)
}
wg.Wait()
}