Skip to content

Commit 7418dad

Browse files
skeeeyskeeey
committed
using independent transports for ce
Signed-off-by: Wei Liu <liuweixa@redhat.com>
1 parent 9493207 commit 7418dad

13 files changed

Lines changed: 2290 additions & 42 deletions

File tree

‎pkg/cloudevents/clients/options/generic.go‎

Lines changed: 72 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@ import (
1212
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/clients"
1313
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/builder"
1414
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
15+
clientsv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/v2/clients"
16+
builderv2 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/v2/options/builder"
1517
)
1618

1719
type GenericClientOptions[T generic.ResourceObject] struct {
@@ -23,6 +25,7 @@ type GenericClientOptions[T generic.ResourceObject] struct {
2325
clusterName string
2426
subscription bool
2527
resync bool
28+
v2 bool
2629
}
2730

2831
// NewGenericClientOptions create a GenericClientOptions
@@ -86,6 +89,11 @@ func (o *GenericClientOptions[T]) WithResyncEnabled(resync bool) *GenericClientO
8689
return o
8790
}
8891

92+
func (o *GenericClientOptions[T]) EnableV2() *GenericClientOptions[T] {
93+
o.v2 = true
94+
return o
95+
}
96+
8997
func (o *GenericClientOptions[T]) ClusterName() string {
9098
return o.clusterName
9199
}
@@ -111,20 +119,39 @@ func (o *GenericClientOptions[T]) AgentClient(ctx context.Context) (generic.Clou
111119
o.watcherStore = store.NewAgentInformerWatcherStore[T]()
112120
}
113121

