MiFaZhan
299 lines
9.3 KiB
Go
299 lines
9.3 KiB
Go
package channelconfig
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"strings"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model/entity"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/service"
|
|
|
|
"sync"
|
|
"time"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/dao"
|
|
|
|
"github.com/gogf/gf/v2/crypto/gmd5"
|
|
"github.com/gogf/gf/v2/encoding/gjson"
|
|
"github.com/gogf/gf/v2/errors/gerror"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
)
|
|
|
|
// channelConfigRow extends the generated entity without losing SQL NULL values.
|
|
type channelConfigRow struct {
|
|
entity.TSimChannels
|
|
IsNormal *int `json:"isNormal" orm:"is_normal"`
|
|
}
|
|
|
|
type sChannelConfig struct {
|
|
sync.RWMutex
|
|
channelCache map[string]*model.SimChannelConfigItemCache
|
|
loadVersion uint8
|
|
chgLogMaxId int64
|
|
running bool
|
|
}
|
|
|
|
func NewChannelConfig() *sChannelConfig {
|
|
return &sChannelConfig{
|
|
channelCache: make(map[string]*model.SimChannelConfigItemCache),
|
|
}
|
|
}
|
|
|
|
func init() {
|
|
service.RegisterChannelConfig(NewChannelConfig())
|
|
}
|
|
|
|
func (s *sChannelConfig) Start(ctx context.Context) error {
|
|
err := s.LoadAllChannelConfig(ctx)
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "ChannelConfig Start err %+v", err)
|
|
return err
|
|
}
|
|
s.running = true
|
|
err = s.Loop(ctx)
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "ChannelConfig Start err %+v", err)
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *sChannelConfig) Stop(ctx context.Context) error {
|
|
s.running = false
|
|
return nil
|
|
}
|
|
|
|
func (s *sChannelConfig) Loop(ctx context.Context) error {
|
|
g.Log().Infof(ctx, "ChannelConfig loop begin.")
|
|
lastLoadAllChannelConfigTime := int64(0)
|
|
go func() {
|
|
for s.running {
|
|
now := time.Now().Unix()
|
|
loadAllInterval := g.Cfg().MustGet(ctx, "channel.load_all_interval", 2*60*60).Int64()
|
|
if now-lastLoadAllChannelConfigTime >= loadAllInterval {
|
|
lastLoadAllChannelConfigTime = now
|
|
s.LoadAllChannelConfig(ctx)
|
|
}
|
|
time.Sleep(1 * time.Second)
|
|
}
|
|
}()
|
|
g.Log().Infof(ctx, "ChannelConfig loop background goroutine started.")
|
|
return nil
|
|
}
|
|
|
|
func (s *sChannelConfig) LoadAllChannelConfig(ctx context.Context) error {
|
|
s.loadVersion++
|
|
const warnRowThreshold = 1000
|
|
|
|
dbSimChannels := make([]*channelConfigRow, 0)
|
|
err := dao.TSimChannels.Ctx(ctx).
|
|
OrderAsc(dao.TSimChannels.Columns().Id).
|
|
Scan(&dbSimChannels)
|
|
if err != nil && err != sql.ErrNoRows {
|
|
return err
|
|
}
|
|
if len(dbSimChannels) >= warnRowThreshold {
|
|
g.Log().Warningf(ctx, "LoadAllChannelConfig TSimChannels row count %d >= %d, confirm small-table assumption or add pagination",
|
|
len(dbSimChannels), warnRowThreshold)
|
|
}
|
|
|
|
dbSimChannelTemplates := make([]*entity.TSimChannelTemplate, 0)
|
|
err = dao.TSimChannelTemplate.Ctx(ctx).
|
|
OrderAsc(dao.TSimChannelTemplate.Columns().Id).
|
|
Scan(&dbSimChannelTemplates)
|
|
if err != nil && err != sql.ErrNoRows {
|
|
return err
|
|
}
|
|
if len(dbSimChannelTemplates) >= warnRowThreshold {
|
|
g.Log().Warningf(ctx, "LoadAllChannelConfig TSimChannelTemplate row count %d >= %d, confirm small-table assumption or add pagination",
|
|
len(dbSimChannelTemplates), warnRowThreshold)
|
|
}
|
|
|
|
templateByKey := make(map[string]*entity.TSimChannelTemplate, len(dbSimChannelTemplates))
|
|
for _, tpl := range dbSimChannelTemplates {
|
|
if tpl != nil {
|
|
templateByKey[tpl.TempKey] = tpl
|
|
}
|
|
}
|
|
|
|
s.Lock()
|
|
for _, dbSimChannel := range dbSimChannels {
|
|
if dbSimChannel == nil {
|
|
continue
|
|
}
|
|
cache := &model.SimChannelConfigItemCache{
|
|
Channel: &dbSimChannel.TSimChannels,
|
|
IsNormal: dbSimChannel.IsNormal,
|
|
Version: s.getChanVersion(ctx, dbSimChannel),
|
|
}
|
|
oldChannelCache, _ := s.channelCache[dbSimChannel.ChanKey]
|
|
if cache.CanCreateInstance() {
|
|
template, ok := templateByKey[dbSimChannel.TempKey]
|
|
if !ok {
|
|
g.Log().Warningf(ctx, "LoadAllChannelConfig no template for tempKey=%s chanKey=%s", dbSimChannel.TempKey, dbSimChannel.ChanKey)
|
|
continue
|
|
}
|
|
cache.Template = template
|
|
cache.Version += s.getTemplateVersion(ctx, template)
|
|
}
|
|
|
|
if oldChannelCache != nil && oldChannelCache.Version == cache.Version {
|
|
continue
|
|
}
|
|
s.channelCache[dbSimChannel.ChanKey] = cache
|
|
}
|
|
s.Unlock()
|
|
return nil
|
|
}
|
|
|
|
// RefreshChannel reloads one channel and its template from the database.
|
|
func (s *sChannelConfig) RefreshChannel(ctx context.Context, chanKey string) error {
|
|
if chanKey == "" {
|
|
return gerror.New("chanKey is required")
|
|
}
|
|
channel := new(channelConfigRow)
|
|
err := dao.TSimChannels.Ctx(ctx).Where(dao.TSimChannels.Columns().ChanKey, chanKey).Scan(channel)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if channel.ChanKey == "" {
|
|
return gerror.Newf("channel not found: %s", chanKey)
|
|
}
|
|
cache := &model.SimChannelConfigItemCache{
|
|
Channel: &channel.TSimChannels,
|
|
IsNormal: channel.IsNormal,
|
|
Version: s.getChanVersion(ctx, channel),
|
|
}
|
|
// Ineligible channels refresh successfully even while their template is incomplete.
|
|
if cache.CanCreateInstance() {
|
|
template := new(entity.TSimChannelTemplate)
|
|
if err := dao.TSimChannelTemplate.Ctx(ctx).Where(dao.TSimChannelTemplate.Columns().TempKey, channel.TempKey).Scan(template); err != nil {
|
|
return err
|
|
}
|
|
if template.TempKey == "" {
|
|
return gerror.Newf("channel template not found: %s", channel.TempKey)
|
|
}
|
|
cache.Template = template
|
|
cache.Version += s.getTemplateVersion(ctx, template)
|
|
}
|
|
s.Lock()
|
|
s.channelCache[chanKey] = cache
|
|
s.Unlock()
|
|
return nil
|
|
}
|
|
|
|
// RemoveChannel removes a channel configuration cache. Missing keys are a
|
|
// successful no-op so rename retries are idempotent.
|
|
func (s *sChannelConfig) RemoveChannel(ctx context.Context, chanKey string) error {
|
|
chanKey = strings.TrimSpace(chanKey)
|
|
if chanKey == "" {
|
|
return gerror.New("chanKey is required")
|
|
}
|
|
s.Lock()
|
|
delete(s.channelCache, chanKey)
|
|
s.Unlock()
|
|
g.Log().Infof(ctx, "channel config removed, chanKey: %s", chanKey)
|
|
return nil
|
|
}
|
|
|
|
// LoadChannelForTest reads an isolated configuration snapshot. It never
|
|
// mutates channelCache, including when the channel is not runtime-eligible.
|
|
func (s *sChannelConfig) LoadChannelForTest(ctx context.Context, channelID int64, tenantID string) (*model.SimChannelConfigItemCache, error) {
|
|
tenantID = strings.TrimSpace(tenantID)
|
|
if channelID <= 0 || tenantID == "" {
|
|
return nil, model.ErrChannelNotAccessible
|
|
}
|
|
channel := new(channelConfigRow)
|
|
if err := dao.TSimChannels.Ctx(ctx).
|
|
Where(dao.TSimChannels.Columns().Id, channelID).
|
|
Where(dao.TSimChannels.Columns().TenantId, tenantID).
|
|
Limit(1).
|
|
Scan(channel); err != nil {
|
|
return nil, err
|
|
}
|
|
if channel.Id == 0 || strings.TrimSpace(channel.ChanKey) == "" {
|
|
return nil, model.ErrChannelNotAccessible
|
|
}
|
|
template := new(entity.TSimChannelTemplate)
|
|
if err := dao.TSimChannelTemplate.Ctx(ctx).
|
|
Where(dao.TSimChannelTemplate.Columns().TempKey, channel.TempKey).
|
|
Where(dao.TSimChannelTemplate.Columns().TenantId, tenantID).
|
|
Limit(1).
|
|
Scan(template); err != nil {
|
|
return nil, err
|
|
}
|
|
if strings.TrimSpace(template.TempKey) == "" {
|
|
return nil, gerror.Newf("channel template not found: %s", channel.TempKey)
|
|
}
|
|
return &model.SimChannelConfigItemCache{
|
|
Channel: &channel.TSimChannels,
|
|
Template: template,
|
|
IsNormal: channel.IsNormal,
|
|
Version: s.getChanVersion(ctx, channel) + s.getTemplateVersion(ctx, template),
|
|
}, nil
|
|
}
|
|
|
|
// get alarm cache by id
|
|
func (s *sChannelConfig) GetChannel(chanKey string) (*model.SimChannelConfigItemCache, bool) {
|
|
s.RLock()
|
|
defer s.RUnlock()
|
|
channelCache, ok := s.channelCache[chanKey]
|
|
return channelCache, ok
|
|
}
|
|
|
|
func (s *sChannelConfig) GetAllChannel(ctx context.Context) ([]*model.SimChannelConfigItemCache, error) {
|
|
s.RLock()
|
|
defer s.RUnlock()
|
|
channels := make([]*model.SimChannelConfigItemCache, 0)
|
|
for _, channelCache := range s.channelCache {
|
|
channels = append(channels, channelCache)
|
|
}
|
|
return channels, nil
|
|
}
|
|
|
|
func (s *sChannelConfig) getChanVersion(ctx context.Context, channel *channelConfigRow) string {
|
|
chanKey := channel.ChanKey
|
|
|
|
channelNew := &channelConfigRow{
|
|
IsNormal: channel.IsNormal,
|
|
TSimChannels: entity.TSimChannels{
|
|
TempKey: channel.TempKey,
|
|
ChanKey: channel.ChanKey,
|
|
ChanName: channel.ChanName,
|
|
ChanConfig: channel.ChanConfig,
|
|
ChanStatus: channel.ChanStatus,
|
|
ChanAlias: channel.ChanAlias,
|
|
RegionCode: channel.RegionCode,
|
|
AppKey: channel.AppKey,
|
|
AppSecret: channel.AppSecret,
|
|
AccessToken: channel.AccessToken,
|
|
RefreshToken: channel.RefreshToken,
|
|
},
|
|
}
|
|
chanVersion, err := gmd5.EncryptString(gjson.MustEncodeString(channelNew))
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "sChannelConfig.getChanVersion encrypt error %+v, chanKey %s", err, chanKey)
|
|
return ""
|
|
}
|
|
g.Log().Debugf(ctx, "sChannelConfig.getChanVersion chanVersion=%s chanKey=%s", chanVersion, chanKey)
|
|
return chanVersion
|
|
}
|
|
|
|
func (s *sChannelConfig) getTemplateVersion(ctx context.Context, template *entity.TSimChannelTemplate) string {
|
|
templateKey := template.TempKey
|
|
templateNew := &entity.TSimChannelTemplate{
|
|
TempKey: template.TempKey,
|
|
TempName: template.TempName,
|
|
TempModel: template.TempModel,
|
|
TempConfig: template.TempConfig,
|
|
TempStatus: template.TempStatus,
|
|
}
|
|
templateVersion, err := gmd5.EncryptString(gjson.MustEncodeString(templateNew))
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "sChannelConfig.getTemplateVersion encrypt error %+v, templateKey %s, templateNew %+v", err, templateKey, templateNew)
|
|
return ""
|
|
}
|
|
g.Log().Debugf(ctx, "sChannelConfig.getTemplateVersion templateVersion=%s templateKey=%s", templateVersion, templateKey)
|
|
return templateVersion
|
|
}
|