Implment Counter in local cache

pkg/console/context/context.go

@@ -21,12 +21,14 @@ import (
 	ctx "context"
 
 	"github.com/apache/dubbo-admin/pkg/config/app"
+	"github.com/apache/dubbo-admin/pkg/console/counter"
 	"github.com/apache/dubbo-admin/pkg/core/manager"
 	"github.com/apache/dubbo-admin/pkg/core/runtime"
 )
 
 type Context interface {
 	ResourceManager() manager.ResourceManager
+	CounterManager() counter.CounterManager
 
 	Config() app.AdminConfig
 
@@ -57,3 +59,15 @@ func (c *context) ResourceManager() manager.ResourceManager {
 	rmc, _ := c.coreRt.GetComponent(runtime.ResourceManager)
 	return rmc.(manager.ResourceManagerComponent).ResourceManager()
 }
+
+func (c *context) CounterManager() counter.CounterManager {
+	comp, err := c.coreRt.GetComponent(counter.ComponentType)
+	if err != nil {
+		return nil
+	}
+	managerComp, ok := comp.(counter.ManagerComponent)
+	if !ok {
+		return nil
+	}
+	return managerComp.CounterManager()
+}

pkg/console/counter/component.go

@@ -0,0 +1,73 @@
+/*
+ * 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 counter
+
+import (
+	"fmt"
+	"math"
+
+	"github.com/apache/dubbo-admin/pkg/core/events"
+	"github.com/apache/dubbo-admin/pkg/core/runtime"
+)
+
+const ComponentType runtime.ComponentType = "counter manager"
+
+func init() {
+	runtime.RegisterComponent(&managerComponent{})
+}
+
+type ManagerComponent interface {
+	runtime.Component
+	CounterManager() CounterManager
+}
+
+var _ ManagerComponent = &managerComponent{}
+
+type managerComponent struct {
+	manager CounterManager
+}
+
+func (c *managerComponent) Type() runtime.ComponentType {
+	return ComponentType
+}
+
+func (c *managerComponent) Order() int {
+	return math.MaxInt - 1
+}
+
+func (c *managerComponent) Init(runtime.BuilderContext) error {
+	mgr := NewCounterManager()
+	c.manager = mgr
+	return nil
+}
+
+func (c *managerComponent) Start(rt runtime.Runtime, _ <-chan struct{}) error {
+	component, err := rt.GetComponent(runtime.EventBus)
+	if err != nil {
+		return err
+	}
+	bus, ok := component.(events.EventBus)
+	if !ok {
+		return fmt.Errorf("component %s does not implement events.EventBus", runtime.EventBus)
+	}
+	return c.manager.Bind(bus)
+}
+
+func (c *managerComponent) CounterManager() CounterManager {
+	return c.manager
+}

pkg/console/counter/counter.go

@@ -0,0 +1,104 @@
+/*
+ * 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 counter
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+type Counter struct {
+	name  string
+	value atomic.Int64
+}
+
+func NewCounter(name string) *Counter {
+	return &Counter{name: name}
+}
+
+func (c *Counter) Get() int64 {
+	return c.value.Load()
+}
+
+func (c *Counter) Increment() {
+	c.value.Add(1)
+}
+
+func (c *Counter) Decrement() {
+	for {
+		current := c.value.Load()
+		if current == 0 {
+			return
+		}
+		if c.value.CompareAndSwap(current, current-1) {
+			return
+		}
+	}
+}
+
+func (c *Counter) Reset() {
+	c.value.Store(0)
+}
+
+type DistributionCounter struct {
+	name string
+	data map[string]int64
+	mu   sync.RWMutex
+}
+
+func NewDistributionCounter(name string) *DistributionCounter {
+	return &DistributionCounter{
+		name: name,
+		data: make(map[string]int64),
+	}
+}
+
+func (c *DistributionCounter) Increment(key string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.data[key]++
+}
+
+func (c *DistributionCounter) Decrement(key string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if value, ok := c.data[key]; ok {
+		value--
+		if value <= 0 {
+			delete(c.data, key)
+		} else {
+			c.data[key] = value
+		}
+	}
+}
+
+func (c *DistributionCounter) GetAll() map[string]int64 {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	result := make(map[string]int64, len(c.data))
+	for k, v := range c.data {
+		result[k] = v
+	}
+	return result
+}
+
+func (c *DistributionCounter) Reset() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.data = make(map[string]int64)
+}