mirror of
https://github.com/fluencelabs/tendermint
synced 2025-06-12 21:01:21 +00:00
One more attempt with a read-only channel
This commit is contained in:
@ -11,11 +11,10 @@ If a long continuous burst of .Set() calls happens, ThrottleTimer fires
|
|||||||
at most once every "dur".
|
at most once every "dur".
|
||||||
*/
|
*/
|
||||||
type ThrottleTimer struct {
|
type ThrottleTimer struct {
|
||||||
Name string
|
Name string
|
||||||
Ch <-chan struct{}
|
Ch chan struct{}
|
||||||
output chan<- struct{}
|
input chan command
|
||||||
input chan command
|
dur time.Duration
|
||||||
dur time.Duration
|
|
||||||
|
|
||||||
timer *time.Timer
|
timer *time.Timer
|
||||||
isSet bool
|
isSet bool
|
||||||
@ -30,27 +29,31 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer {
|
func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer {
|
||||||
c := make(chan struct{})
|
|
||||||
var t = &ThrottleTimer{
|
var t = &ThrottleTimer{
|
||||||
Name: name,
|
Name: name,
|
||||||
Ch: c,
|
Ch: make(chan struct{}),
|
||||||
dur: dur,
|
dur: dur,
|
||||||
output: c,
|
input: make(chan command),
|
||||||
input: make(chan command),
|
timer: time.NewTimer(dur),
|
||||||
timer: time.NewTimer(dur),
|
|
||||||
}
|
}
|
||||||
t.timer.Stop()
|
t.timer.Stop()
|
||||||
go t.run()
|
go t.run()
|
||||||
return t
|
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() {
|
func (t *ThrottleTimer) run() {
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case cmd := <-t.input:
|
case cmd := <-t.input:
|
||||||
// stop goroutine if the input says so
|
// stop goroutine if the input says so
|
||||||
if t.processInput(cmd) {
|
if t.processInput(cmd) {
|
||||||
close(t.output)
|
close(t.Ch)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
case <-t.timer.C:
|
case <-t.timer.C:
|
||||||
@ -59,10 +62,10 @@ func (t *ThrottleTimer) run() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// trySend performs non-blocking send on t.output (t.Ch)
|
// trySend performs non-blocking send on t.Ch
|
||||||
func (t *ThrottleTimer) trySend() {
|
func (t *ThrottleTimer) trySend() {
|
||||||
select {
|
select {
|
||||||
case t.output <- struct{}{}:
|
case t.Ch <- struct{}{}:
|
||||||
t.isSet = false
|
t.isSet = false
|
||||||
default:
|
default:
|
||||||
// if we just want to drop, replace this with t.isSet = false
|
// if we just want to drop, replace this with t.isSet = false
|
||||||
|
@ -46,7 +46,7 @@ func TestThrottle(test *testing.T) {
|
|||||||
t := NewThrottleTimer("foo", delay)
|
t := NewThrottleTimer("foo", delay)
|
||||||
|
|
||||||
// start at 0
|
// start at 0
|
||||||
c := &thCounter{input: t.Ch}
|
c := &thCounter{input: t.C()}
|
||||||
assert.Equal(0, c.Count())
|
assert.Equal(0, c.Count())
|
||||||
go c.Read()
|
go c.Read()
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user