feat(eventbus): support per-subscriber async dispatch with graceful drain

Author: WyRainBow

pkg/config/app/admin.go

@@ -27,6 +27,7 @@ import (
 	"github.com/apache/dubbo-admin/pkg/config/diagnostics"
 	"github.com/apache/dubbo-admin/pkg/config/discovery"
 	"github.com/apache/dubbo-admin/pkg/config/engine"
+	"github.com/apache/dubbo-admin/pkg/config/eventbus"
 	"github.com/apache/dubbo-admin/pkg/config/log"
 	"github.com/apache/dubbo-admin/pkg/config/observability"
 	"github.com/apache/dubbo-admin/pkg/config/store"
@@ -48,18 +49,22 @@ type AdminConfig struct {
 	Discovery []*discovery.Config `json:"discovery" yaml:"discovery"`
 	// Engine configuration
 	Engine *engine.Config `json:"engine" yaml:"engine"`
+	// EventBus configuration
+	EventBus *eventbus.Config `json:"eventBus,omitempty" yaml:"eventBus,omitempty"`
 }
 
 var _ = &AdminConfig{}
 
 var DefaultAdminConfig = func() AdminConfig {
+	eventBusCfg := eventbus.Default()
 	return AdminConfig{
 		Log:           log.DefaultLogConfig(),
 		Store:         store.DefaultStoreConfig(),
 		Engine:        engine.DefaultResourceEngineConfig(),
 		Observability: observability.DefaultObservabilityConfig(),
 		Diagnostics:   diagnostics.DefaultDiagnosticsConfig(),
 		Console:       console.DefaultConsoleConfig(),
+		EventBus:      &eventBusCfg,
 	}
 }
 
@@ -115,7 +120,7 @@ func (c AdminConfig) PostProcess() error {
 	)
 }
 
-func (c AdminConfig) Validate() error {
+func (c *AdminConfig) Validate() error {
 	if c.Log == nil {
 		c.Log = log.DefaultLogConfig()
 	} else if err := c.Log.Validate(); err != nil {
@@ -160,6 +165,12 @@ func (c AdminConfig) Validate() error {
 	} else if err := c.Engine.Validate(); err != nil {
 		return bizerror.Wrap(err, bizerror.ConfigError, "engine config validation failed")
 	}
+	if c.EventBus == nil {
+		cfg := eventbus.Default()
+		c.EventBus = &cfg
+	} else if err := c.EventBus.Validate(); err != nil {
+		return bizerror.Wrap(err, bizerror.ConfigError, "event bus config validation failed")
+	}
 	return nil
 }
 

pkg/config/eventbus/config.go

@@ -29,6 +29,6 @@ func (c Config) Validate() error {
 
 func Default() Config {
 	return Config{
-		BufferSize: 100,
+		BufferSize: 1024,
 	}
 }

pkg/core/discovery/subscriber/instance.go

@@ -55,6 +55,10 @@ func (s *InstanceEventSubscriber) Name() string {
 	return "Discovery-" + s.ResourceKind().ToString()
 }
 
+func (s *InstanceEventSubscriber) AsyncEnabled() bool {
+	return true
+}
+
 func (s *InstanceEventSubscriber) ProcessEvent(event events.Event) error {
 	newObj, ok := event.NewObj().(*meshresource.InstanceResource)
 	if !ok && event.NewObj() != nil {

pkg/core/discovery/subscriber/nacos_service.go

@@ -57,6 +57,10 @@ func (n *NacosServiceEventSubscriber) Name() string {
 	return "Nacos2Discovery-" + n.ResourceKind().ToString()
 }
 
+func (n *NacosServiceEventSubscriber) AsyncEnabled() bool {
+	return true
+}
+
 func (n *NacosServiceEventSubscriber) ProcessEvent(event events.Event) error {
 	newObj, ok := event.NewObj().(*meshresource.NacosServiceResource)
 	if !ok && event.NewObj() != nil {

pkg/core/discovery/subscriber/rpc_instance.go

@@ -63,6 +63,10 @@ func (s *RPCInstanceEventSubscriber) ResourceKind() coremodel.ResourceKind {
 	return meshresource.RPCInstanceKind
 }
 
+func (s *RPCInstanceEventSubscriber) AsyncEnabled() bool {
+	return true
+}
+
 func (s *RPCInstanceEventSubscriber) ProcessEvent(event events.Event) error {
 	newObj, ok := event.NewObj().(*meshresource.RPCInstanceResource)
 	if !ok && event.NewObj() != nil {

pkg/core/discovery/subscriber/service_consumer_metadata.go

@@ -53,6 +53,10 @@ func (s *ServiceConsumerMetadataEventSubscriber) Name() string {
 	return "Discovery-" + s.ResourceKind().ToString()
 }
 
+func (s *ServiceConsumerMetadataEventSubscriber) AsyncEnabled() bool {
+	return true
+}
+
 func (s *ServiceConsumerMetadataEventSubscriber) ProcessEvent(event events.Event) error {
 	newObj, ok := event.NewObj().(*meshresource.ServiceConsumerMetadataResource)
 	if !ok && event.NewObj() != nil {

pkg/core/discovery/subscriber/service_provider_metadata.go

@@ -53,6 +53,10 @@ func (s *ServiceProviderMetadataEventSubscriber) Name() string {
 	return "Discovery-" + s.ResourceKind().ToString()
 }
 
+func (s *ServiceProviderMetadataEventSubscriber) AsyncEnabled() bool {
+	return true
+}
+
 func (s *ServiceProviderMetadataEventSubscriber) ProcessEvent(event events.Event) error {
 	newObj, ok := event.NewObj().(*meshresource.ServiceProviderMetadataResource)
 	if !ok && event.NewObj() != nil {

pkg/core/discovery/subscriber/zk_config.go

@@ -53,6 +53,10 @@ func (z *ZKConfigEventSubscriber) Name() string {
 	return "Discovery-" + z.ResourceKind().ToString()
 }
 
+func (z *ZKConfigEventSubscriber) AsyncEnabled() bool {
+	return true
+}
+
 func (z *ZKConfigEventSubscriber) ProcessEvent(event events.Event) error {
 	newObj, ok := event.NewObj().(*meshresource.ZKConfigResource)
 	if !ok && event.NewObj() != nil {

pkg/core/discovery/subscriber/zk_metadata.go

@@ -53,6 +53,10 @@ func (z *ZKMetadataEventSubscriber) Name() string {
 	return "Discovery-" + z.ResourceKind().ToString()
 }
 
+func (z *ZKMetadataEventSubscriber) AsyncEnabled() bool {
+	return true
+}
+
 func (z *ZKMetadataEventSubscriber) ProcessEvent(event events.Event) error {
 	newObj, ok := event.NewObj().(*meshresource.ZKMetadataResource)
 	if !ok && event.NewObj() != nil {

pkg/core/engine/subscriber/runtime_instance.go

@@ -56,6 +56,10 @@ func (s *RuntimeInstanceEventSubscriber) Name() string {
 	return "Engine-" + s.ResourceKind().ToString()
 }
 
+func (s *RuntimeInstanceEventSubscriber) AsyncEnabled() bool {
+	return true
+}
+
 func (s *RuntimeInstanceEventSubscriber) ProcessEvent(event events.Event) error {
 	newObj, ok := event.NewObj().(*meshresource.RuntimeInstanceResource)
 	if !ok && event.NewObj() != nil {