chore: 初始开源发布
iotlcm-upstream-proxy：渠道缓存与转发网关。 - 内联原私有依赖：yunze-public → internal/pub/（18 文件）， yunze-pb/pb → internal/pb/（9 文件）；Consul 注册改用官方 gogf/gf/contrib/registry/consul/v2（API 兼容） - Go module：yz-svr → git.5iot.com/iotlcm-public/iotlcm-upstream-proxy - 凭据与内部标识脱敏：MySQL/Redis/Consul 凭据、私网与生产 IP、内网域名 - 移除内部资产：CI 触发器 - Dockerfile 改用公共基础镜像，移除写死的 registry 凭据 - 补 Apache-2.0 LICENSE，重写 README，LF 行尾 + .gitattributes
This commit is contained in:
MiFaZhan
@@ -0,0 +1,61 @@
|
||||
package upstreamapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
|
||||
)
|
||||
|
||||
type errCodeTestChannel struct {
|
||||
response model.SimBaseResJson
|
||||
}
|
||||
|
||||
func (f *errCodeTestChannel) SimBase(context.Context, *model.SimBaseReqJson) (model.SimBaseResJson, error) {
|
||||
return f.response, nil
|
||||
}
|
||||
|
||||
func TestSimBasePreservesCarrierErrCode(t *testing.T) {
|
||||
const payload = `{"code":0,"message":"success","data":[{"code":9999,"message":"http status 797","errCode":"S00004","simKey":{"chanKey":"channel_003","iccid":"8986112520409968013"},"resData":{}}]}`
|
||||
|
||||
var upstream model.SimResponseData[model.SimBaseResJson]
|
||||
if err := json.Unmarshal([]byte(payload), &upstream); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(upstream.Data) != 1 {
|
||||
t.Fatalf("decoded results = %d, want 1", len(upstream.Data))
|
||||
}
|
||||
if upstream.Data[0].ErrCode != "S00004" {
|
||||
t.Fatalf("decoded errCode = %q, want S00004", upstream.Data[0].ErrCode)
|
||||
}
|
||||
|
||||
api := NewUpstreamAPI()
|
||||
api.RegisterChan("channel_003", "test", &model.SimCardChannelCache{
|
||||
Instance: &errCodeTestChannel{response: upstream.Data},
|
||||
})
|
||||
output, err := api.SimBase(context.Background(), &model.SimBaseReqJson{
|
||||
SimKeys: []*model.SimKey{{
|
||||
ChanKey: "channel_003",
|
||||
ICCID: "8986112520409968013",
|
||||
}},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(output) != 1 {
|
||||
t.Fatalf("proxied results = %d, want 1", len(output))
|
||||
}
|
||||
if output[0].ErrCode != "S00004" {
|
||||
t.Fatalf("proxied errCode = %q, want S00004", output[0].ErrCode)
|
||||
}
|
||||
|
||||
encoded, err := json.Marshal(output)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(string(encoded), `"errCode":"S00004"`) {
|
||||
t.Fatalf("serialized response dropped errCode: %s", encoded)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
package upstreamapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/logic/channelconfig"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
|
||||
|
||||
_ "github.com/gogf/gf/contrib/drivers/sqlite/v2"
|
||||
"github.com/gogf/gf/v2/database/gdb"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
type normalGateTestChannel struct {
|
||||
model.SimCardChannelI
|
||||
}
|
||||
|
||||
type normalGateTestTemplate struct {
|
||||
calls int
|
||||
lastConfig *model.SimChannelConfigItemCache
|
||||
}
|
||||
|
||||
func (f *normalGateTestTemplate) NewTemp(_ context.Context, cfg *model.SimChannelConfigItemCache) model.SimCardChannelI {
|
||||
f.calls++
|
||||
f.lastConfig = cfg
|
||||
return &normalGateTestChannel{}
|
||||
}
|
||||
|
||||
// Exercise real ORM decoding and config versioning as well as both creation paths.
|
||||
// All SQL runs against a temporary SQLite file, never a configured environment DB.
|
||||
func TestUnregisterChanIsIdempotent(t *testing.T) {
|
||||
api := NewUpstreamAPI()
|
||||
api.RegisterChan("missing", "v1", &model.SimCardChannelCache{})
|
||||
api.UnregisterChan("missing")
|
||||
api.UnregisterChan("missing")
|
||||
if _, ok := api.GetChan("missing"); ok {
|
||||
t.Fatal("channel still registered after repeated unregister")
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelNormalCreationGate(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
if err := gdb.SetConfigGroup("default", gdb.ConfigGroup{{
|
||||
Type: "sqlite", Name: filepath.Join(t.TempDir(), "channels.db"),
|
||||
}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db := g.DB()
|
||||
t.Cleanup(func() { _ = db.Close(ctx) })
|
||||
execSQL := func(query string, args ...any) {
|
||||
t.Helper()
|
||||
if _, err := db.Exec(ctx, query, args...); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
execSQL("CREATE TABLE t_sim_channels (id INTEGER PRIMARY KEY, chan_key TEXT, temp_key TEXT, chan_name TEXT, is_normal INTEGER NULL)")
|
||||
execSQL("CREATE TABLE t_sim_channel_template (id INTEGER PRIMARY KEY, temp_key TEXT, temp_model TEXT)")
|
||||
execSQL("INSERT INTO t_sim_channel_template (id, temp_key, temp_model) VALUES (1, 'gate_test', 'direct')")
|
||||
|
||||
oldConfig := service.ChannelConfig()
|
||||
t.Cleanup(func() { service.RegisterChannelConfig(oldConfig) })
|
||||
|
||||
for _, path := range []string{"loop", "refresh"} {
|
||||
t.Run(path, func(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
value any
|
||||
eligible bool
|
||||
}{
|
||||
{"yes_is_zero", 0, true},
|
||||
{"no_is_one", 1, false},
|
||||
{"null", nil, false},
|
||||
{"unexpected", 2, false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
execSQL("DELETE FROM t_sim_channels")
|
||||
execSQL("INSERT INTO t_sim_channels (id, chan_key, temp_key, is_normal) VALUES (1, 'gate_channel', 'gate_test', ?)", tc.value)
|
||||
config := channelconfig.NewChannelConfig()
|
||||
service.RegisterChannelConfig(config)
|
||||
api := NewUpstreamAPI()
|
||||
factory := &normalGateTestTemplate{}
|
||||
api.RegisterTemp("gate_test", factory)
|
||||
reload := func() {
|
||||
t.Helper()
|
||||
var err error
|
||||
if path == "loop" {
|
||||
err = config.LoadAllChannelConfig(ctx)
|
||||
if err == nil {
|
||||
err = api.loop(ctx)
|
||||
}
|
||||
} else {
|
||||
err = api.RefreshChannel(ctx, "gate_channel")
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("refresh must succeed: %v", err)
|
||||
}
|
||||
}
|
||||
reload()
|
||||
cfg, ok := config.GetChannel("gate_channel")
|
||||
if !ok || cfg.CanCreateInstance() != tc.eligible {
|
||||
t.Fatalf("wrong eligibility after ORM scan: %+v", cfg)
|
||||
}
|
||||
if tc.value == nil && cfg.IsNormal != nil {
|
||||
t.Fatalf("SQL NULL decoded as non-nil: %v", *cfg.IsNormal)
|
||||
}
|
||||
if cfg.Channel.ChanKey != "gate_channel" {
|
||||
t.Fatal("embedded channel fields were not scanned")
|
||||
}
|
||||
_, exists := api.GetChan("gate_channel")
|
||||
if exists != tc.eligible || (factory.calls == 1) != tc.eligible {
|
||||
t.Fatalf("instance exists=%v factory calls=%d eligible=%v", exists, factory.calls, tc.eligible)
|
||||
}
|
||||
initialVersion := cfg.Version
|
||||
|
||||
// Only is_normal changes: the configuration cache must notice it.
|
||||
execSQL("UPDATE t_sim_channels SET is_normal=0")
|
||||
reload()
|
||||
current, ok := api.GetChan("gate_channel")
|
||||
if !ok || current.Instance == nil {
|
||||
t.Fatal("eligible channel was not instantiated")
|
||||
}
|
||||
cfg, _ = config.GetChannel("gate_channel")
|
||||
if !tc.eligible && cfg.Version == initialVersion {
|
||||
t.Fatal("changing only is_normal did not change config version")
|
||||
}
|
||||
oldCalls := factory.calls
|
||||
|
||||
// Changing only is_normal must evict the old runtime instance.
|
||||
execSQL("UPDATE t_sim_channels SET is_normal=1")
|
||||
reload()
|
||||
if _, exists := api.GetChan("gate_channel"); exists || factory.calls != oldCalls {
|
||||
t.Fatal("ineligible channel retained or recreated its runtime instance")
|
||||
}
|
||||
cfg, _ = config.GetChannel("gate_channel")
|
||||
if cfg.CanCreateInstance() || cfg.Version == current.Version {
|
||||
t.Fatal("ineligible configuration was not cached with a new version")
|
||||
}
|
||||
// Disabled refreshes remain successful even if the template disappears.
|
||||
execSQL("UPDATE t_sim_channels SET chan_name='changed while disabled', temp_key='missing_template'")
|
||||
reload()
|
||||
reload()
|
||||
if _, exists := api.GetChan("gate_channel"); exists || factory.calls != oldCalls {
|
||||
t.Fatal("repeated ineligible refresh was not an idempotent removal")
|
||||
}
|
||||
|
||||
// A process restart cannot recreate the disabled instance.
|
||||
|
||||
restarted := NewUpstreamAPI()
|
||||
restarted.RegisterTemp("gate_test", factory)
|
||||
if err := restarted.loop(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := restarted.GetChan("gate_channel"); ok {
|
||||
t.Fatal("reload recreated disabled instance")
|
||||
}
|
||||
|
||||
execSQL("UPDATE t_sim_channels SET is_normal=0, temp_key='gate_test'")
|
||||
reload()
|
||||
restored, exists := api.GetChan("gate_channel")
|
||||
if !exists || restored == current || factory.calls != oldCalls+1 ||
|
||||
factory.lastConfig.Channel.ChanName != "changed while disabled" {
|
||||
t.Fatal("restored channel did not rebuild using the latest configuration")
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,388 @@
|
||||
package upstreamapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"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"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"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)
|
||||
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: json.RawMessage(`{"status": "-2"}`),
|
||||
})
|
||||
}
|
||||
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: json.RawMessage(`{"status": "-2"}`),
|
||||
})
|
||||
}
|
||||
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: json.RawMessage(`{"status": "-2"}`),
|
||||
})
|
||||
}
|
||||
} 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
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user