updated goprocess, for periodic
Juan Batiz-Benet committed
Jan 20, 2015 at 05:53 UTC
c43f97d64e8a7431553009ee588ace009a69ab09
11 files changed
+639
-125
Godeps/Godeps.json
+1
-1
@@ -172,7 +172,7 @@
172
},
173
{
174
"ImportPath": "github.com/jbenet/goprocess",
175
- "Rev": "7f96033e206c3cd4e79d1c61cbdfff57869feaf8"
175
+ "Rev": "c37725a4a97d6ad772818b071ceef82789562142"
176
},
177
{
178
"ImportPath": "github.com/kr/binarydist",
Godeps/_workspace/src/github.com/jbenet/goprocess/.travis.yml
new
+11
@@ -0,0 +1,11 @@
1
+language: go
2
+
3
+go:
4
+ - 1.2
5
+ - 1.3
6
+ - 1.4
7
+ - release
8
+ - tip
9
+
10
+script:
11
+ - go test -v ./...
Godeps/_workspace/src/github.com/jbenet/goprocess/LICENSE
new
+21
@@ -0,0 +1,21 @@
1
+The MIT License (MIT)
2
+
3
+Copyright (c) 2014 Juan Batiz-Benet
4
+
5
+Permission is hereby granted, free of charge, to any person obtaining a copy
6
+of this software and associated documentation files (the "Software"), to deal
7
+in the Software without restriction, including without limitation the rights
8
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9
+copies of the Software, and to permit persons to whom the Software is
10
+furnished to do so, subject to the following conditions:
11
+
12
+The above copyright notice and this permission notice shall be included in
13
+all copies or substantial portions of the Software.
14
+
15
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21
+THE SOFTWARE.
Godeps/_workspace/src/github.com/jbenet/goprocess/README.md
+2
@@ -1,5 +1,7 @@
1
# goprocess - lifecycles in go
2
3
+[](https://travis-ci.org/jbenet/goprocess)
4
+
5
(Based on https://github.com/jbenet/go-ctxgroup)
6
7
- Godoc: https://godoc.org/github.com/jbenet/goprocess
Godeps/_workspace/src/github.com/jbenet/goprocess/goprocess.go
+1
-1
@@ -18,7 +18,7 @@ import (
18
// More specifically, it fits this:
19
//
20
// p := WithTeardown(tf) // new process is created, it is now running.
21
-// p.AddChild(q) // can register children **before** Closing.
21
+// p.AddChild(q) // can register children **before** Closed().
22
// go p.Close() // blocks until done running teardown func.
23
// <-p.Closing() // would now return true.
24
// <-p.childrenDone() // wait on all children to be done
Godeps/_workspace/src/github.com/jbenet/goprocess/impl-goroutines.go
deleted
-114
@@ -1,114 +0,0 @@
1
-// +build ignore
2
-
3
-// WARNING: this implementation is not correct.
4
-// here only for historical purposes.
5
-
6
-package goprocess
7
-
8
-import (
9
- "sync"
10
-)
11
-
12
-// process implements Process
13
-type process struct {
14
- children sync.WaitGroup // wait group for child goroutines
15
- teardown TeardownFunc // called to run the teardown logic.
16
- closing chan struct{} // closed once close starts.
17
- closed chan struct{} // closed once close is done.
18
- closeOnce sync.Once // ensure close is only called once.
19
- closeErr error // error to return to clients of Close()
20
-}
21
-
22
-// newProcess constructs and returns a Process.
23
-// It will call tf TeardownFunc exactly once:
24
-// **after** all children have fully Closed,
25
-// **after** entering <-Closing(), and
26
-// **before** <-Closed().
27
-func newProcess(tf TeardownFunc) *process {
28
- if tf == nil {
29
- tf = nilTeardownFunc
30
- }
31
-
32
- return &process{
33
- teardown: tf,
34
- closed: make(chan struct{}),
35
- closing: make(chan struct{}),
36
- }
37
-}
38
-
39
-func (p *process) WaitFor(q Process) {
40
- p.children.Add(1) // p waits on q to be done
41
- go func(p *process, q Process) {
42
- <-q.Closed() // wait until q is closed
43
- p.children.Done() // p done waiting on q
44
- }(p, q)
45
-}
46
-
47
-func (p *process) AddChildNoWait(child Process) {
48
- go func(p, child Process) {
49
- <-p.Closing() // wait until p is closing
50
- child.Close() // close child
51
- }(p, child)
52
-}
53
-
54
-func (p *process) AddChild(child Process) {
55
- select {
56
- case <-p.Closing():
57
- panic("attempt to add child to closing or closed process")
58
- default:
59
- }
60
-
61
- p.children.Add(1) // p waits on child to be done
62
- go func(p *process, child Process) {
63
- <-p.Closing() // wait until p is closing
64
- child.Close() // close child and wait
65
- p.children.Done() // p done waiting on child
66
- }(p, child)
67
-}
68
-
69
-func (p *process) Go(f ProcessFunc) Process {
70
- select {
71
- case <-p.Closing():
72
- panic("attempt to add child to closing or closed process")
73
- default:
74
- }
75
-
76
- // this is very similar to AddChild, but also runs the func
77
- // in the child. we replicate it here to save one goroutine.
78
- child := newProcessGoroutines(nil)
79
- child.children.Add(1) // child waits on func to be done
80
- p.AddChild(child)
81
- go func() {
82
- f(child)
83
- child.children.Done() // wait on child's children to be done.
84
- child.Close() // close to tear down.
85
- }()
86
- return child
87
-}
88
-
89
-// Close is the external close function.
90
-// it's a wrapper around internalClose that waits on Closed()
91
-func (p *process) Close() error {
92
- p.closeOnce.Do(p.doClose)
93
- <-p.Closed() // sync.Once should block, but this checks chan is closed too
94
- return p.closeErr
95
-}
96
-
97
-func (p *process) Closing() <-chan struct{} {
98
- return p.closing
99
-}
100
-
101
-func (p *process) Closed() <-chan struct{} {
102
- return p.closed
103
-}
104
-
105
-// the _actual_ close process.
106
-func (p *process) doClose() {
107
- // this function should only be called once (hence the sync.Once).
108
- // and it will panic (on closing channels) otherwise.
109
-
110
- close(p.closing) // signal that we're shutting down (Closing)
111
- p.children.Wait() // wait till all children are done (before teardown)
112
- p.closeErr = p.teardown() // actually run the close logic (ok safe to teardown)
113
- close(p.closed) // signal that we're shut down (Closed)
114
-}
Godeps/_workspace/src/github.com/jbenet/goprocess/impl-mutex.go
+22
-9
@@ -92,16 +92,18 @@ func (p *process) Go(f ProcessFunc) Process {
92
// it's a wrapper around internalClose that waits on Closed()
93
func (p *process) Close() error {
94
p.Lock()
95
- defer p.Unlock()
95
97
- // if already closed, get out.
96
+ // if already closing, or closed, get out. (but wait!)
97
select {
99
- case <-p.Closed():
98
+ case <-p.Closing():
99
+ p.Unlock()
100
+ <-p.Closed()
101
return p.closeErr
102
default:
103
}
104
105
p.doClose()
106
+ p.Unlock()
107
return p.closeErr
108
}
109
@@ -120,12 +122,23 @@ func (p *process) doClose() {
122
123
close(p.closing) // signal that we're shutting down (Closing)
124
123
- for _, c := range p.children {
124
- go c.Close() // force all children to shut down
125
- }
126
-
127
- for _, w := range p.waitfors {
128
- <-w.Closed() // wait till all waitfors are fully closed (before teardown)
125
+ for len(p.children) > 0 || len(p.waitfors) > 0 {
126
+ for _, c := range p.children {
127
+ go c.Close() // force all children to shut down
128
+ }
129
+ p.children = nil // clear them
130
+
131
+ // we must be careful not to iterate over waitfors directly, as it may
132
+ // change under our feet.
133
+ wf := p.waitfors
134
+ p.waitfors = nil // clear them
135
+ for _, w := range wf {
136
+ // Here, we wait UNLOCKED, so that waitfors who are in the middle of
137
+ // adding a child to us can finish. we will immediately close the child.
138
+ p.Unlock()
139
+ <-w.Closed() // wait till all waitfors are fully closed (before teardown)
140
+ p.Lock()
141
+ }
142
}
143
144
p.closeErr = p.teardown() // actually run the close logic (ok safe to teardown)
Godeps/_workspace/src/github.com/jbenet/goprocess/periodic/README.md
new
+4
@@ -0,0 +1,4 @@
1
+# goprocess/periodic - periodic process creation
2
+
3
+- goprocess: https://github.com/jbenet/goprocess
4
+- Godoc: https://godoc.org/github.com/jbenet/goprocess/periodic
Godeps/_workspace/src/github.com/jbenet/goprocess/periodic/examples_test.go
new
+85
@@ -0,0 +1,85 @@
1
+package periodicproc_test
2
+
3
+import (
4
+ "fmt"
5
+ "time"
6
+
7
+ goprocess "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
8
+ periodicproc "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/periodic"
9
+)
10
+
11
+func ExampleEvery() {
12
+ tock := make(chan struct{})
13
+
14
+ i := 0
15
+ p := periodicproc.Every(time.Second, func(proc goprocess.Process) {
16
+ tock <- struct{}{}
17
+ fmt.Printf("hello %d\n", i)
18
+ i++
19
+ })
20
+
21
+ <-tock
22
+ <-tock
23
+ <-tock
24
+ p.Close()
25
+
26
+ // Output:
27
+ // hello 0
28
+ // hello 1
29
+ // hello 2
30
+}
31
+
32
+func ExampleTick() {
33
+ p := periodicproc.Tick(time.Second, func(proc goprocess.Process) {
34
+ fmt.Println("tick")
35
+ })
36
+
37
+ <-time.After(3*time.Second + 500*time.Millisecond)
38
+ p.Close()
39
+
40
+ // Output:
41
+ // tick
42
+ // tick
43
+ // tick
44
+}
45
+
46
+func ExampleTickGo() {
47
+
48
+ // with TickGo, execution is not rate limited,
49
+ // there can be many in-flight simultaneously
50
+
51
+ wait := make(chan struct{})
52
+ p := periodicproc.TickGo(time.Second, func(proc goprocess.Process) {
53
+ fmt.Println("tick")
54
+ <-wait
55
+ })
56
+
57
+ <-time.After(3*time.Second + 500*time.Millisecond)
58
+
59
+ wait <- struct{}{}
60
+ wait <- struct{}{}
61
+ wait <- struct{}{}
62
+ p.Close() // blocks us until all children are closed.
63
+
64
+ // Output:
65
+ // tick
66
+ // tick
67
+ // tick
68
+}
69
+
70
+func ExampleOnSignal() {
71
+ sig := make(chan struct{})
72
+ p := periodicproc.OnSignal(sig, func(proc goprocess.Process) {
73
+ fmt.Println("fire!")
74
+ })
75
+
76
+ sig <- struct{}{}
77
+ sig <- struct{}{}
78
+ sig <- struct{}{}
79
+ p.Close()
80
+
81
+ // Output:
82
+ // fire!
83
+ // fire!
84
+ // fire!
85
+}
Godeps/_workspace/src/github.com/jbenet/goprocess/periodic/periodic.go
new
+232
@@ -0,0 +1,232 @@
1
+// Package periodic is part of github.com/jbenet/goprocess.
2
+// It provides a simple periodic processor that calls a function
3
+// periodically based on some options.
4
+//
5
+// For example:
6
+//
7
+// // use a time.Duration
8
+// p := periodicproc.Every(time.Second, func(proc goprocess.Process) {
9
+// fmt.Printf("the time is %s and all is well", time.Now())
10
+// })
11
+//
12
+// <-time.After(5*time.Second)
13
+// p.Close()
14
+//
15
+// // use a time.Time channel (like time.Ticker)
16
+// p := periodicproc.Tick(time.Tick(time.Second), func(proc goprocess.Process) {
17
+// fmt.Printf("the time is %s and all is well", time.Now())
18
+// })
19
+//
20
+// <-time.After(5*time.Second)
21
+// p.Close()
22
+//
23
+// // or arbitrary signals
24
+// signal := make(chan struct{})
25
+// p := periodicproc.OnSignal(signal, func(proc goprocess.Process) {
26
+// fmt.Printf("the time is %s and all is well", time.Now())
27
+// })
28
+//
29
+// signal<- struct{}{}
30
+// signal<- struct{}{}
31
+// <-time.After(5 * time.Second)
32
+// signal<- struct{}{}
33
+// p.Close()
34
+//
35
+package periodicproc
36
+
37
+import (
38
+ "time"
39
+
40
+ gp "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
41
+)
42
+
43
+// Every calls the given ProcessFunc at periodic intervals. Internally, it uses
44
+// <-time.After(interval), so it will have the behavior of waiting _at least_
45
+// interval in between calls. If you'd prefer the time.Ticker behavior, use
46
+// periodicproc.Tick instead.
47
+// This is sequentially rate limited, only one call will be in-flight at a time.
48
+func Every(interval time.Duration, procfunc gp.ProcessFunc) gp.Process {
49
+ return gp.Go(func(proc gp.Process) {
50
+ for {
51
+ select {
52
+ case <-time.After(interval):
53
+ select {
54
+ case <-proc.Go(procfunc).Closed(): // spin it out as a child, and wait till it's done.
55
+ case <-proc.Closing(): // we're told to close
56
+ return
57
+ }
58
+ case <-proc.Closing(): // we're told to close
59
+ return
60
+ }
61
+ }
62
+ })
63
+}
64
+
65
+// EveryGo calls the given ProcessFunc at periodic intervals. Internally, it uses
66
+// <-time.After(interval)
67
+// This is not rate limited, multiple calls could be in-flight at the same time.
68
+func EveryGo(interval time.Duration, procfunc gp.ProcessFunc) gp.Process {
69
+ return gp.Go(func(proc gp.Process) {
70
+ for {
71
+ select {
72
+ case <-time.After(interval):
73
+ proc.Go(procfunc)
74
+ case <-proc.Closing(): // we're told to close
75
+ return
76
+ }
77
+ }
78
+ })
79
+}
80
+
81
+// Tick constructs a ticker with interval, and calls the given ProcessFunc every
82
+// time the ticker fires.
83
+// This is sequentially rate limited, only one call will be in-flight at a time.
84
+//
85
+// p := periodicproc.Tick(time.Second, func(proc goprocess.Process) {
86
+// fmt.Println("fire!")
87
+// })
88
+//
89
+// <-time.After(3 * time.Second)
90
+// p.Close()
91
+//
92
+// // Output:
93
+// // fire!
94
+// // fire!
95
+// // fire!
96
+func Tick(interval time.Duration, procfunc gp.ProcessFunc) gp.Process {
97
+ return gp.Go(func(proc gp.Process) {
98
+ ticker := time.NewTicker(interval)
99
+ callOnTicker(ticker.C, procfunc)(proc)
100
+ ticker.Stop()
101
+ })
102
+}
103
+
104
+// TickGo constructs a ticker with interval, and calls the given ProcessFunc every
105
+// time the ticker fires.
106
+// This is not rate limited, multiple calls could be in-flight at the same time.
107
+//
108
+// p := periodicproc.TickGo(time.Second, func(proc goprocess.Process) {
109
+// fmt.Println("fire!")
110
+// <-time.After(10 * time.Second) // will not block sequential execution
111
+// })
112
+//
113
+// <-time.After(3 * time.Second)
114
+// p.Close()
115
+//
116
+// // Output:
117
+// // fire!
118
+// // fire!
119
+// // fire!
120
+func TickGo(interval time.Duration, procfunc gp.ProcessFunc) gp.Process {
121
+ return gp.Go(func(proc gp.Process) {
122
+ ticker := time.NewTicker(interval)
123
+ goCallOnTicker(ticker.C, procfunc)(proc)
124
+ ticker.Stop()
125
+ })
126
+}
127
+
128
+// Ticker calls the given ProcessFunc every time the ticker fires.
129
+// This is sequentially rate limited, only one call will be in-flight at a time.
130
+func Ticker(ticker <-chan time.Time, procfunc gp.ProcessFunc) gp.Process {
131
+ return gp.Go(callOnTicker(ticker, procfunc))
132
+}
133
+
134
+// TickerGo calls the given ProcessFunc every time the ticker fires.
135
+// This is not rate limited, multiple calls could be in-flight at the same time.
136
+func TickerGo(ticker <-chan time.Time, procfunc gp.ProcessFunc) gp.Process {
137
+ return gp.Go(goCallOnTicker(ticker, procfunc))
138
+}
139
+
140
+func callOnTicker(ticker <-chan time.Time, pf gp.ProcessFunc) gp.ProcessFunc {
141
+ return func(proc gp.Process) {
142
+ for {
143
+ select {
144
+ case <-ticker:
145
+ select {
146
+ case <-proc.Go(pf).Closed(): // spin it out as a child, and wait till it's done.
147
+ case <-proc.Closing(): // we're told to close
148
+ return
149
+ }
150
+ case <-proc.Closing(): // we're told to close
151
+ return
152
+ }
153
+ }
154
+ }
155
+}
156
+
157
+func goCallOnTicker(ticker <-chan time.Time, pf gp.ProcessFunc) gp.ProcessFunc {
158
+ return func(proc gp.Process) {
159
+ for {
160
+ select {
161
+ case <-ticker:
162
+ proc.Go(pf)
163
+ case <-proc.Closing(): // we're told to close
164
+ return
165
+ }
166
+ }
167
+ }
168
+}
169
+
170
+// OnSignal calls the given ProcessFunc every time the signal fires, and waits for it to exit.
171
+// This is sequentially rate limited, only one call will be in-flight at a time.
172
+//
173
+// sig := make(chan struct{})
174
+// p := periodicproc.OnSignal(sig, func(proc goprocess.Process) {
175
+// fmt.Println("fire!")
176
+// <-time.After(time.Second) // delays sequential execution by 1 second
177
+// })
178
+//
179
+// sig<- struct{}
180
+// sig<- struct{}
181
+// sig<- struct{}
182
+//
183
+// // Output:
184
+// // fire!
185
+// // fire!
186
+// // fire!
187
+func OnSignal(sig <-chan struct{}, procfunc gp.ProcessFunc) gp.Process {
188
+ return gp.Go(func(proc gp.Process) {
189
+ for {
190
+ select {
191
+ case <-sig:
192
+ select {
193
+ case <-proc.Go(procfunc).Closed(): // spin it out as a child, and wait till it's done.
194
+ case <-proc.Closing(): // we're told to close
195
+ return
196
+ }
197
+ case <-proc.Closing(): // we're told to close
198
+ return
199
+ }
200
+ }
201
+ })
202
+}
203
+
204
+// OnSignalGo calls the given ProcessFunc every time the signal fires.
205
+// This is not rate limited, multiple calls could be in-flight at the same time.
206
+//
207
+// sig := make(chan struct{})
208
+// p := periodicproc.OnSignalGo(sig, func(proc goprocess.Process) {
209
+// fmt.Println("fire!")
210
+// <-time.After(time.Second) // wont block execution
211
+// })
212
+//
213
+// sig<- struct{}
214
+// sig<- struct{}
215
+// sig<- struct{}
216
+//
217
+// // Output:
218
+// // fire!
219
+// // fire!
220
+// // fire!
221
+func OnSignalGo(sig <-chan struct{}, procfunc gp.ProcessFunc) gp.Process {
222
+ return gp.Go(func(proc gp.Process) {
223
+ for {
224
+ select {
225
+ case <-sig:
226
+ proc.Go(procfunc)
227
+ case <-proc.Closing(): // we're told to close
228
+ return
229
+ }
230
+ }
231
+ })
232
+}
Godeps/_workspace/src/github.com/jbenet/goprocess/periodic/periodic_test.go
new
+260
@@ -0,0 +1,260 @@
1
+package periodicproc
2
+
3
+import (
4
+ "testing"
5
+ "time"
6
+
7
+ ci "github.com/jbenet/go-cienv"
8
+ gp "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
9
+)
10
+
11
+var (
12
+ grace = time.Millisecond * 5
13
+ interval = time.Millisecond * 10
14
+ timeout = time.Second * 5
15
+)
16
+
17
+func init() {
18
+ if ci.IsRunning() {
19
+ grace = time.Millisecond * 500
20
+ interval = time.Millisecond * 1000
21
+ timeout = time.Second * 15
22
+ }
23
+}
24
+
25
+func between(min, diff, max time.Duration) bool {
26
+ return min <= diff && diff <= max
27
+}
28
+
29
+func testBetween(t *testing.T, min, diff, max time.Duration) {
30
+ if !between(min, diff, max) {
31
+ t.Error("time diff incorrect:", min, diff, max)
32
+ }
33
+}
34
+
35
+type intervalFunc func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process)
36
+
37
+func testSeq(t *testing.T, toTest intervalFunc) {
38
+ t.Parallel()
39
+
40
+ last := time.Now()
41
+ times := make(chan time.Time, 10)
42
+ p := toTest(times, nil)
43
+
44
+ for i := 0; i < 5; i++ {
45
+ next := <-times
46
+ testBetween(t, interval-grace, next.Sub(last), interval+grace)
47
+ last = next
48
+ }
49
+
50
+ go p.Close()
51
+ select {
52
+ case <-p.Closed():
53
+ case <-time.After(timeout):
54
+ t.Error("proc failed to close")
55
+ }
56
+}
57
+
58
+func testSeqWait(t *testing.T, toTest intervalFunc) {
59
+ t.Parallel()
60
+
61
+ last := time.Now()
62
+ times := make(chan time.Time, 10)
63
+ wait := make(chan struct{})
64
+ p := toTest(times, wait)
65
+
66
+ for i := 0; i < 5; i++ {
67
+ next := <-times
68
+ testBetween(t, interval-grace, next.Sub(last), interval+grace)
69
+
70
+ <-time.After(interval * 2) // make it wait.
71
+ last = time.Now() // make it now (sequential)
72
+ wait <- struct{}{} // release it.
73
+ }
74
+
75
+ go p.Close()
76
+
77
+ select {
78
+ case <-p.Closed():
79
+ case <-time.After(timeout):
80
+ t.Error("proc failed to close")
81
+ }
82
+}
83
+
84
+func testSeqNoWait(t *testing.T, toTest intervalFunc) {
85
+ t.Parallel()
86
+
87
+ last := time.Now()
88
+ times := make(chan time.Time, 10)
89
+ wait := make(chan struct{})
90
+ p := toTest(times, wait)
91
+
92
+ for i := 0; i < 5; i++ {
93
+ next := <-times
94
+ testBetween(t, 0, next.Sub(last), interval+grace) // min of 0
95
+
96
+ <-time.After(interval * 2) // make it wait.
97
+ last = time.Now() // make it now (sequential)
98
+ wait <- struct{}{} // release it.
99
+ }
100
+
101
+ go p.Close()
102
+
103
+end:
104
+ select {
105
+ case wait <- struct{}{}: // drain any extras.
106
+ goto end
107
+ case <-p.Closed():
108
+ case <-time.After(timeout):
109
+ t.Error("proc failed to close")
110
+ }
111
+}
112
+
113
+func testParallel(t *testing.T, toTest intervalFunc) {
114
+ t.Parallel()
115
+
116
+ last := time.Now()
117
+ times := make(chan time.Time, 10)
118
+ wait := make(chan struct{})
119
+ p := toTest(times, wait)
120
+
121
+ for i := 0; i < 5; i++ {
122
+ next := <-times
123
+ testBetween(t, interval-grace, next.Sub(last), interval+grace)
124
+ last = next
125
+
126
+ <-time.After(interval * 2) // make it wait.
127
+ wait <- struct{}{} // release it.
128
+ }
129
+
130
+ go p.Close()
131
+
132
+end:
133
+ select {
134
+ case wait <- struct{}{}: // drain any extras.
135
+ goto end
136
+ case <-p.Closed():
137
+ case <-time.After(timeout):
138
+ t.Error("proc failed to close")
139
+ }
140
+}
141
+
142
+func TestEverySeq(t *testing.T) {
143
+ testSeq(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
144
+ return Every(interval, func(proc gp.Process) {
145
+ times <- time.Now()
146
+ })
147
+ })
148
+}
149
+
150
+func TestEverySeqWait(t *testing.T) {
151
+ testSeqWait(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
152
+ return Every(interval, func(proc gp.Process) {
153
+ times <- time.Now()
154
+ select {
155
+ case <-wait:
156
+ case <-proc.Closing():
157
+ }
158
+ })
159
+ })
160
+}
161
+
162
+func TestEveryGoSeq(t *testing.T) {
163
+ testSeq(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
164
+ return EveryGo(interval, func(proc gp.Process) {
165
+ times <- time.Now()
166
+ })
167
+ })
168
+}
169
+
170
+func TestEveryGoSeqParallel(t *testing.T) {
171
+ testParallel(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
172
+ return EveryGo(interval, func(proc gp.Process) {
173
+ times <- time.Now()
174
+ select {
175
+ case <-wait:
176
+ case <-proc.Closing():
177
+ }
178
+ })
179
+ })
180
+}
181
+
182
+func TestTickSeq(t *testing.T) {
183
+ testSeq(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
184
+ return Tick(interval, func(proc gp.Process) {
185
+ times <- time.Now()
186
+ })
187
+ })
188
+}
189
+
190
+func TestTickSeqNoWait(t *testing.T) {
191
+ testSeqNoWait(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
192
+ return Tick(interval, func(proc gp.Process) {
193
+ times <- time.Now()
194
+ select {
195
+ case <-wait:
196
+ case <-proc.Closing():
197
+ }
198
+ })
199
+ })
200
+}
201
+
202
+func TestTickGoSeq(t *testing.T) {
203
+ testSeq(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
204
+ return TickGo(interval, func(proc gp.Process) {
205
+ times <- time.Now()
206
+ })
207
+ })
208
+}
209
+
210
+func TestTickGoSeqParallel(t *testing.T) {
211
+ testParallel(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
212
+ return TickGo(interval, func(proc gp.Process) {
213
+ times <- time.Now()
214
+ select {
215
+ case <-wait:
216
+ case <-proc.Closing():
217
+ }
218
+ })
219
+ })
220
+}
221
+
222
+func TestTickerSeq(t *testing.T) {
223
+ testSeq(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
224
+ return Ticker(time.Tick(interval), func(proc gp.Process) {
225
+ times <- time.Now()
226
+ })
227
+ })
228
+}
229
+
230
+func TestTickerSeqNoWait(t *testing.T) {
231
+ testSeqNoWait(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
232
+ return Ticker(time.Tick(interval), func(proc gp.Process) {
233
+ times <- time.Now()
234
+ select {
235
+ case <-wait:
236
+ case <-proc.Closing():
237
+ }
238
+ })
239
+ })
240
+}
241
+
242
+func TestTickerGoSeq(t *testing.T) {
243
+ testSeq(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
244
+ return TickerGo(time.Tick(interval), func(proc gp.Process) {
245
+ times <- time.Now()
246
+ })
247
+ })
248
+}
249
+
250
+func TestTickerGoParallel(t *testing.T) {
251
+ testParallel(t, func(times chan<- time.Time, wait <-chan struct{}) (proc gp.Process) {
252
+ return TickerGo(time.Tick(interval), func(proc gp.Process) {
253
+ times <- time.Now()
254
+ select {
255
+ case <-wait:
256
+ case <-proc.Closing():
257
+ }
258
+ })
259
+ })
260
+}