MiFaZhan
400 lines
12 KiB
Go
400 lines
12 KiB
Go
package upstreamapi
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/consts"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/dao"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model/entity"
|
|
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
|
|
|
|
"github.com/gogf/gf/v2/errors/gerror"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
)
|
|
|
|
var REDIS_KEY = "sim:chan:iccid:"
|
|
|
|
type sUpstreamAPI struct {
|
|
tempMap map[string]model.SimCardTempI
|
|
chanMap map[string]*model.SimCardChannelCache
|
|
channelMux sync.Mutex
|
|
channelOps map[string]*sync.Mutex
|
|
sync.RWMutex
|
|
running bool
|
|
}
|
|
|
|
func NewUpstreamAPI() *sUpstreamAPI {
|
|
return &sUpstreamAPI{
|
|
chanMap: make(map[string]*model.SimCardChannelCache),
|
|
tempMap: make(map[string]model.SimCardTempI),
|
|
channelOps: make(map[string]*sync.Mutex),
|
|
}
|
|
}
|
|
|
|
func init() {
|
|
service.RegisterUpstreamAPI(NewUpstreamAPI())
|
|
}
|
|
|
|
func (u *sUpstreamAPI) Start(ctx context.Context) error {
|
|
u.running = true
|
|
lastReloadTime := int64(0)
|
|
go func() {
|
|
for u.running {
|
|
now := time.Now().Unix()
|
|
reloadInterval := g.Cfg().MustGet(ctx, "upstreamapi.reload_interval", 60).Int64()
|
|
if now-lastReloadTime >= reloadInterval {
|
|
lastReloadTime = now
|
|
err := u.loop(ctx)
|
|
if err != nil {
|
|
g.Log().Errorf(ctx, "sUpstreamAPI loop err %+v", err)
|
|
}
|
|
}
|
|
time.Sleep(1 * time.Second)
|
|
}
|
|
}()
|
|
g.Log().Infof(ctx, "sUpstreamAPI Start success")
|
|
return nil
|
|
}
|
|
|
|
func (u *sUpstreamAPI) Stop(ctx context.Context) error {
|
|
u.running = false
|
|
return nil
|
|
}
|
|
|
|
func (u *sUpstreamAPI) RefreshChannel(ctx context.Context, chanKey string) error {
|
|
unlock := u.lockChannel(chanKey)
|
|
defer unlock()
|
|
return u.refreshChannelLocked(ctx, chanKey)
|
|
}
|
|
|
|
func (u *sUpstreamAPI) refreshChannelLocked(ctx context.Context, chanKey string) error {
|
|
if err := service.ChannelConfig().RefreshChannel(ctx, chanKey); err != nil {
|
|
return err
|
|
}
|
|
channelConf, ok := service.ChannelConfig().GetChannel(chanKey)
|
|
if !ok || channelConf == nil {
|
|
return gerror.Newf("channel config not found: %s", chanKey)
|
|
}
|
|
if !channelConf.CanCreateInstance() {
|
|
u.UnregisterChan(chanKey)
|
|
g.Log().Infof(ctx, "channel instance removed because is_normal is not 0, chanKey: %s", chanKey)
|
|
return nil
|
|
}
|
|
temp, ok := u.GetTemp(channelConf.Template.TempKey)
|
|
if !ok || temp == nil {
|
|
return gerror.Newf("channel template not registered: %s", channelConf.Template.TempKey)
|
|
}
|
|
instance := temp.NewTemp(ctx, channelConf)
|
|
if instance == nil {
|
|
return gerror.Newf("create channel instance failed: %s", chanKey)
|
|
}
|
|
u.RegisterChan(chanKey, channelConf.Version, &model.SimCardChannelCache{
|
|
Instance: instance,
|
|
Version: channelConf.Version,
|
|
})
|
|
g.Log().Infof(ctx, "channel refreshed, chanKey: %s, version: %s", chanKey, channelConf.Version)
|
|
return nil
|
|
}
|
|
|
|
// RemoveChannel drops the cached configuration and runtime instance for a
|
|
// channel. Deleting an already absent key is a successful no-op.
|
|
func (u *sUpstreamAPI) RemoveChannel(ctx context.Context, chanKey string) error {
|
|
chanKey = strings.TrimSpace(chanKey)
|
|
if chanKey == "" {
|
|
return gerror.New("chanKey is required")
|
|
}
|
|
unlock := u.lockChannel(chanKey)
|
|
defer unlock()
|
|
if err := service.ChannelConfig().RemoveChannel(ctx, chanKey); err != nil {
|
|
return err
|
|
}
|
|
u.UnregisterChan(chanKey)
|
|
g.Log().Infof(ctx, "channel removed, chanKey: %s", chanKey)
|
|
return nil
|
|
}
|
|
|
|
func (u *sUpstreamAPI) loop(ctx context.Context) error {
|
|
channelConfigs, err := service.ChannelConfig().GetAllChannel(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, snapshot := range channelConfigs {
|
|
if snapshot == nil || snapshot.Channel == nil {
|
|
continue
|
|
}
|
|
chanKey := snapshot.Channel.ChanKey
|
|
func() {
|
|
unlock := u.lockChannel(chanKey)
|
|
defer unlock()
|
|
if err := service.ChannelConfig().RefreshChannel(ctx, chanKey); err != nil {
|
|
g.Log().Errorf(ctx, "reload channel config failed, chanKey: %s, err: %+v", chanKey, err)
|
|
return
|
|
}
|
|
channelConf, ok := service.ChannelConfig().GetChannel(chanKey)
|
|
if !ok || channelConf == nil {
|
|
return
|
|
}
|
|
if !channelConf.CanCreateInstance() {
|
|
u.UnregisterChan(chanKey)
|
|
return
|
|
}
|
|
tempKey := channelConf.Template.TempKey
|
|
channelCache, ok := u.GetChan(chanKey)
|
|
if ok && channelCache != nil && channelCache.Version == channelConf.Version {
|
|
return
|
|
}
|
|
|
|
isNew := !ok || channelCache == nil
|
|
g.Log().Infof(ctx, "Create channel instance begin, chanKey: %s, isNew: %v", chanKey, isNew)
|
|
temp, ok := u.GetTemp(tempKey)
|
|
if !ok || temp == nil {
|
|
g.Log().Errorf(ctx, "Create channel instance failed, Temp not found, tempKey: %s", tempKey)
|
|
return
|
|
}
|
|
instance := temp.NewTemp(ctx, channelConf)
|
|
if instance == nil {
|
|
g.Log().Errorf(ctx, "Create channel instance failed, Temp new instance failed, tempKey: %s", tempKey)
|
|
return
|
|
}
|
|
channelCache = &model.SimCardChannelCache{Instance: instance, Version: channelConf.Version}
|
|
u.RegisterChan(chanKey, channelConf.Version, channelCache)
|
|
g.Log().Infof(ctx, "Create channel instance success, chanKey: %s, version: %s", chanKey, channelConf.Version)
|
|
}()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (u *sUpstreamAPI) lockChannel(chanKey string) func() {
|
|
u.channelMux.Lock()
|
|
lock := u.channelOps[chanKey]
|
|
if lock == nil {
|
|
lock = &sync.Mutex{}
|
|
u.channelOps[chanKey] = lock
|
|
}
|
|
u.channelMux.Unlock()
|
|
lock.Lock()
|
|
return lock.Unlock
|
|
}
|
|
|
|
func (u *sUpstreamAPI) UnregisterChan(chanKey string) {
|
|
u.Lock()
|
|
delete(u.chanMap, chanKey)
|
|
u.Unlock()
|
|
}
|
|
|
|
func (u *sUpstreamAPI) RegisterChan(chanKey string, version string, channel *model.SimCardChannelCache) {
|
|
u.Lock()
|
|
defer u.Unlock()
|
|
u.chanMap[chanKey] = channel
|
|
}
|
|
|
|
func (u *sUpstreamAPI) removeChan(chanKey string) {
|
|
u.Lock()
|
|
defer u.Unlock()
|
|
delete(u.chanMap, chanKey)
|
|
}
|
|
|
|
func (u *sUpstreamAPI) RegisterTemp(tempKey string, channel model.SimCardTempI) {
|
|
u.Lock()
|
|
defer u.Unlock()
|
|
u.tempMap[tempKey] = channel
|
|
}
|
|
|
|
func (u *sUpstreamAPI) GetTemp(tempKey string) (model.SimCardTempI, bool) {
|
|
u.RLock()
|
|
defer u.RUnlock()
|
|
temp, ok := u.tempMap[tempKey]
|
|
return temp, ok
|
|
}
|
|
|
|
func (u *sUpstreamAPI) GetChan(chanKey string) (*model.SimCardChannelCache, bool) {
|
|
u.RLock()
|
|
defer u.RUnlock()
|
|
channel, ok := u.chanMap[chanKey]
|
|
return channel, ok
|
|
}
|
|
|
|
func (u *sUpstreamAPI) packChanKey(ctx context.Context, simKeys []*model.SimKey) (map[string][]*model.SimKey, error) {
|
|
chanKeyMap := make(map[string][]*model.SimKey)
|
|
simKeyMap := make(map[string]*model.SimKey, len(simKeys))
|
|
unknownChanKeyIccids := make([]string, 0)
|
|
queryRedisIccidKeys := make([]string, 0)
|
|
queryDbIccids := make([]string, 0)
|
|
// 检查通道id是都携带
|
|
for _, simKey := range simKeys {
|
|
simKeyMap[simKey.ICCID] = simKey
|
|
if simKey.ChanKey == "" {
|
|
unknownChanKeyIccids = append(unknownChanKeyIccids, simKey.ICCID)
|
|
queryRedisIccidKeys = append(queryRedisIccidKeys, REDIS_KEY+simKey.ICCID)
|
|
} else {
|
|
chanKeyMap[simKey.ChanKey] = append(chanKeyMap[simKey.ChanKey], simKey)
|
|
}
|
|
}
|
|
if len(unknownChanKeyIccids) == 0 {
|
|
return chanKeyMap, nil
|
|
}
|
|
|
|
allIccid := make(map[string]bool)
|
|
for _, iccid := range unknownChanKeyIccids {
|
|
allIccid[iccid] = false
|
|
}
|
|
// TODO: 临时数据库查询，后面添加内存缓存和Redis缓存
|
|
// 从Redis中查询通道id
|
|
redisChanKeys, err := g.Redis().MGet(ctx, queryRedisIccidKeys...)
|
|
if err != nil {
|
|
g.Log().Error(ctx, "sUpstreamAPI.packChanKey, query redis error %+v, queryRedisIccidKeys %+v", err, queryRedisIccidKeys)
|
|
return nil, gerror.New("查询Redis通道id失败")
|
|
}
|
|
for iccidKey, chanKey := range redisChanKeys {
|
|
iccidSplit := strings.Split(iccidKey, ":")
|
|
if len(iccidSplit) != 4 {
|
|
g.Log().Debug(ctx, "sUpstreamAPI.packChanKey, iccidKey format error, iccidKey:%s", iccidKey)
|
|
continue
|
|
}
|
|
iccid := iccidSplit[3]
|
|
chanKeyStr := chanKey.String()
|
|
if chanKeyStr == "" {
|
|
// 查询redis为空
|
|
g.Log().Debug(ctx, "sUpstreamAPI.packChanKey, query redis empty, iccid:%s, simKeys:%v", iccid, simKeys)
|
|
queryDbIccids = append(queryDbIccids, iccid)
|
|
continue
|
|
}
|
|
allIccid[iccid] = true
|
|
simKey := simKeyMap[iccid]
|
|
if simKey == nil {
|
|
g.Log().Debug(ctx, "sUpstreamAPI.packChanKey, simKey not found, iccid:%s, simKeys:%v", iccid, simKeys)
|
|
continue
|
|
}
|
|
chanKeyMap[chanKeyStr] = append(chanKeyMap[chanKeyStr], &model.SimKey{
|
|
ChanKey: chanKeyStr,
|
|
ICCID: iccid,
|
|
IMSI: simKey.IMSI,
|
|
MSISDN: simKey.MSISDN,
|
|
})
|
|
}
|
|
if len(queryDbIccids) == 0 {
|
|
return chanKeyMap, nil
|
|
}
|
|
|
|
var simCardInfos []*entity.TSimCardInfo
|
|
g.Log().Info(ctx, "sUpstreamAPI.packChanKey, query db, %v", queryDbIccids)
|
|
err = dao.TSimCardInfo.Ctx(ctx).
|
|
Fields(dao.TSimCardInfo.Columns().ChanKey, dao.TSimCardInfo.Columns().Iccid).
|
|
WhereIn(dao.TSimCardInfo.Columns().Iccid, queryDbIccids).Scan(&simCardInfos)
|
|
if err != nil {
|
|
g.Log().Error(ctx, "sUpstreamAPI.packChanKey, query db error %+v, queryDbIccids %+v, simKeys %+v", err, queryDbIccids, simKeys)
|
|
return nil, gerror.New("查询数据库数据失败")
|
|
}
|
|
|
|
for _, simCardInfo := range simCardInfos {
|
|
allIccid[simCardInfo.Iccid] = true
|
|
simKey, _ := simKeyMap[simCardInfo.Iccid]
|
|
chanKeyMap[simCardInfo.ChanKey] = append(chanKeyMap[simCardInfo.ChanKey], &model.SimKey{
|
|
ChanKey: simCardInfo.ChanKey,
|
|
ICCID: simCardInfo.Iccid,
|
|
IMSI: simKey.IMSI,
|
|
MSISDN: simKey.MSISDN,
|
|
})
|
|
}
|
|
// 异步写入Redis
|
|
go u.SetRedis(ctx, simCardInfos)
|
|
for iccid, found := range allIccid {
|
|
if !found {
|
|
chanKeyMap[""] = append(chanKeyMap[""], &model.SimKey{
|
|
ChanKey: "",
|
|
ICCID: iccid,
|
|
})
|
|
}
|
|
}
|
|
return chanKeyMap, nil
|
|
}
|
|
|
|
// 物联网卡
|
|
func (u *sUpstreamAPI) SimBase(ctx context.Context, input *model.SimBaseReqJson) (model.SimBaseResJson, error) {
|
|
output := make([]*model.SimBaseRes[json.RawMessage], 0)
|
|
inputMap := make(map[string]*model.SimBaseReqJson)
|
|
// 提交/恢复/查询阶段的代理异常不是运营商办理失败。
|
|
failureResData := json.RawMessage(`{"status":"-2"}`)
|
|
var apnRequest struct {
|
|
SubmitOnly bool `json:"submitOnly"`
|
|
RecoverOnly bool `json:"recoverOnly"`
|
|
QueryJobID string `json:"queryJobId"`
|
|
}
|
|
request := g.RequestFromCtx(ctx)
|
|
if request != nil && request.URL.Path == "/sim/change/sim-apn" && json.Unmarshal(input.ReqData, &apnRequest) == nil && (apnRequest.SubmitOnly || apnRequest.RecoverOnly || apnRequest.QueryJobID != "") {
|
|
failureResData = json.RawMessage(`{"uncertain":true}`)
|
|
}
|
|
chanKeyMap, err := u.packChanKey(ctx, input.SimKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for chanKey, simKeys := range chanKeyMap {
|
|
inputMap[chanKey] = &model.SimBaseReqJson{
|
|
SimKeys: simKeys,
|
|
ReqData: input.ReqData,
|
|
}
|
|
}
|
|
for chanKey, inputNew := range inputMap {
|
|
if chanKey == "" {
|
|
for _, simKey := range inputNew.SimKeys {
|
|
output = append(output, &model.SimBaseRes[json.RawMessage]{
|
|
SimKey: simKey,
|
|
Code: consts.ApiStatusError,
|
|
Message: "通道未配置",
|
|
ResData: failureResData,
|
|
})
|
|
}
|
|
continue
|
|
}
|
|
channel, ok := u.GetChan(chanKey)
|
|
if !ok || channel.Instance == nil {
|
|
for _, simKey := range inputNew.SimKeys {
|
|
output = append(output, &model.SimBaseRes[json.RawMessage]{
|
|
SimKey: simKey,
|
|
Code: consts.ApiStatusError,
|
|
Message: "通道未实例化",
|
|
ResData: failureResData,
|
|
})
|
|
}
|
|
continue
|
|
}
|
|
out, err := channel.Instance.SimBase(ctx, inputNew)
|
|
if err != nil {
|
|
for _, simKey := range inputNew.SimKeys {
|
|
output = append(output, &model.SimBaseRes[json.RawMessage]{
|
|
SimKey: simKey,
|
|
Code: consts.ApiStatusError,
|
|
Message: err.Error(),
|
|
ResData: failureResData,
|
|
})
|
|
}
|
|
} else {
|
|
for _, item := range out {
|
|
output = append(output, &model.SimBaseRes[json.RawMessage]{
|
|
SimKey: item.SimKey,
|
|
Code: item.Code,
|
|
Message: item.Message,
|
|
ErrCode: item.ErrCode,
|
|
ResData: item.ResData,
|
|
})
|
|
}
|
|
}
|
|
}
|
|
return output, nil
|
|
}
|
|
|
|
// 异步写入redis
|
|
func (u *sUpstreamAPI) SetRedis(ctx context.Context, simCardInfos []*entity.TSimCardInfo) {
|
|
for _, simCardInfo := range simCardInfos {
|
|
_, err := g.Redis().Set(ctx, REDIS_KEY+simCardInfo.Iccid, simCardInfo.ChanKey)
|
|
if err != nil {
|
|
g.Log().Debugf(ctx, "sUpstreamAPI.SetRedis, Set Redis error %+v, key %s, simCardInfo %+v", err, REDIS_KEY+simCardInfo.Iccid, simCardInfo)
|
|
continue
|
|
}
|
|
}
|
|
}
|