OMC/be.ems
1
0
Fork 0
Code Issues Releases 3 Activity

feat: support rest data return channel result

Browse Source
This commit is contained in:
zhangsz
2025-04-10 15:46:09 +08:00
parent a1eaaaebd7
commit a00a7da016
7 changed files with 88 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -37,5 +37,6 @@ vendor
*.bak*
*.exe
__debug_bin*.exe
__debug_bin*
tools/evaluate/*.go

35
features/ue/service/ims_user.go
View File

@@ -5,20 +5,20 @@ import (
"strconv"
"strings"
"be.ems/src/framework/redis"
neService "be.ems/src/modules/network_element/service"
"be.ems/features/ue/model"
"be.ems/features/ue/repository"
"be.ems/src/framework/redis"
neService "be.ems/src/modules/network_element/service"
)
// 实例化服务层 IMSUserService 结构体
var NewIMSUserService = &IMSUserService{
imsUserRepository: repository.NewIMSUserRepository,
imsUserRepository: repository.NewIMSUserRepository,
}
// VoLTE用户信息 服务层处理
type IMSUserService struct {
imsUserRepository *repository.IMSUserRepository // VoLTE用户信息数据信息
imsUserRepository *repository.IMSUserRepository // VoLTE用户信息数据信息
}
// dataByRedis UDM签约用户 db:0 中 volte:*
@@ -50,12 +50,12 @@ func (r *IMSUserService) dataByRedis(imsi, neId string) []model.IMSUser {
for k, m := range mkv {
var imsi, msisdn string
KeyParts := strings.Split(k, ":")
switch len(KeyParts) {
switch len(KeyParts) {
case 0, 1:
// 处理单个部分的情况
continue
case 2:
// 处理两个部分的情况
// 处理两个部分的情况
imsi = KeyParts[1]
msisdn = "-"
case 3:
@@ -71,14 +71,14 @@ func (r *IMSUserService) dataByRedis(imsi, neId string) []model.IMSUser {
var vni string = "-"
impiParts := strings.Split(m["impi"], "@")
if len(impiParts) > 1 {
vni = impiParts[1] // 输出: ims.mnc001.mcc110.3gppnetwork.org
}
vni = impiParts[1] // 输出: ims.mnc001.mcc110.3gppnetwork.org
}
a := model.IMSUser{
NeId: neId,
IMSI: imsi, // volte:360000100000130:8612300000130
MSISDN: msisdn, // 8612300000130
VoLTE: m["tag"], // volte = tag
VNI: vni, // ims.mnc001.mcc110.3gppnetwork.org
IMSI: imsi, // volte:360000100000130:8612300000130
MSISDN: msisdn, // 8612300000130
VoLTE: m["tag"], // volte = tag
VNI: vni, // ims.mnc001.mcc110.3gppnetwork.org
}
arr = append(arr, a)
}
@@ -209,3 +209,14 @@ func (r *IMSUserService) ParseCommandParams(item model.IMSUser) string {
return strings.Join(conditions, ",")
}
// ResetDataWithResult 重置鉴权用户数据，清空数据库重新同步Redis数据
// 通过 channel 返回 ClearAndInsert 的执行结果
func (r *IMSUserService) ResetDataWithResult(neId string) chan int64 {
arr := r.dataByRedis("*", neId)
resultCh := make(chan int64, 1)
go func() {
resultCh <- r.imsUserRepository.ClearAndInsert(neId, arr)
}()
return resultCh
}

29
features/ue/service/voip_auth.go
View File

@@ -5,20 +5,20 @@ import (
"strconv"
"strings"
"be.ems/src/framework/redis"
neService "be.ems/src/modules/network_element/service"
"be.ems/features/ue/model"
"be.ems/features/ue/repository"
"be.ems/src/framework/redis"
neService "be.ems/src/modules/network_element/service"
)
// 实例化服务层 VoIPAuthService 结构体
var NewVoIPAuthService = &VoIPAuthService{
voipAuthRepository: repository.NewVoIPAuthRepository,
voipAuthRepository: repository.NewVoIPAuthRepository,
}
// VoLTE用户信息 服务层处理
type VoIPAuthService struct {
voipAuthRepository *repository.VoIPAuthRepository // VoLTE用户信息数据信息
voipAuthRepository *repository.VoIPAuthRepository // VoLTE用户信息数据信息
}
// dataByRedis UDM签约用户 db:0 中 volte:*
@@ -51,13 +51,13 @@ func (r *VoIPAuthService) dataByRedis(userName, neId string) []model.VoIPAuth {
var userName string
KeyParts := strings.Split(k, ":")
if len(KeyParts) > 1 {
userName = KeyParts[1]
}
userName = KeyParts[1]
}
a := model.VoIPAuth{
NeId: neId,
UserName: userName, // userName
Password: m["password"], //
NeId: neId,
UserName: userName, // userName
Password: m["password"], //
}
arr = append(arr, a)
}
@@ -168,3 +168,14 @@ func (r *VoIPAuthService) ParseCommandParams(item model.VoIPAuth) string {
return strings.Join(conditions, ",")
}
// ResetDataWithResult 重置鉴权用户数据，清空数据库重新同步Redis数据
// 通过 channel 返回 ClearAndInsert 的执行结果
func (r *VoIPAuthService) ResetDataWithResult(neId string) chan int64 {
arr := r.dataByRedis("*", neId)
resultCh := make(chan int64, 1)
go func() {
resultCh <- r.voipAuthRepository.ClearAndInsert(neId, arr)
}()
return resultCh
}

18
restagent/etc/restconf.yaml
View File

@@ -20,7 +20,7 @@ pprof:
rest:
- ipv4: 0.0.0.0
ipv6:
port: 33040
port: 33030
- ipv4: 0.0.0.0
ipv6:
port: 33443
@@ -31,12 +31,12 @@ rest:
keyFile: ./etc/certs/omc-server.key
webServer:
enabled: false
rootDir: d:/omc.git/fe.ems.vue3/dist # front-end build dist directory
enabled: true
rootDir: /home/simon/omc.git/fe.ems.vue3/dist # front-end build dist directory
listen:
- addr: :80
- addr: :8080
schema: http
- addr: :443
- addr: :8443
schema: https
clientAuthType: 0
caFile: ./etc/certs/omc-ca.crt
@@ -47,8 +47,8 @@ database:
type: mysql
user: root
password: 1000omc@kp!
host: 192.168.9.58
port: 13306
host: 192.168.2.211
port: 33066
name: "tenants_db"
connParam: charset=utf8mb4&collation=utf8mb4_general_ci&parseTime=True&interpolateParams=True
backup: ./database
@@ -58,8 +58,8 @@ redis:
dataSource:
# OMC system db
default:
port: 16379 # Redis port
host: "192.168.9.58" # Redis host
port: 6379 # Redis port
host: "192.168.2.211" # Redis host
password: "helloearth"
db: 10 # Redis db_num
# used to specify the default data source for multiple data resourece

57
src/modules/crontask/processor/exportUEData/exportUEData.go
View File

@@ -103,24 +103,18 @@ func (s *BarProcessor) exportUEData(param BarParams) (map[string]any, error) {
var affectedArr []int64
for _, neID := range neIDs {
// 1. 加载最新数据, 如果数据服务存在，则重新加载数据
result := ResetDataWithResult(param.ServiceName, neID)
log.Trace("reset data count=", result.Count)
// 后续可以等待异步操作返回结果
if err := <-result.Err; err != nil {
log.Error("failed to reset data: ", err)
dataService, err := GetService(param.ServiceName)
if err != nil {
log.Warn("failed to get data service:", err)
} else if dataService != nil {
// 重新加载数据
resultChan := dataService.ResetDataWithResult(neID)
affected := <-resultChan // Receive value from the channel
if affected == 0 {
return nil, fmt.Errorf("reload data failed, affected=0")
}
log.Trace("load data affected=", affected)
}
// dataService, err := GetService(param.ServiceName)
// if err != nil {
// log.Warn("failed to get data service:", err)
// } else if dataService != nil {
// // 重新加载数据
// data := dataService.ResetData(neID)
// if data == 0 {
// log.Error("failed to load data:", err)
// return nil, err
// }
// log.Trace("load data:", data)
// }
// 2. 构造查询语句
var query string
@@ -231,35 +225,10 @@ func (s *BarProcessor) exportData(query, filePath string, fileType string) (int6
return affected, nil
}
type ResetDataResult struct {
Count int64 // 同步返回的数据长度
Err <-chan error // 异步返回 ClearAndInsert 的执行结果
}
// ResetDataWithResult 重置鉴权用户数据，清空数据库重新同步Redis数据
// 立即返回数据量，同时通过 channel 返回 ClearAndInsert 的执行结果
func ResetDataWithResult(serivceName, neID string) ResetDataResult {
dataService, err := GetService(serivceName)
if err != nil {
log.Warn("failed to get data service:", err)
}
authArr := dataService.dataByRedis("*", neID)
resultCh := make(chan error, 1)
go func() {
resultCh <- dataService.ClearAndInsert(neID, authArr)
}()
return ResetDataResult{
Count: int64(len(authArr)),
Err: resultCh,
}
}
// ResettableService 接口定义
type ResettableService interface {
dataByRedis(key, neID string) []map[string]string // 从Redis获取数据
ClearAndInsert(neID string, authArr []map[string]string) error // 清空数据库并插入数据
//ResetData(neID string) int64
ResetData(neID string) int64
ResetDataWithResult(neID string) chan int64
}
// 服务注册表

11
src/modules/network_data/service/udm_auth.go
View File

@@ -200,3 +200,14 @@ func (r *UDMAuthUser) ParseCommandParams(item model.UDMAuthUser) string {
}
return strings.Join(conditions, ",")
}
// ResetDataWithResult 重置鉴权用户数据，清空数据库重新同步Redis数据
// 通过 channel 返回 ClearAndInsert 的执行结果
func (r *UDMAuthUser) ResetDataWithResult(neId string) chan int64 {
arr := r.dataByRedis("*", neId)
resultCh := make(chan int64, 1)
go func() {
resultCh <- r.udmAuthRepository.ClearAndInsert(neId, arr)
}()
return resultCh
}

11
src/modules/network_data/service/udm_sub.go
View File

@@ -362,3 +362,14 @@ func (r *UDMSubUser) ParseCommandParams(item model.UDMSubUser) string {
conditions = append(conditions, fmt.Sprintf("cag=%s", item.Cag))
return strings.Join(conditions, ",")
}
// ResetDataWithResult 重置鉴权用户数据，清空数据库重新同步Redis数据
// 通过 channel 返回 ClearAndInsert 的执行结果
func (r *UDMSubUser) ResetDataWithResult(neId string) chan int64 {
arr := r.dataByRedis("*", neId)
resultCh := make(chan int64, 1)
go func() {
resultCh <- r.udmSubRepository.ClearAndInsert(neId, arr)
}()
return resultCh
}