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

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,22 +49,26 @@ 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,
 	}
 }
 
-func (c AdminConfig) Sanitize() {
+func (c *AdminConfig) Sanitize() {
 	c.Engine.Sanitize()
 	for _, d := range c.Discovery {
 		d.Sanitize()
@@ -75,7 +80,7 @@ func (c AdminConfig) Sanitize() {
 	c.Log.Sanitize()
 }
 
-func (c AdminConfig) PreProcess() error {
+func (c *AdminConfig) PreProcess() error {
 	discoveryPreProcess := func() error {
 		for _, d := range c.Discovery {
 			if err := d.PreProcess(); err != nil {
@@ -95,7 +100,7 @@ func (c AdminConfig) PreProcess() error {
 	)
 }
 
-func (c AdminConfig) PostProcess() error {
+func (c *AdminConfig) PostProcess() error {
 	discoveryPostProcess := func() error {
 		for _, d := range c.Discovery {
 			if err := d.PostProcess(); err != nil {
@@ -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/console/counter/manager.go

@@ -355,6 +355,10 @@ func (s *counterEventSubscriber) Name() string {
 	return s.name
 }
 
+func (s *counterEventSubscriber) AsyncEnabled() bool {
+	return false
+}
+
 func (s *counterEventSubscriber) ProcessEvent(event events.Event) error {
 	if s.handler == nil {
 		return nil

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 {

pkg/core/events/async.go

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package events
+
+import "sync/atomic"
+
+type subscriberState struct {
+	subscriber     Subscriber
+	async          bool
+	ch             chan Event
+	done           chan struct{}
+	closed         atomic.Bool
+	drainerStarted atomic.Bool
+}