@@ -32,6 +32,7 @@ import (
3232 "github.com/linkall-labs/vanus/internal/controller/eventbus/volume"
3333 "github.com/linkall-labs/vanus/internal/kv"
3434 "github.com/linkall-labs/vanus/internal/kv/etcd"
35+ "github.com/linkall-labs/vanus/internal/primitive"
3536 "github.com/linkall-labs/vanus/internal/primitive/vanus"
3637 "github.com/linkall-labs/vanus/observability/log"
3738 "github.com/linkall-labs/vanus/observability/metrics"
@@ -94,7 +95,6 @@ func (ctrl *controller) Start(_ context.Context) error {
9495 return err
9596 }
9697 ctrl .kvStore = store
97-
9898 ctrl .cancelCtx , ctrl .cancelFunc = context .WithCancel (context .Background ())
9999 go ctrl .member .RegisterMembershipChangedProcessor (ctrl .membershipChangedProcessor )
100100 return nil
@@ -113,9 +113,48 @@ func (ctrl *controller) StopNotify() <-chan error {
113113}
114114
115115func (ctrl * controller ) CreateEventBus (ctx context.Context ,
116+ req * ctrlpb.CreateEventBusRequest ) (* metapb.EventBus , error ) {
117+ if err := isValidEventbusName (req .Name ); err != nil {
118+ return nil , err
119+ }
120+ return ctrl .createEventBus (ctx , req )
121+ }
122+
123+ func isValidEventbusName (name string ) error {
124+ name = strings .ToLower (name )
125+ for _ , v := range name {
126+ if v == '.' || v == '_' || v == '-' {
127+ continue
128+ }
129+ c := v - 'a'
130+ if c >= 0 || c <= 26 {
131+ continue
132+ } else {
133+ c = v - '0'
134+ if c >= 0 || c <= 9 {
135+ continue
136+ }
137+ return errors .ErrInvalidRequest .WithMessage ("eventbus name must be insist of 0-9a-zA-Z.-_" )
138+ }
139+ }
140+ return nil
141+ }
142+
143+ func (ctrl * controller ) CreateSystemEventBus (ctx context.Context ,
144+ req * ctrlpb.CreateEventBusRequest ) (* metapb.EventBus , error ) {
145+ if ! strings .HasPrefix (req .Name , primitive .SystemEventbusNamePrefix ) {
146+ return nil , errors .ErrInvalidRequest .WithMessage ("system eventbus must start with __" )
147+ }
148+ return ctrl .createEventBus (ctx , req )
149+ }
150+
151+ func (ctrl * controller ) createEventBus (ctx context.Context ,
116152 req * ctrlpb.CreateEventBusRequest ) (* metapb.EventBus , error ) {
117153 ctrl .mutex .Lock ()
118154 defer ctrl .mutex .Unlock ()
155+ if ! ctrl .isReady (ctx ) {
156+ return nil , errors .ErrServerNotRunning .WithMessage ("the cluster isn't ready to create eventbus" )
157+ }
119158 logNum := req .LogNumber
120159 if logNum == 0 {
121160 logNum = 1
@@ -451,12 +490,23 @@ func (ctrl *controller) ReportSegmentBlockIsFull(ctx context.Context,
451490 return & emptypb.Empty {}, nil
452491}
453492
454- func (ctrl * controller ) Ping (_ context.Context , _ * emptypb.Empty ) (* ctrlpb.PingResponse , error ) {
493+ func (ctrl * controller ) Ping (ctx context.Context , _ * emptypb.Empty ) (* ctrlpb.PingResponse , error ) {
455494 return & ctrlpb.PingResponse {
456- LeaderAddr : ctrl .member .GetLeaderAddr (),
495+ LeaderAddr : ctrl .member .GetLeaderAddr (),
496+ IsEventbusReady : ctrl .isReady (ctx ),
457497 }, nil
458498}
459499
500+ func (ctrl * controller ) isReady (ctx context.Context ) bool {
501+ if ctrl .member == nil {
502+ return false
503+ }
504+ if ! ctrl .member .IsLeader () && ! ctrl .member .IsReady () || ctrl .member .GetLeaderAddr () == "" {
505+ return false
506+ }
507+ return ctrl .ssMgr .CanCreateEventbus (ctx , int (ctrl .cfg .Replicas ))
508+ }
509+
460510func (ctrl * controller ) ReportSegmentLeader (ctx context.Context ,
461511 req * ctrlpb.ReportSegmentLeaderRequest ) (* emptypb.Empty , error ) {
462512 err := ctrl .eventLogMgr .UpdateSegmentReplicas (ctx , vanus .NewIDFromUint64 (req .LeaderId ), req .Term )
0 commit comments