mirror of
https://github.com/rancher/steve.git
synced 2025-08-11 03:01:38 +00:00
Only send counts every second if there is a change
This commit is contained in:
parent
ece0f7bce5
commit
e27f384795
40
pkg/resources/counts/buffer.go
Normal file
40
pkg/resources/counts/buffer.go
Normal file
@ -0,0 +1,40 @@
|
|||||||
|
package counts
|
||||||
|
|
||||||
|
import (
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/rancher/apiserver/pkg/types"
|
||||||
|
)
|
||||||
|
|
||||||
|
func buffer(c chan types.APIEvent) chan types.APIEvent {
|
||||||
|
result := make(chan types.APIEvent)
|
||||||
|
go func() {
|
||||||
|
defer close(result)
|
||||||
|
debounce(result, c)
|
||||||
|
}()
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
func debounce(result, input chan types.APIEvent) {
|
||||||
|
t := time.NewTicker(time.Second)
|
||||||
|
defer t.Stop()
|
||||||
|
|
||||||
|
var (
|
||||||
|
lastEvent *types.APIEvent
|
||||||
|
)
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case event, ok := <-input:
|
||||||
|
if ok {
|
||||||
|
lastEvent = &event
|
||||||
|
} else {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
case <-t.C:
|
||||||
|
if lastEvent != nil {
|
||||||
|
result <- *lastEvent
|
||||||
|
lastEvent = nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
@ -118,6 +118,14 @@ func (s *Store) Watch(apiOp *types.APIRequest, schema *types.APISchema, w types.
|
|||||||
countLock sync.Mutex
|
countLock sync.Mutex
|
||||||
)
|
)
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
<-apiOp.Context().Done()
|
||||||
|
countLock.Lock()
|
||||||
|
close(result)
|
||||||
|
result = nil
|
||||||
|
countLock.Unlock()
|
||||||
|
}()
|
||||||
|
|
||||||
counts = s.getCount(apiOp).Counts
|
counts = s.getCount(apiOp).Counts
|
||||||
for id := range counts {
|
for id := range counts {
|
||||||
schema := apiOp.Schemas.LookupSchema(id)
|
schema := apiOp.Schemas.LookupSchema(id)
|
||||||
@ -128,14 +136,6 @@ func (s *Store) Watch(apiOp *types.APIRequest, schema *types.APISchema, w types.
|
|||||||
gvrToSchema[attributes.GVR(schema)] = schema
|
gvrToSchema[attributes.GVR(schema)] = schema
|
||||||
}
|
}
|
||||||
|
|
||||||
go func() {
|
|
||||||
<-apiOp.Context().Done()
|
|
||||||
countLock.Lock()
|
|
||||||
close(result)
|
|
||||||
result = nil
|
|
||||||
countLock.Unlock()
|
|
||||||
}()
|
|
||||||
|
|
||||||
onChange := func(add bool, gvr schema2.GroupVersionResource, _ string, obj, oldObj runtime.Object) error {
|
onChange := func(add bool, gvr schema2.GroupVersionResource, _ string, obj, oldObj runtime.Object) error {
|
||||||
countLock.Lock()
|
countLock.Lock()
|
||||||
defer countLock.Unlock()
|
defer countLock.Unlock()
|
||||||
@ -205,7 +205,7 @@ func (s *Store) Watch(apiOp *types.APIRequest, schema *types.APISchema, w types.
|
|||||||
return onChange(false, gvr, key, obj, nil)
|
return onChange(false, gvr, key, obj, nil)
|
||||||
})
|
})
|
||||||
|
|
||||||
return result, nil
|
return buffer(result), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Store) schemasToWatch(apiOp *types.APIRequest) (result []*types.APISchema) {
|
func (s *Store) schemasToWatch(apiOp *types.APIRequest) (result []*types.APISchema) {
|
||||||
|
Loading…
Reference in New Issue
Block a user