feat(middleware): add middleware work function

pkg/middle/define.go

@@ -1,7 +1,15 @@
 package middle
 
-import getMiddlewareMap "gitee.com/timedb/wheatCache/plugins/config"
+import (
+	"sync"
+)
 
-func Init() {
-	getMiddlewareMap.GetMiddlewareMap()
-}
+var (
+	oneMiddle        sync.Once
+	MiddleWareDriver *MiddleWare
+)
+
+const (
+	defaultConsumerCount = 5
+	defaultDriverCount   = 1000
+)

pkg/middle/middleware.go

@@ -0,0 +1,99 @@
+package middle
+
+import (
+	"context"
+
+	_ "gitee.com/timedb/wheatCache/conf"
+	"gitee.com/timedb/wheatCache/pkg/event"
+	"gitee.com/timedb/wheatCache/pkg/logx"
+	middleMsg "gitee.com/timedb/wheatCache/pkg/middle-msg"
+	"gitee.com/timedb/wheatCache/plugins"
+	"gitee.com/timedb/wheatCache/plugins/config"
+	"github.com/spf13/viper"
+)
+
+type MiddleWare struct {
+	eventDriver   event.DriverInterface
+	eventConsumer event.ConsumerInterface
+	eventProduce  event.ProduceInterface
+	plugins       map[string][]plugins.MiddleToolsInterface
+	consumerCount int
+}
+
+func NewMiddleWare() *MiddleWare {
+	oneMiddle.Do(func() {
+		_, driverCount := loadConfigAndDefault()
+
+		driver := event.NewDriver(driverCount)
+		MiddleWareDriver = &MiddleWare{
+			eventDriver:   driver,
+			eventConsumer: event.NewConsumer(driver),
+			eventProduce:  event.NewProduce(driver),
+		}
+		MiddleWareDriver.loadPlugins()
+
+	})
+	return MiddleWareDriver
+}
+
+func (m *MiddleWare) GetEventDriver() event.DriverInterface {
+	return m.eventDriver
+}
+
+func (m *MiddleWare) loadPlugins() {
+	plug := viper.GetStringMapStringSlice("plugins-control")
+
+	pluginsMap := config.GetMiddlewareMap()
+
+	pluginsContext := make(map[string][]plugins.MiddleToolsInterface)
+
+	for middleMsg, pluNames := range plug {
+		pulgSingle := make([]plugins.MiddleToolsInterface, 0)
+		for _, name := range pluNames {
+			pulgSingle = append(pulgSingle, pluginsMap[name])
+		}
+
+		pluginsContext[middleMsg] = pulgSingle
+	}
+
+	m.plugins = pluginsContext
+}
+
+func loadConfigAndDefault() (int, int) {
+	// 加载 consumerCount
+	consumerCount := viper.GetInt("middleware-driver.middleConsumerCount")
+	if consumerCount == 0 {
+		consumerCount = defaultConsumerCount
+	}
+
+	driverCount := viper.GetInt("middleware-driver.driverCount")
+	if driverCount == 0 {
+		driverCount = defaultDriverCount
+	}
+	return consumerCount, driverCount
+
+}
+
+func (m *MiddleWare) startWork() {
+
+	for i := 0; i < m.consumerCount; i++ {
+		go func() {
+			ctx := context.Background()
+			for {
+				workEvent := m.eventConsumer.Receive(ctx)
+				plugs := m.plugins[workEvent.GetEventName()]
+
+				msg, ok := workEvent.GetValue(middleMsg.MiddleMsgKey)
+				if !ok {
+					logx.With(ctx, m.eventProduce).Error("get event value err，not key:%s", middleMsg.MiddleMsgKey)
+					continue
+				}
+
+				for _, val := range plugs {
+					val.Exec(msg)
+				}
+			}
+
+		}()
+	}
+}

pkg/middle/middleware_test.go

@@ -0,0 +1,23 @@
+package middle
+
+import (
+	"testing"
+
+	"gitee.com/timedb/wheatCache/pkg/event"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_middleware_driver(t *testing.T) {
+	middleware := NewMiddleWare()
+	require.Equal(t, middleware.plugins["logcontext"][0].Name(), "logMiddle")
+}
+
+func TestWorker(t *testing.T) {
+
+	event := event.NewEvent("logcontext")
+
+	m := NewMiddleWare()
+	m.eventDriver.Put(event)
+
+	m.startWork()
+}