/
source.go
54 lines (45 loc) · 1.13 KB
/
source.go
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
package main
import (
"context"
"github.com/yangsijie666/controller-manager/pkg/event"
"github.com/yangsijie666/controller-manager/pkg/handler"
"github.com/yangsijie666/controller-manager/pkg/predicate"
"github.com/yangsijie666/controller-manager/pkg/source"
"k8s.io/client-go/util/workqueue"
"math/rand"
"time"
)
/**
* @Author: yangsijie666
* @Date: 2023/5/24 15:08
*/
var _ source.Source[int] = &Source{}
type Source struct {
}
func (s *Source) Start(ctx context.Context, evtHdr handler.EventHandler[int], q workqueue.RateLimitingInterface, _ ...predicate.Predicate[int]) error {
go func(ctx context.Context) {
t := time.NewTicker(time.Second * 5)
defer t.Stop()
for {
s.fetchPendingObjects(ctx, evtHdr, q)
select {
case <-ctx.Done():
return
case <-t.C:
continue
}
}
}(ctx)
return nil
}
func (s *Source) fetchPendingObjects(ctx context.Context, hdr handler.EventHandler[int], q workqueue.RateLimitingInterface) {
defer func() {
if err := recover(); err != nil {
panic(err)
}
}()
rand.Seed(time.Now().UnixNano())
hdr.Generic(ctx, event.GenericEvent[int]{
Object: rand.Intn(100),
}, q)
}