114-
options, err := builder.BuildCloudEventsAgentOptions(o.config, o.clusterName, o.clientID)
115-
if err != nil {
116-
return nil, err
117-
}
122+
var cloudEventsClient generic.CloudEventsClient[T]
123+
if o.v2 {
124+
options, err := builderv2.BuildCloudEventsAgentOptions(o.config, o.clusterName, o.clientID)
125+
if err != nil {
126+
return nil, err
127+
}
128+
129+
cloudEventsClient, err = clientsv2.NewCloudEventAgentClient(
130+
ctx,
131+
options,
132+
store.NewAgentWatcherStoreLister(o.watcherStore),
133+
statushash.StatusHash,
134+
o.codec,
135+
)
136+
if err != nil {
137+
return nil, err
138+
}
139+
} else {
140+
options, err := builder.BuildCloudEventsAgentOptions(o.config, o.clusterName, o.clientID)
141+
if err != nil {
142+
return nil, err
143+
}
118144

119-
cloudEventsClient, err := clients.NewCloudEventAgentClient(
120-
ctx,
121-
options,
122-
store.NewAgentWatcherStoreLister(o.watcherStore),
123-
statushash.StatusHash,
124-
o.codec,
125-
)
126-
if err != nil {
127-
return nil, err
145+
cloudEventsClient, err = clients.NewCloudEventAgentClient(
146+
ctx,
147+
options,
148+
store.NewAgentWatcherStoreLister(o.watcherStore),
149+
statushash.StatusHash,
150+
o.codec,
151+
)
152+
if err != nil {
153+
return nil, err
154+
}
128155
}
129156

130157
if o.subscription {
@@ -182,20 +209,39 @@ func (o *GenericClientOptions[T]) SourceClient(ctx context.Context) (generic.Clo
182209
return nil, fmt.Errorf("a watcher store is required")
183210
}
184211

185-
options, err := builder.BuildCloudEventsSourceOptions(o.config, o.clientID, o.sourceID)
186-
if err != nil {
187-
return nil, err
188-
}
212+
var cloudEventsClient generic.CloudEventsClient[T]
213+
if o.v2 {
214+
options, err := builderv2.BuildCloudEventsSourceOptions(o.config, o.clientID, o.sourceID)
215+
if err != nil {
216+
return nil, err
217+
}
189218

190-
cloudEventsClient, err := clients.NewCloudEventSourceClient(
191-
ctx,
192-
options,
193-
store.NewSourceWatcherStoreLister(o.watcherStore),
194-
statushash.StatusHash,
195-
o.codec,
196-
)
197-
if err != nil {
198-
return nil, err
219+
cloudEventsClient, err = clientsv2.NewCloudEventSourceClient(
220+
ctx,
221+
options,
222+
store.NewSourceWatcherStoreLister(o.watcherStore),
223+
statushash.StatusHash,
224+
o.codec,
225+
)
226+
if err != nil {
227+
return nil, err
228+
}
229+
} else {
230+
options, err := builder.BuildCloudEventsSourceOptions(o.config, o.clientID, o.sourceID)
231+
if err != nil {
232+
return nil, err
233+
}
234+
235+
cloudEventsClient, err = clients.NewCloudEventSourceClient(
236+
ctx,
237+
options,
238+
store.NewSourceWatcherStoreLister(o.watcherStore),
239+
statushash.StatusHash,
240+
o.codec,
241+
)
242+
if err != nil {
243+
return nil, err
244+
}
199245
}
200246

201247
if o.subscription {

‎pkg/cloudevents/generic/options/options.go‎

Lines changed: 3 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
cloudevents "github.com/cloudevents/sdk-go/v2"
77
"github.com/cloudevents/sdk-go/v2/protocol"
88
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
9+
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/utils"
910
)
1011

1112
// CloudEventsOptions provides cloudevents clients to send/receive cloudevents based on different event protocol.
@@ -35,17 +36,6 @@ type CloudEventsProtocol interface {
3536
protocol.Closer
3637
}
3738

38-
// EventRateLimit for limiting the event sending rate.
39-
type EventRateLimit struct {
40-
// QPS indicates the maximum QPS to send the event.
41-
// If it's less than or equal to zero, the DefaultQPS (50) will be used.
42-
QPS float32
43-
44-
// Maximum burst for throttle.
45-
// If it's less than or equal to zero, the DefaultBurst (100) will be used.
46-
Burst int
47-
}
48-
4939
// CloudEventsSourceOptions provides the required options to build a source CloudEventsClient
5040
type CloudEventsSourceOptions struct {
5141
// CloudEventsOptions provides cloudevents clients to send/receive cloudevents based on different event protocol.
@@ -57,7 +47,7 @@ type CloudEventsSourceOptions struct {
5747
SourceID string
5848

5949
// EventRateLimit limits the event sending rate.
60-
EventRateLimit EventRateLimit
50+
EventRateLimit utils.EventRateLimit
6151
}
6252

6353
// CloudEventsAgentOptions provides the required options to build an agent CloudEventsClient
@@ -73,5 +63,5 @@ type CloudEventsAgentOptions struct {
7363
ClusterName string
7464

7565
// EventRateLimit limits the event sending rate.
76-
EventRateLimit EventRateLimit
66+
EventRateLimit utils.EventRateLimit
7767
}

‎pkg/cloudevents/generic/utils/ratelimiter.go‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,6 @@ import (
44
"time"
55

66
"k8s.io/client-go/util/flowcontrol"
7-
8-
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
97
)
108

119
// LongThrottleLatency defines threshold for logging requests. All requests being
@@ -19,7 +17,18 @@ const (
1917
DefaultBurst int = 100
2018
)
2119

22-
func NewRateLimiter(limit options.EventRateLimit) flowcontrol.RateLimiter {
20+
// EventRateLimit for limiting the event sending rate.
21+
type EventRateLimit struct {
22+
// QPS indicates the maximum QPS to send the event.
23+
// If it's less than or equal to zero, the DefaultQPS (50) will be used.
24+
QPS float32
25+
26+
// Maximum burst for throttle.
27+
// If it's less than or equal to zero, the DefaultBurst (100) will be used.
28+
Burst int
29+
}
30+
31+
func NewRateLimiter(limit EventRateLimit) flowcontrol.RateLimiter {
2332
qps := limit.QPS
2433
if qps <= 0.0 {
2534
qps = DefaultQPS

0 commit comments

Comments
 (0)