baize/framework/core/infrastructure/infrastructure.go

package infrastructure

import (
	"git.sxidc.com/go-framework/baize/framework/core/infrastructure/cache"
	"git.sxidc.com/go-framework/baize/framework/core/infrastructure/cache/local"
	redisCache "git.sxidc.com/go-framework/baize/framework/core/infrastructure/cache/redis"
	"git.sxidc.com/go-framework/baize/framework/core/infrastructure/database"
	"git.sxidc.com/go-framework/baize/framework/core/infrastructure/database/data_service"
	"git.sxidc.com/go-framework/baize/framework/core/infrastructure/database/operations"
	messageQueueCommon "git.sxidc.com/go-framework/baize/framework/core/infrastructure/message_queue/common"
	mqttMessageQueue "git.sxidc.com/go-framework/baize/framework/core/infrastructure/message_queue/mqtt"
	redisMessageQueue "git.sxidc.com/go-framework/baize/framework/core/infrastructure/message_queue/redis"
	"git.sxidc.com/go-tools/utils/strutils"
)

type Config struct {
	DatabaseConfig
	CacheConfig
	MessageQueueConfig
}

type DatabaseConfig struct {
	Operations  *operations.Config   `json:"operations" yaml:"operations"`
	DataService *data_service.Config `json:"data_service" yaml:"data_service"`
}

type CacheConfig struct {
	Namespace string       `json:"namespace" yaml:"namespace"`
	Redis     *RedisConfig `json:"redis" yaml:"redis"`
}

type MessageQueueConfig struct {
	Redis *RedisConfig `json:"redis" yaml:"redis"`
	Mqtt  *MqttConfig  `json:"mqtt" yaml:"mqtt"`
}

type RedisConfig struct {
	Address            string `json:"address" yaml:"address"`
	UserName           string `json:"user_name" yaml:"user_name"`
	Password           string `json:"password" yaml:"password"`
	DB                 int    `json:"db" yaml:"db"`
	MaxLen             int64  `json:"max_len" yaml:"max_len"`
	ConsumerNum        int    `json:"consumer_num" yaml:"consumer_num"`
	VisibilityAgainSec int    `json:"visibility_again_sec" yaml:"visibility_again_sec"`
	InFlightCount      int    `json:"in_flight_count" yaml:"in_flight_count"`
}

type MqttConfig struct {
	Address  string `json:"address" yaml:"address"`
	UserName string `json:"user_name" yaml:"user_name"`
	Password string `json:"password" yaml:"password"`
}

// Infrastructure 基础设施结构
type Infrastructure struct {
	dbExecutor        database.Executor
	localCache        cache.Cache
	redisCache        cache.Cache
	redisMessageQueue messageQueueCommon.MessageQueue
	mqttMessageQueue  messageQueueCommon.MessageQueue
}

func NewInfrastructure(config Config) *Infrastructure {
	i := new(Infrastructure)

	// 数据库执行器多选一
	if config.DatabaseConfig.Operations != nil {
		op, err := operations.NewOperations(config.DatabaseConfig.Operations)
		if err != nil {
			panic("创建数据库操作失败: " + err.Error())
		}

		i.dbExecutor = op
	} else if config.DatabaseConfig.DataService != nil {
		executor, err := data_service.NewExecutor(config.DatabaseConfig.DataService)
		if err != nil {
			panic("创建数据服务执行器失败: " + err.Error())
		}

		i.dbExecutor = executor
	}

	// 初始化缓存
	if strutils.IsStringNotEmpty(config.CacheConfig.Namespace) {
		namespace := config.CacheConfig.Namespace

		i.localCache = local.New(namespace)

		if config.CacheConfig.Redis != nil {
			redisConf := config.CacheConfig.Redis

			newRedisCache, err := redisCache.New(redisConf.Address, redisConf.UserName, redisConf.Password, redisConf.DB, namespace)
			if err != nil {
				panic("初始化Redis缓存失败: " + err.Error())
			}

			i.redisCache = newRedisCache
		}
	}

	// 初始化Redis消息队列
	if config.MessageQueueConfig.Redis != nil {
		redisConf := config.MessageQueueConfig.Redis

		newRedisMessageQueue := redisMessageQueue.New(redisConf.Address, redisConf.UserName, redisConf.Password, redisConf.DB,
			redisMessageQueue.WithMaxLen(redisConf.MaxLen),
			redisMessageQueue.WithConsumerNum(redisConf.ConsumerNum),
			redisMessageQueue.WithInFlightCount(redisConf.InFlightCount),
			redisMessageQueue.WithVisibilityAgainSec(redisConf.VisibilityAgainSec))

		i.redisMessageQueue = newRedisMessageQueue
	}

	// 初始化Mqtt消息队列
	if config.MessageQueueConfig.Mqtt != nil {
		mqttConf := config.MessageQueueConfig.Mqtt

		newMqttMessageQueue, err := mqttMessageQueue.New(mqttConf.Address, mqttConf.UserName, mqttConf.Password)
		if err != nil {
			panic("初始化MQTT消息队列失败: " + err.Error())
		}

		i.mqttMessageQueue = newMqttMessageQueue
	}

	return i
}

func DestroyInfrastructure(i *Infrastructure) {
	if i == nil {
		return
	}

	if i.redisMessageQueue != nil {
		redisMessageQueue.Destroy(i.redisMessageQueue.(*redisMessageQueue.MessageQueue))
	}

	if i.redisCache != nil {
		err := redisCache.Destroy(i.redisCache.(*redisCache.Cache))
		if err != nil {
			panic("销毁Redis缓存失败: " + err.Error())
		}
	}

	if i.localCache != nil {
		local.Destroy(i.localCache.(*local.Cache))
	}

	if i.dbExecutor != nil {
		switch dbExecutor := i.dbExecutor.(type) {
		case *operations.Operations:
			err := operations.DestroyOperation(dbExecutor)
			if err != nil {
				panic("销毁数据库操作失败: " + err.Error())
			}
		case *data_service.Executor:
			err := data_service.DestroyExecutor(dbExecutor)
			if err != nil {
				panic("销毁数据库操作失败: " + err.Error())
			}
		default:
			panic("不支持的数据库执行器类型")
		}

	}

	return
}

// DBExecutor 获取数据库基础设施
// 参数: 无
// 返回值:
// - 数据库基础设施的接口
func (i Infrastructure) DBExecutor() database.Executor {
	return i.dbExecutor
}

// LocalCache 获取本地缓存基础设施
// 参数: 无
// 返回值:
// - 缓存基础设施的接口
func (i Infrastructure) LocalCache() cache.Cache {
	return i.localCache
}

// RedisCache 获取Redis缓存基础设施
// 参数: 无
// 返回值:
// - 缓存基础设施的接口
func (i Infrastructure) RedisCache() cache.Cache {
	return i.redisCache
}

// RedisMessageQueue 获取Redis消息队列基础设施
// 参数: 无
// 返回值:
// - 消息队列基础设施的接口
func (i Infrastructure) RedisMessageQueue() messageQueueCommon.MessageQueue {
	return i.redisMessageQueue
}

// MqttMessageQueue 获取Mqtt消息队列基础设施
// 参数: 无
// 返回值:
// - 消息队列基础设施的接口
func (i Infrastructure) MqttMessageQueue() messageQueueCommon.MessageQueue {
	return i.mqttMessageQueue
}