Browse Source

return back output internal channel (way go does with Timer)

pull/1842/head
Anton Kaliaev 7 years ago
parent
commit
3779310c72
No known key found for this signature in database GPG Key ID: 7B6881D965918214
2 changed files with 30 additions and 19 deletions
  1. +29
    -18
      common/throttle_timer.go
  2. +1
    -1
      common/throttle_timer_test.go

+ 29
- 18
common/throttle_timer.go View File

@ -11,10 +11,11 @@ If a long continuous burst of .Set() calls happens, ThrottleTimer fires
at most once every "dur".
*/
type ThrottleTimer struct {
Name string
Ch chan struct{}
input chan command
dur time.Duration
Name string
Ch <-chan struct{}
input chan command
output chan<- struct{}
dur time.Duration
timer *time.Timer
isSet bool
@ -28,25 +29,22 @@ const (
Quit
)
// NewThrottleTimer creates a new ThrottleTimer.
func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer {
c := make(chan struct{})
var t = &ThrottleTimer{
Name: name,
Ch: make(chan struct{}),
dur: dur,
input: make(chan command),
timer: time.NewTimer(dur),
Name: name,
Ch: c,
dur: dur,
input: make(chan command),
output: c,
timer: time.NewTimer(dur),
}
t.timer.Stop()
go t.run()
return t
}
// C is the proper way to listen to the timer output.
// t.Ch will be made private in the (near?) future
func (t *ThrottleTimer) C() <-chan struct{} {
return t.Ch
}
func (t *ThrottleTimer) run() {
for {
select {
@ -65,7 +63,7 @@ func (t *ThrottleTimer) run() {
// trySend performs non-blocking send on t.Ch
func (t *ThrottleTimer) trySend() {
select {
case t.Ch <- struct{}{}:
case t.output <- struct{}{}:
t.isSet = false
default:
// if we just want to drop, replace this with t.isSet = false
@ -105,8 +103,21 @@ func (t *ThrottleTimer) Unset() {
t.input <- Unset
}
// For ease of .Stop()'ing services before .Start()'ing them,
// we ignore .Stop()'s on nil ThrottleTimers
// Stop prevents the ThrottleTimer from firing. It always returns true. Stop does not
// close the channel, to prevent a read from the channel succeeding
// incorrectly.
//
// To prevent a timer created with NewThrottleTimer from firing after a call to
// Stop, check the return value and drain the channel.
//
// For example, assuming the program has not received from t.C already:
//
// if !t.Stop() {
// <-t.C
// }
//
// For ease of stopping services before starting them, we ignore Stop on nil
// ThrottleTimers.
func (t *ThrottleTimer) Stop() bool {
if t == nil {
return false


+ 1
- 1
common/throttle_timer_test.go View File

@ -49,7 +49,7 @@ func TestThrottle(test *testing.T) {
t := NewThrottleTimer("foo", delay)
// start at 0
c := &thCounter{input: t.C()}
c := &thCounter{input: t.Ch}
assert.Equal(0, c.Count())
go c.Read()


Loading…
Cancel
Save