fix: 说明文件

avsd

.gitignore

@@ -0,0 +1,47 @@
+# ---> VisualStudioCode
+.vscode/*
+!.vscode/settings.json
+!.vscode/tasks.json
+!.vscode/launch.json
+!.vscode/extensions.json
+!.vscode/*.code-snippets
+
+# Local History for Visual Studio Code
+.history/
+
+# Run temp file and dir
+crontask/log/
+crontask/ftp/
+crontask/database/
+crontask/export/
+crontask/temp
+crontask/crontask
+
+restagent/backup/
+restagent/log/
+restagent/upload/
+restagent/software/
+restagent/database/
+restagent/restagent
+
+sshsvc/sshsvc
+sshsvc/mmllog/
+sshsvc/mmlhome/
+sshsvc/log/
+
+tools/loadmconf/loadmconf
+
+vendor
+
+# Built Visual Studio Code Extensions
+*.vsix
+*.log
+*.log-*
+*.bak
+
+.idea
+omc_log
+data/ftp
+lib/FTP
+assets
+__debug_*

.vscode/launch.json

@@ -0,0 +1,17 @@
+{
+    // 使用 IntelliSense 了解相关属性。 
+    // 悬停以查看现有属性的描述。
+    // 欲了解更多信息，请访问: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "调试",
+            "type": "go",
+            "request": "launch",
+            "mode": "debug",
+            "program": "./nbi_alarm.go",
+            "args": ["-c", "conf/nbi_alarm.json"],
+            "console": "integratedTerminal"
+        }
+    ]
+}

README.md

@@ -1,3 +0,0 @@
-# nbi_alarm
-
-北向-移动集采测试

conf/global.go

@@ -0,0 +1,77 @@
+package conf
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/aceld/zinx/zconf"
+	"github.com/aceld/zinx/zlog"
+)
+
+type Channel struct {
+	TCPPort    int    `json:"tcp_port"`    //当前通道的TCP监听端口
+	BindFlag   string `json:"bind_flag"`   //当前通道bind 的网元信息
+	Province   string `json:"province"`    //网元所在省份
+	DeviceCode string `json:"device_code"` //主机编码 四位，每1位可用0-9、A-Z编码
+}
+
+type Config struct {
+	// 网元通道
+	Channel []Channel `json:"channel"`
+
+	// 数据库连接
+	Mysql string `json:"mysql"`
+
+	//以下是zinx 的配置
+	Name            string `json:"name"`
+	MaxConn         int    `json:"max_conn"`
+	WorkerPoolSize  int    `json:"worker_pool_size"`
+	HeartbeatMax    int    `json:"heartbeat_max"`
+	HbcheckInterval int    `json:"hbcheck_interval"`
+	LogDir          string `json:"log_dir"`
+	LogFile         string `json:"log_file"`
+
+	// FTP文件服务
+	FTPRoot string `json:"ftp_root"`
+}
+
+var OmcConf Config
+
+func Reload(path string) error {
+	confFileExists, _ := zconf.PathExists(path)
+	if !confFileExists {
+		zlog.Ins().ErrorF("Config File %s is not exist!!", path)
+		return errors.New("config file is not exist")
+	}
+
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return err
+	}
+
+	err = json.Unmarshal(data, &OmcConf)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func Init(path string) {
+	if path == "" || strings.HasSuffix(path, "zinx.json") {
+		pwd, err := os.Getwd()
+		if err != nil {
+			pwd = "."
+		}
+		path = pwd + "/conf/nbi_alarm.json"
+		fmt.Println("path", path)
+	}
+
+	if err := Reload(path); err != nil {
+		zlog.Ins().ErrorF("Config File %s error, ", err)
+
+		panic(err)
+	}
+}

conf/nbi_alarm.json

@@ -0,0 +1,23 @@
+{
+  "channel": [
+    {
+      "tcp_port": 31232,
+      "bind_flag": "SMF#SZ_01",
+      "province": "BJ",
+      "device_code": "0001"
+    }
+  ],
+
+  "mysql": "root:1000omc@kp!@tcp(192.168.0.229:33066)/omc_db?charset=utf8mb4&parseTime=True&loc=Local",
+
+  "mame":"nbi north alarm agent",
+  "max_conn": 3000,
+  "worker_pool_size": 100,
+  "heartbeat_max": 180,
+  "hbcheck_interval": 60,
+  "log_dir": "./assets/nbi_alarm",
+  "log_file":"nbi_alarm.log",
+
+  "ftp_root": "/data/ftp"
+  
+}

core/consts/consts.go

@@ -0,0 +1,7 @@
+package consts
+
+//定义user type
+const (
+	MSG  = "msg"
+	FILE = "ftp"
+)

core/db/mysql.go

@@ -0,0 +1,37 @@
+package db
+
+import (
+	"omc/conf"
+
+	"github.com/aceld/zinx/zlog"
+	"gorm.io/driver/mysql"
+	"gorm.io/gorm"
+)
+
+var Client *gorm.DB
+
+func Init() error {
+	// 日志输出sql观察
+	// newLogger := logger.New(
+	// 	log.New(os.Stdout, "\r\n", log.LstdFlags), // io writer
+	// 	logger.Config{
+	// 		SlowThreshold:             time.Second, // Slow SQL threshold
+	// 		LogLevel:                  logger.Info, // Log level
+	// 		IgnoreRecordNotFoundError: true,        // Ignore ErrRecordNotFound error for logger
+	// 		ParameterizedQueries:      false,       // Don't include params in the SQL log
+	// 		Colorful:                  false,       // Disable color
+	// 	},
+	// )
+	// db, err := gorm.Open(mysql.Open(conf.OmcConf.Mysql), &gorm.Config{Logger: newLogger})
+
+	db, err := gorm.Open(mysql.Open(conf.OmcConf.Mysql), &gorm.Config{})
+	if err != nil {
+		zlog.Ins().ErrorF("open mysql %s error, ", conf.OmcConf.Mysql, err)
+		panic(err)
+	}
+	sqlDB, _ := db.DB()
+	sqlDB.SetMaxOpenConns(20)
+	sqlDB.SetMaxIdleConns(10)
+	Client = db
+	return nil
+}

core/decoder/decode_omc.go

@@ -0,0 +1,88 @@
+package decoder
+
+import (
+	"bytes"
+	"encoding/binary"
+	"encoding/hex"
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+	"math"
+)
+
+//ffff 01 00999999 000a 68656c6c6f2074657374
+// +---------------+---------------+--------------------+----------------+-------------------
+// |    开始标志     |     type       |     秒时间戳       |  长度           |     消息体
+// | 0xffff(2byte) |  uint8(1byte)  |     uint32(4byte) |  uint16(2byte) |    bytes(N byte)
+// +---------------+---------------+--------------------+----------------+-------------------
+
+const OmcMsgHeaderSize = 9 //表示OMC空包长度
+
+type OmcDecoder struct {
+	StartSign uint16
+	MsgType   uint8
+	TimeStamp uint32
+	LenOfBody uint16
+	Value     []byte
+}
+
+func NewOmcDecoder() ziface.IDecoder {
+	return &OmcDecoder{}
+}
+
+func (omc *OmcDecoder) GetLengthField() *ziface.LengthField {
+	return &ziface.LengthField{
+		MaxFrameLength:      math.MaxUint16 + 9,
+		LengthFieldOffset:   7,
+		LengthFieldLength:   2,
+		LengthAdjustment:    0,
+		InitialBytesToStrip: 0,
+		//注意现在默认是大端，使用小端需要指定编码方式
+		Order: binary.BigEndian,
+	}
+}
+
+func (omc *OmcDecoder) decode(data []byte) *OmcDecoder {
+	ltvData := OmcDecoder{}
+	ltvData.StartSign = binary.BigEndian.Uint16(data[0:2])
+	ltvData.MsgType = data[2]
+	ltvData.TimeStamp = binary.BigEndian.Uint32(data[3:7])
+	ltvData.LenOfBody = binary.BigEndian.Uint16(data[7:9])
+	//Determine the length of V. (确定V的长度)
+	ltvData.Value = make([]byte, ltvData.LenOfBody)
+
+	//5. Get V
+	binary.Read(bytes.NewBuffer(data[OmcMsgHeaderSize:OmcMsgHeaderSize+ltvData.LenOfBody]), binary.BigEndian, ltvData.Value)
+
+	return &ltvData
+}
+
+func (omc *OmcDecoder) Intercept(chain ziface.IChain) ziface.IcResp {
+	//1. Get the IMessage of zinx
+	iMessage := chain.GetIMessage()
+	if iMessage == nil {
+		// Go to the next layer in the chain of responsibility
+		return chain.ProceedWithIMessage(iMessage, nil)
+	}
+
+	//2. Get Data
+	data := iMessage.GetData()
+	zlog.Ins().DebugF("omc-RawData size:%d data:%s\n", len(data), hex.EncodeToString(data))
+
+	// (读取的数据不超过包头，直接进入下一层)
+	if len(data) < OmcMsgHeaderSize {
+		return chain.ProceedWithIMessage(iMessage, nil)
+	}
+
+	//4.  Decode
+	ltvData := omc.decode(data)
+	zlog.Ins().DebugF("omc-decode  %v", ltvData)
+
+	// (将解码后的数据重新设置到IMessage中, Zinx的Router需要MsgID来寻址)
+	iMessage.SetDataLen(uint32(ltvData.LenOfBody))
+	iMessage.SetMsgID(uint32(ltvData.MsgType))
+	iMessage.SetData(ltvData.Value)
+
+	//6. Pass the decoded data to the next layer.
+	// (将解码后的数据进入下一层)
+	return chain.ProceedWithIMessage(iMessage, *ltvData)
+}

core/dpack/datapack_omc.go

@@ -0,0 +1,117 @@
+package dpack
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"github.com/aceld/zinx/zlog"
+	"github.com/aceld/zinx/zpack"
+	"time"
+
+	"github.com/aceld/zinx/zconf"
+	"github.com/aceld/zinx/ziface"
+)
+
+// +---------------+---------------+--------------------+----------------+-------------------
+// |    开始标志     |     type       |     秒时间戳       |  长度           |     消息体
+// | 0xffff(2byte) |  uint8(1byte)  |     uint32(4byte) |  uint16(2byte) |    bytes(N byte)
+// +---------------+---------------+--------------------+----------------+-------------------
+
+var defaultHeaderLen uint32 = 9
+
+type DataPack struct{}
+
+// NewDataPack initializes a packing and unpacking instance
+// (封包拆包实例初始化方法)
+func NewDataPack() ziface.IDataPack {
+	return &DataPack{}
+}
+
+// GetHeadLen returns the length of the message header
+// (获取包头长度方法)
+func (dp *DataPack) GetHeadLen() uint32 {
+	//ID uint32(4 bytes) +  DataLen uint32(4 bytes)
+	return defaultHeaderLen
+}
+
+// Pack packs the message (compresses the data)
+// (封包方法,压缩数据)
+func (dp *DataPack) Pack(msg ziface.IMessage) ([]byte, error) {
+	zlog.Ins().InfoF("my pack: %v", msg)
+	// Create a buffer to store the bytes
+	// (创建一个存放bytes字节的缓冲)
+	dataBuff := bytes.NewBuffer([]byte{})
+
+	// Write the oxffff
+	if err := binary.Write(dataBuff, binary.BigEndian, uint16(0xffff)); err != nil {
+		return nil, err
+	}
+	//Write the type
+	if err := binary.Write(dataBuff, binary.BigEndian, uint8(msg.GetMsgID())); err != nil {
+		return nil, err
+	}
+
+	//Write the timestamp
+	if err := binary.Write(dataBuff, binary.BigEndian, uint32(time.Now().Unix())); err != nil {
+		return nil, err
+	}
+
+	//Write the length
+	if err := binary.Write(dataBuff, binary.BigEndian, uint16(msg.GetDataLen())); err != nil {
+		return nil, err
+	}
+
+	// Write the data
+	if err := binary.Write(dataBuff, binary.BigEndian, msg.GetData()); err != nil {
+		return nil, err
+	}
+
+	return dataBuff.Bytes(), nil
+}
+
+// Unpack unpacks the message (decompresses the data)
+// (拆包方法,解压数据)
+func (dp *DataPack) Unpack(binaryData []byte) (ziface.IMessage, error) {
+	// Create an ioReader for the input binary data
+	dataBuff := bytes.NewReader(binaryData)
+
+	// Only unpack the header information to obtain the data length and message ID
+	// (只解压head的信息，得到dataLen和msgID)
+	msg := &zpack.Message{}
+
+	// Read the startSign
+	var startSign uint16
+	if err := binary.Read(dataBuff, binary.BigEndian, &startSign); err != nil {
+		return nil, err
+	}
+
+	// Read the msgID
+	var msgType uint
+	if err := binary.Read(dataBuff, binary.BigEndian, &msgType); err != nil {
+		return nil, err
+	}
+	msg.ID = uint32(msgType)
+
+	// read timeStamp
+	var timeStamp uint32
+	if err := binary.Read(dataBuff, binary.BigEndian, &timeStamp); err != nil {
+		return nil, err
+	}
+
+	// Read the data length
+	var length uint16
+	if err := binary.Read(dataBuff, binary.BigEndian, &length); err != nil {
+		return nil, err
+	}
+	msg.DataLen = uint32(length)
+
+	// Check whether the data length exceeds the maximum allowed packet size
+	// (判断dataLen的长度是否超出我们允许的最大包长度)
+	if zconf.GlobalObject.MaxPacketSize > 0 && msg.GetDataLen() > zconf.GlobalObject.MaxPacketSize {
+		return nil, errors.New("too large msg data received")
+	}
+
+	// Only the header data needs to be unpacked, and then another data read is performed from the connection based on the header length
+	// (这里只需要把head的数据拆包出来就可以了，然后再通过head的长度，再从conn读取一次数据)
+	return msg, nil
+}

core/emun/orig_severity.go

@@ -0,0 +1,17 @@
+package emun
+
+// 原始级别 中文转数值
+func OrigSeverity(os string) int64 {
+	switch os {
+	case "Critical":
+		return 1
+	case "Major":
+		return 2
+	case "Minor":
+		return 3
+	case "Warning":
+		return 4
+	default:
+		return 0
+	}
+}

core/heart_beat/heart_beat.go

@@ -0,0 +1,25 @@
+package heartbeat
+
+import (
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+)
+
+var HeadBeatMsgID uint32 = 255
+
+// MyHeartBeatMsg 用户自定义的心跳检测消息处理方法
+func MyHeartBeatMsg(conn ziface.IConnection) []byte {
+	return []byte("heartbeat, I am server, I am alive")
+}
+
+// MyHeartBeat 用户自定义心跳发送函数
+func MyHeartBeat(conn ziface.IConnection) error {
+	return nil
+}
+
+// MyOnRemoteNotAlive 用户自定义的远程连接不存活时的处理方法
+func MyOnRemoteNotAlive(conn ziface.IConnection) {
+	zlog.Ins().ErrorF("myOnRemoteNotAlive is Called, connID=%v", conn.GetConnID(), "remoteAddr =%v ", conn.RemoteAddr())
+	//关闭链接
+	conn.Stop()
+}

core/interceptor/global.go

@@ -0,0 +1,42 @@
+package interceptor
+
+import (
+	"encoding/hex"
+	"fmt"
+	"omc/core/parse"
+	"time"
+
+	"github.com/aceld/zinx/ziface"
+)
+
+type GlobalInterceptor struct{}
+
+func (m *GlobalInterceptor) Intercept(chain ziface.IChain) ziface.IcResp {
+	request := chain.Request() //从责任链中获取当前拦截器的输入数据
+
+	// 这一层是自定义拦截器处理逻辑，这里只是简单打印输入
+	iRequest := request.(ziface.IRequest) //注意：由于Zinx的Request类型，这里需要做一下断言转换
+
+	fmt.Println("\n\n=========自定义拦截器=====")
+
+	body, err := parse.RequestBodyDecode(iRequest, nil)
+	fmt.Printf("消息ID: %v \n", iRequest.GetMsgID())
+	fmt.Printf("原始数据: %v \n", body.RawData)
+	fmt.Printf("原始字符: %v \n", hex.EncodeToString(body.RawData))
+	fmt.Printf("原始字符: %v \n", string(body.RawData))
+	fmt.Printf("用户ID: %v \n", body.UID)
+	fmt.Printf("收到消息: %v  %v \n", body.Name, body.Data)
+	fmt.Printf("错误：%v \n", err)
+
+	// return chain.Proceed(chain.Request()) //进入并执行下一个拦截器
+
+	iMessage := chain.GetIMessage()
+	resp := chain.ProceedWithIMessage(iMessage, iRequest)
+	fmt.Printf("目标消息ID: %v \n", iMessage.GetMsgID())
+	fmt.Printf("收到消息长度: %v \n", iMessage.GetDataLen())
+	fmt.Printf("信息时间：%v \n", time.Now().String())
+
+	fmt.Print("=========自定义拦截器=====\n\n\n")
+
+	return resp
+}

core/manage/manage.go

@@ -0,0 +1,51 @@
+package manage
+
+import (
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+)
+
+// OnConnectionAdd 当客户端建立连接的时候的hook函数
+func OnConnectionAdd(conn ziface.IConnection) {
+	//创建一个user
+
+	user := NewUser(conn, conn.RemoteAddrString())
+	//将当前新上线玩家添加到ChannelManager中
+	m := GetManager(conn.GetName())
+	if m == nil {
+		zlog.Ins().ErrorF("server internal error in GetManager")
+		conn.Stop()
+		return
+	}
+	m.AddUser(user)
+
+	//将该连接绑定属性PID
+	conn.SetProperty("UID", user.UID)
+
+	zlog.Ins().InfoF("====> User uID = %s", user.UID, " arrived ====", "")
+}
+
+// OnConnectionLost 当客户端断开连接的时候的hook函数
+func OnConnectionLost(conn ziface.IConnection) {
+	//获取当前连接的PID属性
+	uID, _ := conn.GetProperty("UID")
+	var userID string
+	if uID != nil {
+		userID = uID.(string)
+	}
+
+	//根据pID获取对应usr
+	m := GetManager(conn.GetName())
+	if m == nil {
+		zlog.Ins().ErrorF("server internal error in GetManager")
+		return
+	}
+	user := m.GetUserByPID(userID)
+
+	//触发玩家下线业务
+	if user != nil {
+		user.LostConnection(m)
+	}
+
+	zlog.Ins().InfoF("====> User  %s-%s", user.UID, user.UserName, " left =====")
+}

core/manage/user.go

@@ -0,0 +1,50 @@
+package manage
+
+import (
+	"fmt"
+
+	"github.com/aceld/zinx/ziface"
+	"github.com/google/uuid"
+)
+
+// User User 对象
+type User struct {
+	UID        string             //ID
+	Conn       ziface.IConnection //当前User的连接
+	LoginState bool
+	LoginCount int
+	AlarmType  string
+	UserName   string
+	SeqNo      string // 登录随机码
+	RemoteIp   string
+}
+
+// NewUser 创建一个user对象
+func NewUser(conn ziface.IConnection, addr string) *User {
+	p := &User{
+		UID:      uuid.NewString(),
+		Conn:     conn,
+		RemoteIp: addr,
+	}
+	return p
+}
+
+// LostConnection User下线
+func (p *User) LostConnection(m *ChannelManager) {
+	m.RemoveUserByPID(p.UID)
+}
+
+// SendMsg /*
+func (p *User) SendMsg(msgID uint32, msg []byte) {
+	if p.Conn == nil {
+		fmt.Println("connection in player is nil")
+		return
+	}
+
+	//调用SendMsg发包
+	if err := p.Conn.SendMsg(msgID, msg); err != nil {
+		fmt.Println("Player SendMsg error !")
+		return
+	}
+
+}

core/manage/user_manager.go

@@ -0,0 +1,237 @@
+package manage
+
+import (
+	"encoding/json"
+	"errors"
+	"omc/core/consts"
+	"omc/core/db"
+	"omc/core/parse"
+	"omc/handle/model"
+	"omc/handle/service"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/aceld/zinx/zlog"
+)
+
+/*
+ChannelManager 通道管理管理模块
+*/
+type ChannelManager struct {
+	Name       string
+	User       map[string]*User //当前在线的User集合
+	BindFlag   string           //bind 的网元标识，ne_type#ne_id 格式
+	Province   string           //网元所在省份
+	DeviceCode string           //主机编码 四位，每1位可用0-9、A-Z编码
+	AlarmSeq   int64            //当前告警的序列号
+	pLock      sync.RWMutex     //保护User的互斥读写机制
+}
+
+// MgrObj 多个通道管理模块集合
+var MgrObj []*ChannelManager
+
+// NewManager New 提供ChannelManager 初始化 方法
+func NewManager(name, bind, province, deviceCode string) *ChannelManager {
+	newManager := ChannelManager{
+		Name:       name,
+		BindFlag:   bind,
+		Province:   province,
+		DeviceCode: deviceCode,
+		User:       make(map[string]*User),
+	}
+	MgrObj = append(MgrObj, &newManager)
+	return &newManager
+}
+
+func GetManager(name string) *ChannelManager {
+	for i := range MgrObj {
+		if MgrObj[i].Name == name {
+			return MgrObj[i]
+		}
+	}
+	return nil
+}
+
+// AddUser 提供添加一个user的功能
+func (wm *ChannelManager) AddUser(User *User) {
+	//将User添加到 世界管理器中
+	wm.pLock.Lock()
+	wm.User[User.UID] = User
+	wm.pLock.Unlock()
+}
+
+// Talk User广播消息
+func (wm *ChannelManager) Talk(msgID uint32, msg []byte) {
+	//2. 得到所有的在线user
+	users := wm.GetAllUser()
+
+	//3. 向所有的user发送消息
+	for _, user := range users {
+		if user.LoginState && user.AlarmType == consts.MSG {
+			user.SendMsg(msgID, msg)
+		}
+	}
+}
+
+// LoginSuccess  登录成功管理
+func (wm *ChannelManager) LoginSuccess(UID, name, tp string) error {
+	wm.pLock.Lock()
+	defer wm.pLock.Unlock()
+	//判断是否重复登录
+	for _, v := range wm.User {
+		if v.UserName == name && v.AlarmType == tp && v.LoginState {
+			v.Conn.Stop()
+			return errors.New("repeat login for the account")
+		}
+	}
+	user, ok := wm.User[UID]
+	if !ok {
+		return errors.New("server internal error")
+	}
+	user.UserName = name
+	user.AlarmType = tp
+	user.LoginState = true
+	wm.User[UID] = user
+	return nil
+}
+
+// SetSeqNo 设置登录随机码
+func (wm *ChannelManager) SetSeqNo(UID, seqNo string) error {
+	wm.pLock.Lock()
+	defer wm.pLock.Unlock()
+	//判断是否重复登录
+	user, ok := wm.User[UID]
+	if !ok {
+		return errors.New("server internal error")
+	}
+	user.SeqNo = seqNo
+	return nil
+}
+
+// LoginFail 登录失败管理
+func (wm *ChannelManager) LoginFail(UID string) (bool, error) {
+	wm.pLock.Lock()
+	defer wm.pLock.Unlock()
+	//判断是否重复登录
+	user, ok := wm.User[UID]
+	if !ok {
+		return true, errors.New("server internal error")
+	}
+	user.LoginCount++
+	wm.User[UID] = user
+	if user.LoginCount >= 3 {
+		return true, errors.New("too many attempts, close connect")
+	}
+	return false, nil
+}
+
+// RemoveUserByPID 从信息表中移除一个user
+func (wm *ChannelManager) RemoveUserByPID(uid string) {
+	wm.pLock.Lock()
+	delete(wm.User, uid)
+	wm.pLock.Unlock()
+}
+
+// GetUserByPID  通过user ID 获取对应user信息
+func (wm *ChannelManager) GetUserByPID(uID string) *User {
+	wm.pLock.RLock()
+	defer wm.pLock.RUnlock()
+
+	return wm.User[uID]
+}
+
+// GetAllUser 获取所有user的信息
+func (wm *ChannelManager) GetAllUser() []*User {
+	wm.pLock.RLock()
+	defer wm.pLock.RUnlock()
+
+	//创建返回的User集合切片
+	User := make([]*User, 0)
+
+	//添加切片
+	for _, v := range wm.User {
+		if v.LoginState && v.AlarmType == consts.MSG {
+			User = append(User, v)
+		}
+	}
+
+	//返回
+	return User
+}
+
+func (wm *ChannelManager) RealTimeAlarm() {
+	for {
+		wm.pLock.RLock()
+		//查询
+		var alarms []model.OmcAlarm
+		neBind, _ := parse.ConvertBindFlag(wm.BindFlag)
+		var lastAlarmSeq = wm.AlarmSeq
+		if wm.AlarmSeq == 0 {
+			lastAlarmSeq = service.LastAlarmSeq(neBind.NeType, neBind.NeId)
+		} else {
+			list, err := service.List(neBind.NeType, neBind.NeId, wm.AlarmSeq)
+			if err != nil {
+				zlog.Ins().ErrorF("db error %s", err)
+			}
+			alarms = service.ConvertOMCAlarm(list)
+		}
+		//上报实时告警信息
+		if len(alarms) > 0 {
+			go wm.SendAlarm(alarms)
+			lastAlarmSeq = service.MaxAlarmSeq(lastAlarmSeq, alarms)
+		}
+		var users []string
+		for _, user := range wm.User {
+			if user.LoginState && user.AlarmType == consts.MSG {
+				userInfo := strings.Join([]string{user.UserName, user.RemoteIp}, ";")
+				users = append(users, userInfo)
+			}
+		}
+
+		//更新AlarmSeq
+		wm.AlarmSeq = lastAlarmSeq + 1
+		wm.pLock.RUnlock()
+
+		//记录日志到alarm 日志表
+		for _, v := range alarms {
+			for _, u := range users {
+				var alarmLog model.NbiAlarmLog
+				ui := strings.Split(u, ";")
+				if len(ui) == 2 {
+					alarmLog.OpUser = ui[0]
+					alarmLog.SrcIp = ui[1]
+				}
+				alarmLog.NeType = v.NeType
+				alarmLog.NeId = v.NeUID
+				alarmLog.AlarmSeq = int64(v.AlarmSeq)
+				alarmLog.AlarmId = v.AlarmId
+				alarmLog.EventTime = v.EventTime
+				alarmLog.LogTime = time.Now()
+
+				db.Client.Create(&alarmLog)
+			}
+		}
+		time.Sleep(3 * time.Second)
+	}
+}
+
+func (wm *ChannelManager) SendAlarm(alarms []model.OmcAlarm) error {
+	for _, v := range alarms {
+		//生产告警内容
+		data, err := json.Marshal(v)
+		if err != nil {
+			zlog.Ins().ErrorF("SendAlarm json Marshalerror %v", err)
+		} else {
+			//发送告警内容
+			wm.Talk(0, data)
+		}
+	}
+	return nil
+}
+
+func (wm *ChannelManager) UpdateAlarmSeq(alarmSeq int64) {
+	wm.pLock.Lock()
+	wm.AlarmSeq = alarmSeq
+	wm.pLock.Unlock()
+}

core/model/body.go

@@ -0,0 +1,9 @@
+package model
+
+// 请求数据
+type Body struct {
+	UID     string            // 连接实例ID
+	RawData []byte            // 原始数据
+	Name    string            // 请求名
+	Data    map[string]string // 数据Key
+}

core/parse/parse.go

@@ -0,0 +1,126 @@
+package parse
+
+import (
+	"errors"
+	"fmt"
+	"omc/core/model"
+	"strings"
+
+	"github.com/aceld/zinx/ziface"
+)
+
+// 网元类型#网元标记
+type NeBind struct {
+	NeType string
+	NeId   string
+}
+
+// 转换解析服务端绑定的网元
+func ConvertBindFlag(bindFlag string) (NeBind, error) {
+	var neBind NeBind
+	nb := strings.Split(bindFlag, "#")
+	if len(nb) != 2 {
+		return neBind, errors.New("ne bind flag invalid")
+	}
+	neBind.NeType = nb[0]
+	neBind.NeId = nb[1]
+	return neBind, nil
+}
+
+// RequestBodyDecode 请求消息解析
+// checker 检查参数必传
+func RequestBodyDecode(request ziface.IRequest, checker []string) (model.Body, error) {
+	// 消息处理
+	body := model.Body{}
+	err := Decode(request.GetData(), &body)
+	if err != nil {
+		return body, errors.New("inlaid message body")
+	}
+
+	// 检查key
+	if len(checker) > 0 {
+		for _, v := range checker {
+			if _, ok := body.Data[v]; !ok {
+				return body, errors.New("missing parameter body " + v)
+			}
+		}
+	}
+
+	// 当前连接实例ID
+	uID, err := request.GetConnection().GetProperty("UID")
+	if err != nil {
+		request.GetConnection().Stop()
+		return body, errors.New("server internal error")
+	}
+	body.UID = uID.(string)
+	return body, nil
+}
+
+// Decode 数据解析
+// reqLoginAlarm;user=yiy;key=qw#$@;type=msg
+func Decode(data []byte, body *model.Body) error {
+	body.RawData = data
+
+	multi := strings.Split(string(data), ";")
+	if len(multi) < 1 {
+		return errors.New("invalid msg body")
+	}
+
+	// 获取函数名
+	if multi[0] != "" {
+		name := multi[0]
+
+		idx := strings.LastIndex(name, "\x14")
+		if idx == -1 {
+			idx = strings.LastIndex(name, "\x00")
+		}
+		if idx == -1 {
+			idx = strings.LastIndex(name, "\xe2")
+		}
+
+		if idx > 0 {
+			name = name[idx+1:]
+			name = strings.Replace(name, "\"", "", 1)
+			name = strings.Replace(name, "'", "", 1)
+			name = strings.Replace(name, "#", "", 1)
+		}
+		body.Name = name
+	}
+
+	// 解析data KEY
+	body.Data = make(map[string]string)
+	for i := 1; i < len(multi); i++ {
+		str := multi[i]
+		idx := strings.Index(str, "=")
+		if idx > 0 {
+			key := str[:idx]
+			value := str[idx+1:]
+			body.Data[key] = value
+		}
+	}
+
+	return nil
+}
+
+// Pack 数据压缩
+func Pack(name string, data map[string]string) []byte {
+	var multi []string
+	multi = append(multi, name)
+	// 允许的拓展字段 "seqNo", "fileName"
+	// 固定值顺序 "result", "reqId", "resDesc"
+	keys := []string{"seqNo", "reqId", "result", "fileName", "resDesc"}
+	for _, key := range keys {
+		if v, ok := data[key]; ok {
+			// 长度小于32个字符，不允许带分号“;”
+			if key == "resDesc" && len(v) > 32 {
+				v = v[0:32]
+			}
+			if key == "resDesc" && strings.Contains(v, ";") {
+				v = strings.ReplaceAll(v, ";", "")
+			}
+			multi = append(multi, fmt.Sprintf("%s=%s", key, v))
+		}
+	}
+	raw := strings.Join(multi, ";")
+	return []byte(raw)
+}

core/result.go

@@ -0,0 +1,39 @@
+package core
+
+import (
+	"omc/core/parse"
+)
+
+// Result
+// 配合 request.GetConnection().SendMsg()
+// resDesc 原因长度小于32字符
+func Result(name string, data map[string]string) []byte {
+	return parse.Pack(name, data)
+}
+
+// ResultError ackLoginAlarm;result=fail;resDesc=username-error
+// request.GetConnection().SendMsg(omc.AckSyncAlarmMsg, core.ResultError("ackSyncAlarmMsg", err.Error(), ""))
+func ResultError(name string, desc, reqID string) []byte {
+	data := map[string]string{
+		"result":  "fail",
+		"resDesc": desc,
+	}
+	if reqID != "" {
+		data["reqId"] = reqID
+	}
+
+	return Result(name, data)
+}
+
+// request.GetConnection().SendMsg(omc.AckSyncAlarmMsg, core.ResultSuccess("ackSyncAlarmMsg", err.Error(), ""))
+func ResultSuccess(name string, desc, reqID string) []byte {
+	data := map[string]string{
+		"result":  "succ",
+		"resDesc": desc,
+	}
+	if reqID != "" {
+		data["reqId"] = reqID
+	}
+
+	return Result(name, data)
+}

core/utils/bcrypt.go

@@ -0,0 +1,14 @@
+package utils
+
+import "golang.org/x/crypto/bcrypt"
+
+// Encrypt 加密明文密码
+func Encrypt(password string) (string, error) {
+	hashedBytes, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
+	return string(hashedBytes), err
+}
+
+// Compare 密文校验
+func Compare(hashedPassword, password string) error {
+	return bcrypt.CompareHashAndPassword([]byte(hashedPassword), []byte(password))
+}

core/utils/cakey.go

@@ -0,0 +1,41 @@
+package utils
+
+import (
+	"crypto"
+	"crypto/rsa"
+	"crypto/sha256"
+	"crypto/x509"
+	"encoding/pem"
+	"fmt"
+	"strings"
+)
+
+// 验证私钥的签名， publicKey从证书里获取公钥 certificate.PublicKey
+func VerifySignature(publicKey any, data, signature []byte) bool {
+	rsaPublicKey := publicKey.(*rsa.PublicKey)
+	hashed := sha256.Sum256(data)
+	err := rsa.VerifyPKCS1v15(rsaPublicKey, crypto.SHA256, hashed[:], signature)
+
+	return err == nil
+}
+
+// 解析证书信息
+func ParseCert(carCertificate string) (*x509.Certificate, error) {
+	// 解析头尾需要换行符
+	result := strings.Split(carCertificate, "-----")
+	result[2] = "\n" + result[2] + "\n"
+	carCertificate = strings.Join(result, "-----")
+
+	// 解码证书
+	block, _ := pem.Decode([]byte(carCertificate))
+	if block == nil {
+		return nil, fmt.Errorf("无法解码证书")
+	}
+
+	// 解析证书
+	certificate, err := x509.ParseCertificate(block.Bytes)
+	if err != nil {
+		return nil, err
+	}
+	return certificate, nil
+}

core/utils/file.go

@@ -0,0 +1,145 @@
+package utils
+
+import (
+	"archive/zip"
+	"encoding/json"
+	"fmt"
+	"io"
+	"omc/handle/model"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/aceld/zinx/zlog"
+)
+
+//BJ/HX/RJ/OMC/FM/告警文件生成时间
+// /FTP根目录/省份简称/专业简称/厂家编码/OMC名称/数据类别/日期或时间/
+//<省份简称>-<数据类别>-<网元类型>[-网元子类]-<主机编号>-<数据版本>-<数据时间>[-登录用户名][-同步请求标识][-Ri][-统计周期] [-序列号].<后缀>
+// BJ-FM-OMC-主机编码-v0-告警文件生成时间-001.txt
+
+type FileNameMeta struct {
+	FTPRoot    string `json:"ftp_root"`    // FTP根目录
+	Province   string `json:"province"`    // 网元所在省份
+	DeviceCode string `json:"device_code"` // 主机编码 四位，每1位可用0-9、A-Z编码
+	ReqId      string `json:"req_id"`      // 同步请求标识
+	Suffix     string `json:"suffix"`      // 文件后缀
+}
+
+// 生成告警文件文件名
+// GD-FM-OMC-100A-V1.1.0-20150611011603-001.txt.zip
+func FileName(meta *FileNameMeta) string {
+	// 路径分层
+	timeDayStr := time.Now().Format("20060102")
+	filePath := path.Join(meta.FTPRoot, meta.Province, "HX", "RJ", "OMC", "FM", timeDayStr)
+
+	// 文件名组合
+	reqIdStr := fmt.Sprintf("%03s", meta.ReqId)
+	timeStr := time.Now().Format("20060102150405")
+	fileNameGroup := []string{meta.Province, "FM", "OMC", meta.DeviceCode, "V0", timeStr, reqIdStr}
+	fileName := strings.Join(fileNameGroup, "-")
+	fileName = strings.ToUpper(fileName) + meta.Suffix
+	return path.Join(filePath, fileName)
+}
+
+func CreateFile(filePath string, data []model.OmcAlarm) error {
+	// 获取文件所在的目录路径
+	dirPath := filepath.Dir(filePath)
+
+	// 确保文件夹路径存在
+	err := os.MkdirAll(dirPath, os.ModePerm)
+	if err != nil {
+		zlog.Ins().ErrorF("创建文件夹失败 CreateFile %v", err)
+	}
+
+	// 创建或打开文件
+	file, err := os.Create(filePath)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	// 遍历切片并将每个元素序列化为 JSON 字符串并写入文件
+	for _, alarm := range data {
+		jsonData, err := json.Marshal(alarm)
+		if err != nil {
+			return err
+		}
+
+		// 写入 JSON 字符串到文件，并换行
+		_, err = file.WriteString(string(jsonData) + "\r\n")
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// 将文件添加到 ZIP 压缩文件
+func FileToZip(zipFileName, filePath string) error {
+	// 创建一个新的 ZIP 文件
+	newZipFile, err := os.Create(zipFileName)
+	if err != nil {
+		return fmt.Errorf("创建 ZIP 文件失败: %w", err)
+	}
+	defer newZipFile.Close()
+
+	// 创建 ZIP 写入器
+	zipWriter := zip.NewWriter(newZipFile)
+	defer zipWriter.Close()
+
+	fileToCompress, err := os.Open(filePath)
+	if err != nil {
+		return fmt.Errorf("打开文件失败: %w", err)
+	}
+	defer fileToCompress.Close()
+
+	// 获取文件信息
+	fileInfo, err := fileToCompress.Stat()
+	if err != nil {
+		return fmt.Errorf("获取文件信息失败: %w", err)
+	}
+
+	// 创建文件头
+	header, err := zip.FileInfoHeader(fileInfo)
+	if err != nil {
+		return fmt.Errorf("创建文件头失败: %w", err)
+	}
+
+	// 设置文件头中的名称
+	header.Name = fileInfo.Name()
+
+	// 创建文件在 ZIP 中的写入器
+	writer, err := zipWriter.CreateHeader(header)
+	if err != nil {
+		return fmt.Errorf("创建文件写入器失败: %w", err)
+	}
+
+	// 将文件内容复制到 ZIP 文件中
+	_, err = io.Copy(writer, fileToCompress)
+	if err != nil {
+		return fmt.Errorf("将文件内容复制到 ZIP 失败: %w", err)
+	}
+
+	return nil
+}
+
+// 文件同级进行zip压缩
+func FileToCompress(filePath string, data []model.OmcAlarm, delOrgFile bool) (string, error) {
+	defer os.Remove(filePath)
+	// 导出文件
+	err := CreateFile(filePath, data)
+	if err != nil {
+		return "", err
+	}
+
+	// 压缩文件
+	zipFfileNmae := filePath + ".zip"
+	err = FileToZip(zipFfileNmae, filePath)
+	if err != nil {
+		return "", err
+	}
+	return zipFfileNmae, nil
+}

core/utils/strrand.go

@@ -0,0 +1,25 @@
+package utils
+
+import (
+	"math/rand"
+	"time"
+)
+
+// 生成编号 SeqNo 0-9A-Z
+func SeqNo(length int) string {
+	items := []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "A", "B", "C", "D", "E", "F"}
+
+	// 创建基于时间的随机生成器
+	source := rand.NewSource(time.Now().UnixNano())
+	rng := rand.New(source)
+
+	// 生成32位长度的字符串值
+	result := ""
+	for i := 0; i < length; i++ {
+		randomIndex := rng.Intn(len(items))
+		randomItem := items[randomIndex]
+		result += randomItem
+	}
+
+	return result
+}

go.mod

@@ -0,0 +1,18 @@
+module omc
+
+go 1.18
+
+require (
+	github.com/aceld/zinx v1.1.21
+	github.com/google/uuid v1.3.0
+	golang.org/x/crypto v0.11.0
+	gorm.io/driver/mysql v1.5.1
+	gorm.io/gorm v1.25.1
+)
+
+require (
+	github.com/go-sql-driver/mysql v1.7.0 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
+)

go.sum

@@ -0,0 +1,40 @@
+github.com/aceld/zinx v1.1.21 h1:8zoZ+hcEAd7gDsl8xOKPaWPEs9vZDRQOvhjG3vuvAnQ=
+github.com/aceld/zinx v1.1.21/go.mod h1:nITkdASGtkLSwNKZ5yj88IpcCHTCFCP6cL12JWms1Fo=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc=
+github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
+github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
+github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
+github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+golang.org/x/crypto v0.11.0 h1:6Ewdq3tDic1mg5xRO4milcWCfMVQhI4NkqWWvqejpuA=
+golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gorm.io/driver/mysql v1.5.1 h1:WUEH5VF9obL/lTtzjmML/5e6VfFR/788coz2uaVCAZw=
+gorm.io/driver/mysql v1.5.1/go.mod h1:Jo3Xu7mMhCyj8dlrb3WoCaRd1FhsVh+yMXb1jUInf5o=
+gorm.io/gorm v1.25.1 h1:nsSALe5Pr+cM3V1qwwQ7rOkw+6UeLrX5O4v3llhHa64=
+gorm.io/gorm v1.25.1/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k=

handle/api/close_conn_alarm.go

@@ -0,0 +1,22 @@
+package api
+
+import (
+	"omc/core"
+
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/znet"
+)
+
+var CloseConnAlarmMsgID uint32 = 10
+var CloseConnAlarmMsgType uint32 = 10
+var CloseConnAlarmMsgName string = "closeConnAlarm"
+
+// closeConnAlarm  关闭连接
+type CloseConnAlarm struct {
+	znet.BaseRouter
+}
+
+func (s *CloseConnAlarm) Handle(request ziface.IRequest) {
+	request.GetConnection().Stop()
+	request.GetConnection().SendMsg(CloseConnAlarmMsgType, core.ResultSuccess(CloseConnAlarmMsgName, "", ""))
+}

handle/api/req_cmca_login_alarm.go

@@ -0,0 +1,80 @@
+package api
+
+import (
+	"omc/core"
+	"omc/core/manage"
+	"omc/core/parse"
+	"omc/handle/service"
+
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+	"github.com/aceld/zinx/znet"
+)
+
+var ReqCMCALoginAlarmMsgID uint32 = 11
+var ReqCMCALoginAlarmMsgType uint32 = 11
+var ReqCMCALoginAlarmMsgName string = "reqCMCALoginAlarm"
+
+// reqCMCALoginAlarm  CMCA认证方式登录
+type ReqCMCALoginAlarm struct {
+	znet.BaseRouter
+}
+
+func (*ReqCMCALoginAlarm) Handle(request ziface.IRequest) {
+	// 登录消息处理
+	body, err := parse.RequestBodyDecode(request, []string{"user", "key", "cert", "type"})
+	username := body.Data["user"]
+	key := body.Data["key"]
+	cert := body.Data["cert"]
+	tp := body.Data["type"]
+	if err != nil || username == "" || cert == "" || key == "" || tp == "" {
+		zlog.Ins().ErrorF("inlaid message body %s", err.Error())
+		request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
+		return
+	}
+
+	// 获取当前请求的通道
+	m := manage.GetManager(request.GetConnection().GetName())
+	if m == nil {
+		zlog.Ins().ErrorF("server internal error")
+		request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, "server internal error", ""))
+		return
+	}
+	uid, err := request.GetConnection().GetProperty("UID")
+	if err != nil {
+		zlog.Ins().ErrorF("GetProperty UID error %s", err)
+		request.GetConnection().Stop()
+		return
+	}
+
+	// 得到连接实例的随机码
+	seqNo := m.GetUserByPID(uid.(string)).SeqNo
+	content := username + ":" + seqNo
+
+	// 检查判断是否对应随机码
+	if err := service.CMCALogin(username, cert, content, key); err != nil {
+		zlog.Ins().ErrorF("LoginFail %s", err)
+		request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
+		// 已登录的，登录错误超过3次,断开连接
+		if uid != nil || uid != "" {
+			isClose, _ := m.LoginFail(uid.(string))
+			if isClose {
+				request.GetConnection().Stop()
+				return
+			}
+		}
+		return
+	}
+
+	// manager 用户登录更新
+	err = m.LoginSuccess(uid.(string), username, tp)
+	if err != nil {
+		zlog.Ins().ErrorF("manager:%s", err)
+		request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
+		return
+	}
+
+	zlog.Ins().InfoF("user login loginSuccess,username:%s, type:%s, channel:%s", username, tp, request.GetConnection().GetName())
+	request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultSuccess(ReqLoginAlarmMsgName, "null", ""))
+
+}

handle/api/req_cmca_login_seq.go

@@ -0,0 +1,47 @@
+package api
+
+import (
+	"omc/core"
+	"omc/core/manage"
+	"omc/core/utils"
+
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+	"github.com/aceld/zinx/znet"
+)
+
+var ReqCMCALoginSeqMsgID uint32 = 12
+var ReqCMCALoginSeqMsgType uint32 = 13
+var ReqCMCALoginSeqMsgName string = "ackCMCALoginSeq"
+
+// reqCMCALoginSeq CMCA认证方式登录随机码
+type ReqCMCALoginSeq struct {
+	znet.BaseRouter
+}
+
+// reqCMCALoginAlarm;user=omc;key=base64Key;cert=cer;type=msg"
+func (s *ReqCMCALoginSeq) Handle(request ziface.IRequest) {
+
+	// 获取当前请求的通道
+	m := manage.GetManager(request.GetConnection().GetName())
+	if m == nil {
+		zlog.Ins().ErrorF("server internal error")
+		request.GetConnection().SendMsg(ReqCMCALoginSeqMsgType, core.ResultError(ReqCMCALoginSeqMsgName, "server internal error", ""))
+		return
+	}
+	uid, err := request.GetConnection().GetProperty("UID")
+	if err != nil {
+		zlog.Ins().ErrorF("GetProperty UID error %s", err)
+		request.GetConnection().Stop()
+		return
+	}
+	// seqNo 32位长度 0-9A-Z
+	seqNo := utils.SeqNo(32)
+	m.SetSeqNo(uid.(string), seqNo)
+
+	// 发送随机码去sha256加密
+	msgData := map[string]string{
+		"seqNo": seqNo,
+	}
+	request.GetConnection().SendMsg(ReqCMCALoginSeqMsgType, core.Result(ReqCMCALoginSeqMsgName, msgData))
+}

handle/api/req_heart_beat.go

@@ -0,0 +1,40 @@
+package api
+
+import (
+	"omc/core"
+	"omc/core/parse"
+
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+	"github.com/aceld/zinx/znet"
+)
+
+var ReqHeartBeatMsgID uint32 = 8
+var ReqHeartBeatMsgType uint32 = 9
+var ReqHeartBeatName string = "ackHeartBeat"
+
+// ReqHeartBeat  心跳
+type ReqHeartBeat struct {
+	znet.BaseRouter
+}
+
+// reqHeartBeat;reqId=12
+func (s *ReqHeartBeat) Handle(request ziface.IRequest) {
+	// 解包
+	body, err := parse.RequestBodyDecode(request, nil)
+	if err != nil {
+		zlog.Ins().ErrorF("inlaid message body %s", err.Error())
+		request.GetConnection().SendMsg(ReqHeartBeatMsgType, core.ResultError(ReqHeartBeatName, "inlaid message body", ""))
+		return
+	}
+
+	reqId, ok := body.Data["reqId"]
+	if !ok {
+		zlog.Ins().ErrorF("missing parameter of message body")
+		request.GetConnection().SendMsg(ReqHeartBeatMsgType, core.ResultError(ReqHeartBeatName, "missing reqId", ""))
+		return
+	}
+
+	//ack
+	request.GetConnection().SendMsg(ReqHeartBeatMsgType, core.ResultSuccess(ReqHeartBeatName, "", reqId))
+}

handle/api/req_login_alarm.go

@@ -0,0 +1,76 @@
+package api
+
+import (
+	"omc/core"
+	"omc/core/manage"
+	"omc/core/parse"
+	"omc/handle/service"
+
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+	"github.com/aceld/zinx/znet"
+)
+
+var ReqLoginAlarmMsgID uint32 = 1
+var ReqLoginAlarmMsgType uint32 = 2
+var ReqLoginAlarmMsgName string = "ackLoginAlarm"
+
+// reqLoginAlarm  登录
+type ReqLoginAlarm struct {
+	znet.BaseRouter
+}
+
+// reqLoginAlarm;user=omc;key=omc;type=msg
+func (s *ReqLoginAlarm) Handle(request ziface.IRequest) {
+	// 登录消息处理
+	body, err := parse.RequestBodyDecode(request, []string{"user", "key", "type"})
+	// 账户密码
+	username := body.Data["user"]
+	key := body.Data["key"]
+	tp := body.Data["type"]
+	if err != nil || username == "" || key == "" {
+		zlog.Ins().ErrorF("inlaid message body %s", err.Error())
+		request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, "inlaid message body", ""))
+		return
+	}
+
+	// 获取当前请求的通道
+	m := manage.GetManager(request.GetConnection().GetName())
+	if m == nil {
+		zlog.Ins().ErrorF("server internal error")
+		request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, "server internal error", ""))
+		return
+	}
+	uid, err := request.GetConnection().GetProperty("UID")
+	if err != nil {
+		zlog.Ins().ErrorF("GetProperty UID error %s", err)
+		request.GetConnection().Stop()
+		return
+	}
+
+	// 登录信息
+	err = service.UserLogin(username, key)
+	if err != nil {
+		zlog.Ins().ErrorF("LoginFail %s", err)
+		request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
+		// 已登录的，登录错误超过3次,断开连接
+		if uid != nil || uid != "" {
+			isClose, _ := m.LoginFail(uid.(string))
+			if isClose {
+				request.GetConnection().Stop()
+				return
+			}
+		}
+		return
+	}
+
+	// manager 用户登录更新
+	err = m.LoginSuccess(uid.(string), username, tp)
+	if err != nil {
+		zlog.Ins().ErrorF("manager:%s", err)
+		request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
+		return
+	}
+	zlog.Ins().InfoF("user login loginSuccess,username:%s, type:%s, channel:%s", username, tp, request.GetConnection().GetName())
+	request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultSuccess(ReqLoginAlarmMsgName, "null", ""))
+}

handle/api/req_sync_alarm_file.go

@@ -0,0 +1,113 @@
+package api
+
+import (
+	"fmt"
+	"omc/conf"
+	"omc/core"
+	"omc/core/manage"
+	"omc/core/parse"
+	"omc/core/utils"
+	"omc/handle/service"
+	"strconv"
+
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+	"github.com/aceld/zinx/znet"
+)
+
+var ReqSyncAlarmFileID uint32 = 5
+var ReqSyncAlarmFileType uint32 = 6
+var ReqSyncAlarmFileName string = "ackSyncAlarmFile"
+var ReqSyncAlarmFileResultType uint32 = 7
+var ReqSyncAlarmFileResultName string = "ackSyncAlarmFileResult"
+
+// reqSyncAlarmFile 文件方式同步告警请求
+type ReqSyncAlarmFile struct {
+	znet.BaseRouter
+}
+
+func (s *ReqSyncAlarmFile) Handle(request ziface.IRequest) {
+	// 消息处理
+	checker := []string{"reqId"}
+	body, err := parse.RequestBodyDecode(request, checker)
+	reqId := body.Data["reqId"]
+	if err != nil || reqId == "" {
+		zlog.Ins().ErrorF("inlaid message body %s", err.Error())
+		request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultError(ReqSyncAlarmFileName, err.Error(), ""))
+		return
+	}
+
+	// 参数预设初始值 alarmSeq不一定有
+	alarmSeq, alarmSeqOk := body.Data["alarmSeq"]
+	if !alarmSeqOk || alarmSeq == "" {
+		fmt.Println("== ", alarmSeq)
+		alarmSeq = "1"
+	}
+
+	//管理模块
+	m := manage.GetManager(request.GetConnection().GetName())
+	if m == nil {
+		zlog.Ins().ErrorF("server internal error")
+		request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultError(ReqSyncAlarmFileName, "server internal error", reqId))
+		return
+	}
+
+	// 检查用户是否登录
+	u := m.GetUserByPID(body.UID)
+	if !u.LoginState {
+		zlog.Ins().ErrorF("no permissions ")
+		request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultError(ReqSyncAlarmFileName, "no permissions", reqId))
+		return
+	}
+
+	// 同步的消息序号
+	seq, err := strconv.Atoi(alarmSeq)
+	if err != nil {
+		seq = 0
+	}
+
+	// check alarmSeq 是否存在
+	neBind, _ := parse.ConvertBindFlag(m.BindFlag)
+	alarm, err := service.ListForSaveFile(neBind.NeType, neBind.NeId, int64(seq), body.Data["startTime"], body.Data["endTime"], body.Data["syncSource"])
+	if err != nil {
+		zlog.Ins().ErrorF("no permissions ")
+		request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultError(ReqSyncAlarmFileName, "no permissions", reqId))
+		return
+	}
+	if len(alarm) > 0 {
+		data := service.ConvertOMCAlarm(alarm)
+
+		// 生成文件名
+		fileNmae := utils.FileName(&utils.FileNameMeta{
+			FTPRoot:    conf.OmcConf.FTPRoot,
+			Province:   m.Province,
+			DeviceCode: m.DeviceCode,
+			ReqId:      reqId,
+			Suffix:     ".txt",
+		})
+
+		// 打压缩
+		zipFfileNmae, err := utils.FileToCompress(fileNmae, data, true)
+		if err != nil {
+			zlog.Ins().ErrorF("ReqSyncAlarmFileResult FileToCompress %v", err)
+			request.GetConnection().SendMsg(ReqSyncAlarmFileResultType, core.Result(ReqSyncAlarmFileResultName, map[string]string{
+				"reqId":    reqId,
+				"result":   "fail",
+				"resDesc":  err.Error(),
+				"fileName": "null",
+			}))
+			return
+		}
+
+		//发送文件同步信息
+		request.GetConnection().SendMsg(ReqSyncAlarmFileResultType, core.Result(ReqSyncAlarmFileResultName, map[string]string{
+			"reqId":    reqId,
+			"result":   "succ",
+			"resDesc":  "null",
+			"fileName": zipFfileNmae,
+		}))
+		return
+	}
+
+	request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultSuccess(ReqSyncAlarmFileName, "null", reqId))
+}

handle/api/req_sync_alarm_msg.go

@@ -0,0 +1,75 @@
+package api
+
+import (
+	"omc/core"
+	"omc/core/manage"
+	"omc/core/parse"
+	"omc/handle/service"
+	"strconv"
+
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/zlog"
+	"github.com/aceld/zinx/znet"
+)
+
+var ReqSyncAlarmMsgID uint32 = 3
+var ReqSyncAlarmMsgType uint32 = 4
+var ReqSyncAlarmMsgName string = "ackSyncAlarmMsg"
+
+// reqSyncAlarmMsg 消息同步
+type ReqSyncAlarmMsg struct {
+	znet.BaseRouter
+}
+
+// reqSyncAlarmMsg
+func (s *ReqSyncAlarmMsg) Handle(request ziface.IRequest) {
+	// 消息处理
+	checker := []string{"reqId", "alarmSeq"}
+	body, err := parse.RequestBodyDecode(request, checker)
+	reqId := body.Data["reqId"]
+	if err != nil || reqId == "" {
+		zlog.Ins().ErrorF("inlaid message body %s", err.Error())
+		request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, err.Error(), ""))
+		return
+	}
+
+	//管理模块
+	m := manage.GetManager(request.GetConnection().GetName())
+	if m == nil {
+		zlog.Ins().ErrorF("server internal error")
+		request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "server internal error", reqId))
+		return
+	}
+
+	// 检查用户是否登录
+	u := m.GetUserByPID(body.UID)
+	if !u.LoginState {
+		zlog.Ins().ErrorF("no permissions ")
+		request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "no permissions", reqId))
+		return
+	}
+
+	alarmSeq, err := strconv.Atoi(body.Data["alarmSeq"])
+	if err != nil || alarmSeq < 1 {
+		zlog.Ins().ErrorF("invalid alarmSeq of message body")
+		request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "invalid alarmSeq", reqId))
+		return
+	}
+
+	//check alarmSeq 是否存在
+	neBind, _ := parse.ConvertBindFlag(m.BindFlag)
+	alarmSeqCount := service.AlarmSeqCount(neBind.NeType, neBind.NeId, int64(alarmSeq))
+	if alarmSeqCount > 0 {
+		if alarmSeqCount > 1000 {
+			request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "over 1000 rows", reqId))
+		} else {
+			// 发送更新上报的alarm seq
+			m.UpdateAlarmSeq(int64(alarmSeq))
+			request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultSuccess(ReqSyncAlarmMsgName, "", reqId))
+		}
+		return
+	}
+
+	zlog.Ins().ErrorF("not has of alarmSeq %d", alarmSeq)
+	request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "alarmSeq does not exist", reqId))
+}

handle/model/alarm.go

@@ -0,0 +1,43 @@
+package model
+
+import "time"
+
+// Alarm 实时告警上报信息
+type Alarm struct {
+	ID                int64     `json:"id" gorm:"id"`
+	AlarmSeq          int64     `json:"alarm_seq" gorm:"alarm_seq"`
+	AlarmId           string    `json:"alarm_id" gorm:"alarm_id"`
+	AlarmTitle        string    `json:"alarm_title" gorm:"alarm_title"`
+	NeType            string    `json:"ne_type" gorm:"ne_type"`
+	NeId              string    `json:"ne_id" gorm:"ne_id"`
+	AlarmCode         int64     `json:"alarm_code" gorm:"alarm_code"`
+	EventTime         time.Time `json:"event_time" gorm:"event_time"`
+	AlarmType         string    `json:"alarm_type" gorm:"alarm_type"`
+	OrigSeverity      string    `json:"orig_severity" gorm:"orig_severity"`           // 1: Critical, 2: Major, 3: Minor, 4: Warning, 5: Event(Only VNF)
+	PerceivedSeverity string    `json:"perceived_severity" gorm:"perceived_severity"` // 1: Critical, 2: Major, 3: Minor, 4: Warning, 5: Event(Only VNF)
+	PvFlag            string    `json:"pv_flag" gorm:"pv_flag"`
+	NeName            string    `json:"ne_name" gorm:"ne_name"`
+	ObjectUid         string    `json:"object_uid" gorm:"object_uid"`
+	ObjectName        string    `json:"object_name" gorm:"object_name"`
+	ObjectType        string    `json:"object_type" gorm:"object_type"`
+	LocationInfo      string    `json:"location_info" gorm:"location_info"`
+	Province          string    `json:"province" gorm:"province"`
+	AlarmStatus       int64     `json:"alarm_status" gorm:"alarm_status"`
+	SpecificProblem   string    `json:"specific_problem" gorm:"specific_problem"`
+	SpecificProblemId string    `json:"specific_problem_id" gorm:"specific_problem_id"`
+	AddInfo           string    `json:"add_info" gorm:"add_info"`
+	Counter           int64     `json:"counter" gorm:"counter"`
+	LatestEventTime   time.Time `json:"latest_event_time" gorm:"latest_event_time"`
+	AckState          int8      `json:"ack_state" gorm:"ack_state"` // 0: Unacked, 1: Acked
+	AckTime           time.Time `json:"ack_time" gorm:"ack_time"`
+	AckUser           string    `json:"ack_user" gorm:"ack_user"`
+	ClearType         int8      `json:"clear_type" gorm:"clear_type"` // 0: Unclear, 1: AutoClear, 2: ManualClear
+	ClearTime         time.Time `json:"clear_time" gorm:"clear_time"`
+	ClearUser         string    `json:"clear_user" gorm:"clear_user"`
+	Timestamp         time.Time `json:"timestamp" gorm:"timestamp"`
+}
+
+// TableName 表名称
+func (*Alarm) TableName() string {
+	return "alarm"
+}

handle/model/nbi_alarm_log.go

@@ -0,0 +1,24 @@
+package model
+
+import "time"
+
+// NbiAlarmLog undefined
+type NbiAlarmLog struct {
+	ID          int64     `json:"id" gorm:"id"`
+	OpUser      string    `json:"op_user" gorm:"op_user"`
+	SrcIp       string    `json:"src_ip" gorm:"src_ip"`
+	NeType      string    `json:"ne_type" gorm:"ne_type"`
+	NeId        string    `json:"ne_id" gorm:"ne_id"`
+	AlarmSeq    int64     `json:"alarm_seq" gorm:"alarm_seq"`
+	AlarmId     string    `json:"alarm_id" gorm:"alarm_id"`
+	AlarmCode   int64     `json:"alarm_code" gorm:"alarm_code"`
+	AlarmStatus int64     `json:"alarm_status" gorm:"alarm_status"` // 0/1: 清除告警/活动告警
+	EventTime   string    `json:"event_time" gorm:"event_time"`
+	LogTime     time.Time `json:"log_time" gorm:"log_time"`
+	AId         int64     `json:"a_id" gorm:"a_id"`
+}
+
+// TableName 表名称
+func (*NbiAlarmLog) TableName() string {
+	return "nbi_alarm_log"
+}

handle/model/omc_alarm.go

@@ -0,0 +1,25 @@
+package model
+
+// OmcAlarm 告警上报的结构体信息
+type OmcAlarm struct {
+	AlarmSeq          int64  `json:"alarmSeq"`          //告警序列号
+	AlarmTitle        string `json:"alarmTitle"`        //告警事件标题
+	AlarmStatus       int64  `json:"alarmStatus"`       //告警状态
+	AlarmType         string `json:"alarmType"`         //告警类型
+	OrigSeverity      int64  `json:"origSeverity"`      //原始级别
+	EventTime         string `json:"eventTime"`         //事件发生时间
+	AlarmId           string `json:"alarmId"`           //告警事件唯一标识
+	SpecificProblemID string `json:"specificProblemID"` //告警问题原因ID
+	SpecificProblem   string `json:"specificProblem"`   //告警问题原因
+	NeUID             string `json:"neUID"`             //告警网元UID
+	NeName            string `json:"neName"`            //告警网元名称
+	NeType            string `json:"neType"`            //告警网元设备类型
+	ObjectUID         string `json:"objectUID"`         //告警定位对象UID
+	ObjectName        string `json:"objectName"`        //告警定位对象名称
+	ObjectType        string `json:"objectType"`        //告警定位对象资源类型
+	LocationInfo      string `json:"locationInfo"`      //告警定位信息
+	AddInfo           string `json:"addInfo"`           //告警辅助信息[条件必选]
+	PVFlag            string `json:"PVFlag"`            //网元虚实性[条件必选]
+	Province          string `json:"province"`          //网元服务省份
+	Unit              string `json:"unit"`              //网元服务省份
+}

handle/model/user.go

@@ -0,0 +1,41 @@
+package model
+
+import "time"
+
+// User 用户表实体信息
+type User struct {
+	ID                 int64     `json:"id" gorm:"id"`
+	AccountId          string    `json:"account_id" gorm:"account_id"`
+	Name               string    `json:"name" gorm:"name"`
+	RealName           string    `json:"real_name" gorm:"real_name"`
+	Sn                 string    `json:"sn" gorm:"sn"`
+	Gender             string    `json:"gender" gorm:"gender"`
+	Email              string    `json:"email" gorm:"email"`
+	IdCardNumber       string    `json:"id_card_number" gorm:"id_card_number"`
+	Description        string    `json:"description" gorm:"description"`
+	TelephoneNumber    string    `json:"telephone_number" gorm:"telephone_number"`
+	Phone              string    `json:"phone" gorm:"phone"`
+	Mobile             string    `json:"mobile" gorm:"mobile"`
+	EmployeeNumber     string    `json:"employee_number" gorm:"employee_number"`
+	EmployeeType       string    `json:"employee_type" gorm:"employee_type"`
+	Organize           string    `json:"organize" gorm:"organize"`
+	SupporterCorpName  string    `json:"supporter_corp_name" gorm:"supporter_corp_name"`
+	StartTime          string    `json:"start_time" gorm:"start_time"`
+	EndTime            string    `json:"end_time" gorm:"end_time"`
+	Password           string    `json:"password" gorm:"password"`
+	PasswordSha512     string    `json:"password_sha512" gorm:"password_sha512"`
+	ChangePasswordFlag int8      `json:"change_password_flag" gorm:"change_password_flag"`
+	PasswordExpiration time.Time `json:"password_expiration" gorm:"password_expiration"`
+	Status             string    `json:"status" gorm:"status"`
+	UserExpiration     time.Time `json:"user_expiration" gorm:"user_expiration"`
+	GroupName          string    `json:"group_name" gorm:"group_name"`
+	Profile            string    `json:"profile" gorm:"profile"`
+	CreateTime         time.Time `json:"create_time" gorm:"create_time"`
+	UpdateTime         time.Time `json:"update_time" gorm:"update_time"`
+	Unit               string    `json:"unit" gorm:"unit"`
+}
+
+// TableName 表名称
+func (*User) TableName() string {
+	return "user"
+}

handle/service/alarm.go

@@ -0,0 +1,191 @@
+package service
+
+import (
+	"errors"
+	"omc/core/db"
+	"omc/core/emun"
+	"omc/handle/model"
+	"time"
+
+	"github.com/aceld/zinx/zlog"
+)
+
+// AlarmSeqCount alarm_seq大于等于的记录数
+func AlarmSeqCount(neType, neId string, alarmSeq int64) int64 {
+	var num int64
+	query := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ?", neType, neId)
+	if alarmSeq > 0 {
+		query = query.Where("alarm_seq >= ?", alarmSeq)
+	}
+	if err := query.Order("alarm_seq desc").Count(&num).Error; err != nil {
+		zlog.Ins().ErrorF("HasAlarmSeq db error %s", err)
+		return num
+	}
+	return num
+}
+
+// List 获取告警信息 alarm_seq大于等于的
+func List(neType, neId string, alarmSeq int64) ([]model.Alarm, error) {
+	// 从数据库读取告警记录
+	var alarms []model.Alarm
+	if err := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ? and alarm_seq >= ?", neType, neId, alarmSeq).Order("alarm_seq asc").Find(&alarms).Error; err != nil {
+		zlog.Ins().ErrorF("AlarmList db error %s", err)
+		return nil, err
+	}
+	return alarms, nil
+}
+
+// ConvertOMCAlarm 转换上报的告警信息
+func ConvertOMCAlarm(alarms []model.Alarm) []model.OmcAlarm {
+	result := make([]model.OmcAlarm, 0)
+	for _, v := range alarms {
+		var item model.OmcAlarm
+		item.AlarmSeq = v.AlarmSeq
+		item.AlarmTitle = v.AlarmTitle
+		item.AlarmStatus = v.AlarmStatus
+		item.AlarmType = v.AlarmType
+		item.OrigSeverity = emun.OrigSeverity(v.OrigSeverity)
+		item.EventTime = v.EventTime.Format("2006-01-02 15:04:05")
+		item.AlarmId = v.AlarmId
+		item.SpecificProblemID = v.SpecificProblemId
+		item.SpecificProblem = v.SpecificProblem
+		item.NeUID = v.NeId
+		item.NeName = v.NeName
+		item.NeType = v.NeType
+		item.ObjectUID = v.ObjectUid
+		item.ObjectName = v.NeName
+		item.ObjectType = v.ObjectType
+		item.LocationInfo = v.LocationInfo
+		item.AddInfo = v.AddInfo
+		item.PVFlag = v.PvFlag
+		item.Province = v.Province
+		result = append(result, item)
+	}
+	return result
+}
+
+// 检查列表里最大的AlarmSeq
+func MaxAlarmSeq(current int64, alarms []model.OmcAlarm) int64 {
+	var maxAlarmSeq = current
+	for _, v := range alarms {
+		if v.AlarmSeq > maxAlarmSeq {
+			maxAlarmSeq = v.AlarmSeq
+		}
+	}
+	return maxAlarmSeq
+}
+
+// LastAlarmSeq 获取最新的alarm_seq 序号
+func LastAlarmSeq(neType, neId string) int64 {
+	var alarm model.Alarm
+	if err := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ?", neType, neId).Order("alarm_seq desc").First(&alarm).Error; err != nil {
+		zlog.Ins().ErrorF("LastAlarmSeq db error %s", err)
+		return 0
+	}
+	return alarm.AlarmSeq
+}
+
+// List 获取告警信息为文件保存
+func ListForSaveFile(neType, neId string, alarmSeq int64, startTime, endTime, syncSource string) ([]model.Alarm, error) {
+	// 基于OMC数据库的存量活动告警数据文件
+	if syncSource == "0" {
+		// 取告警发生的时间，即告警本身的event_time
+		return alarmByEventTime(neType, neId, startTime, endTime)
+	}
+
+	// 基于告警消息日志的告警消息流水数据文件
+	if syncSource == "1" {
+		if alarmSeq > 0 {
+			// 只适用于syncSource=1的情况 同步告警的起始告警消息序号
+			return alarmByAlarmSeq(neType, neId, alarmSeq)
+		} else {
+			// 取北向接口记录日志流水的时间
+			return alarmByLog(neType, neId, startTime, endTime)
+		}
+	}
+	return []model.Alarm{}, nil
+}
+
+// 取告警发生的时间，即告警本身的event_time
+func alarmByEventTime(neType, neId, startTime, endTime string) ([]model.Alarm, error) {
+	var alarms []model.Alarm
+	if startTime == "" && endTime == "" {
+		return alarms, nil
+	}
+
+	query := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ?", neType, neId)
+
+	// 从当前告警中最早的开始
+	if startTime != "" {
+		startT, err := time.Parse("2006-01-02 15:04:05", startTime)
+		if err != nil {
+			return nil, errors.New("startTime invalid")
+		}
+		query = query.Where("event_time >= ?", startT)
+	}
+	// 取到当前最新的告警截止
+	if endTime != "" {
+		endT, err := time.Parse("2006-01-02 15:04:05", endTime)
+		if err != nil {
+			return nil, errors.New("endTime invalid")
+		}
+		query = query.Where("event_time <= ?", endT)
+	}
+
+	// 查询
+	if err := query.Order("alarm_seq asc").Find(&alarms).Error; err != nil {
+		return nil, err
+	}
+	return alarms, nil
+}
+
+// 同步告警的起始告警消息序号
+func alarmByAlarmSeq(neType, neId string, alarmSeq int64) ([]model.Alarm, error) {
+	var alarms []model.Alarm
+	// 查询
+	query := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ? and alarm_seq >= ?", neType, neId, alarmSeq)
+	if err := query.Order("alarm_seq asc").Find(&alarms).Error; err != nil {
+		return nil, err
+	}
+	return alarms, nil
+}
+
+// 取北向接口记录日志流水的时间
+func alarmByLog(neType, neId, startTime, endTime string) ([]model.Alarm, error) {
+	var alarms []model.Alarm
+	if startTime == "" && endTime == "" {
+		return alarms, nil
+	}
+
+	var aIDs []int64
+	query := db.Client.Model(&model.NbiAlarmLog{}).Select("distinct a_id").Where("ne_type = ? and ne_id = ?", neType, neId)
+
+	// 从当前告警中最早的开始
+	if startTime != "" {
+		startT, err := time.Parse("2006-01-02 15:04:05", startTime)
+		if err != nil {
+			return nil, errors.New("startTime invalid")
+		}
+		query = query.Where("log_time >= ?", startT)
+	}
+	// 取到当前最新的告警截止
+	if endTime != "" {
+		endT, err := time.Parse("2006-01-02 15:04:05", endTime)
+		if err != nil {
+			return nil, errors.New("endTime invalid")
+		}
+		query = query.Where("log_time <= ?", endT)
+	}
+
+	// 查询记录日志流水的id组
+	if err := query.Order("alarm_seq asc").Find(&aIDs).Error; err != nil {
+		return nil, err
+	}
+
+	// 查询告警本身的记录
+	if err := db.Client.Model(&model.Alarm{}).Where("id in (?)", aIDs).Find(&alarms).Error; err != nil {
+		return nil, err
+	}
+
+	return alarms, nil
+}

handle/service/login.go

@@ -0,0 +1,80 @@
+package service
+
+import (
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"omc/core/db"
+	"omc/core/utils"
+	"omc/handle/model"
+	"time"
+
+	"github.com/aceld/zinx/zlog"
+)
+
+// UserLogin 用户登录
+func UserLogin(name, passwd string) error {
+	// 用户名密码校验
+	var user model.User
+	err := db.Client.Model(&model.User{}).Where("account_id=?", name).First(&user).Error
+	if err != nil {
+		zlog.Ins().ErrorF("Password Login[%s]:%s", name, err)
+		return errors.New("username-error")
+	}
+
+	if err := utils.Compare(user.Password, passwd); err != nil {
+		zlog.Ins().ErrorF("Password Login[%s]:%s", name, err)
+		return errors.New("incorrect username and password")
+	}
+	return nil
+}
+
+// CMCALogin 验证随机码
+// content "user:seqNo" 组合的字符
+// signatureBase64 签名数据（base64编码）
+func CMCALogin(name, carCertificate, content, signatureBase64 string) error {
+	// 用户名密码校验
+	var user model.User
+	err := db.Client.Model(&model.User{}).Where("account_id=?", name).First(&user).Error
+	if err != nil {
+		zlog.Ins().ErrorF("Password Login[%s]:%s", name, err)
+		return errors.New("username-error")
+	}
+
+	if user.AccountId != name {
+		return fmt.Errorf("username-error")
+	}
+
+	// 解析证书信息
+	certificate, err := utils.ParseCert(carCertificate)
+	if err != nil {
+		zlog.Ins().ErrorF("ParseCert Login[%s]:%s", name, err)
+		return fmt.Errorf("certificate-error")
+	}
+
+	// 判断证书到期时间
+	if time.Now().After(certificate.NotAfter) {
+		zlog.Ins().ErrorF("certificate Not After Now time Login[%s]:%s", name, err)
+		return errors.New("the certificate has expired")
+	}
+
+	// 判断证书持有人
+	if certificate.Subject.CommonName != name {
+		zlog.Ins().ErrorF("certificate Subject CommonName Login[%s]:%s", name, err)
+		return errors.New("the certificate cn-error")
+	}
+
+	// 解码base64得到签名字节串
+	signature, err := base64.StdEncoding.DecodeString(signatureBase64)
+	if err != nil {
+		zlog.Ins().ErrorF("signatureBase64 Login[%s]:%s", name, err)
+		return fmt.Errorf("the certificate signature-error")
+	}
+
+	// 验证签名
+	valid := utils.VerifySignature(certificate.PublicKey, []byte(content), signature)
+	if valid {
+		return nil
+	}
+	return errors.New("seqNo-error")
+}

nbi_alarm.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/aceld/zinx/zutils/commandline/args"
+
+	"omc/conf"
+	"omc/core/db"
+	"omc/core/manage"
+	"omc/router"
+
+	"os"
+	"os/signal"
+
+	"github.com/aceld/zinx/zlog"
+)
+
+// go run nbi_alarm.go -c conf\nbi_alarm.json
+func main() {
+	// 配置初始化
+	conf.Init(args.Args.ConfigFile)
+	// 初始数据库
+	db.Init()
+
+	//创建服务器句柄
+	for _, cg := range conf.OmcConf.Channel {
+		serverName := fmt.Sprintf("%s:%d", cg.BindFlag, cg.TCPPort)
+		//注册用户管理模块
+		m := manage.NewManager(serverName, cg.BindFlag, cg.Province, cg.DeviceCode)
+
+		//启动服务
+		s := router.LoadServer(cg)
+		go s.Serve()
+
+		//启动实时告警
+		go m.RealTimeAlarm()
+	}
+
+	// close
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt, os.Kill)
+	sig := <-c
+	zlog.Ins().InfoF("===exit=== %s", sig)
+}

nbi_alarm_test.go

@@ -0,0 +1,188 @@
+package main
+
+import (
+	"bytes"
+	"encoding/binary"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/aceld/zinx/zconf"
+)
+
+type Message struct {
+	StartSign uint16
+	MsgType   uint8
+	TimeStamp uint32
+	LenOfBody uint16
+	Value     []byte
+}
+
+type TcpClient struct {
+	conn     net.Conn
+	PID      int32
+	isOnline chan bool
+}
+
+func (s *TcpClient) Unpack(headdata []byte) (head *Message, err error) {
+	fmt.Println("unpack:", hex.EncodeToString(headdata))
+	headBuf := bytes.NewReader(headdata)
+
+	head = &Message{}
+
+	// Read the startSign
+	if err := binary.Read(headBuf, binary.BigEndian, &head.StartSign); err != nil {
+		return nil, err
+	}
+
+	// Read the msgID
+	if err := binary.Read(headBuf, binary.BigEndian, &head.MsgType); err != nil {
+		return nil, err
+	}
+
+	// read timeStamp
+	if err := binary.Read(headBuf, binary.BigEndian, &head.TimeStamp); err != nil {
+		return nil, err
+	}
+
+	// Read the data length
+	if err := binary.Read(headBuf, binary.BigEndian, &head.LenOfBody); err != nil {
+		return nil, err
+	}
+
+	// Check whether the data length exceeds the maximum allowed packet size
+	// (判断dataLen的长度是否超出我们允许的最大包长度)
+	if zconf.GlobalObject.MaxPacketSize > 0 && uint32(head.LenOfBody) > zconf.GlobalObject.MaxPacketSize {
+		return nil, errors.New("too large msg data received")
+	}
+
+	return head, nil
+}
+
+func (s *TcpClient) Pack(msgID uint32, dataBytes []byte) (out []byte, err error) {
+	outbuff := bytes.NewBuffer([]byte{})
+
+	// Write the oxffff
+	if err := binary.Write(outbuff, binary.BigEndian, uint16(0xffff)); err != nil {
+		return nil, err
+	}
+	//Write the type
+	if err := binary.Write(outbuff, binary.BigEndian, uint8(msgID)); err != nil {
+		return nil, err
+	}
+
+	//Write the timestamp
+	if err := binary.Write(outbuff, binary.BigEndian, uint32(0x999999)); err != nil {
+		return nil, err
+	}
+
+	//Write the length
+	if err := binary.Write(outbuff, binary.BigEndian, uint16(len(dataBytes))); err != nil {
+		return nil, err
+	}
+
+	// Write the data
+	if err := binary.Write(outbuff, binary.BigEndian, dataBytes); err != nil {
+		return nil, err
+	}
+
+	out = outbuff.Bytes()
+	return
+}
+
+func (s *TcpClient) SendMsg(msgID uint32, data []byte) {
+	sendData, err := s.Pack(msgID, data)
+	if err == nil {
+		_, err := s.conn.Write(sendData)
+		fmt.Println("send msg:", hex.EncodeToString(sendData), " err:", err)
+	} else {
+		fmt.Println(err)
+	}
+	return
+}
+
+func (s *TcpClient) Receive() {
+	for {
+		//读取服务端发来的数据 ==》 SyncPID
+		//1.读取8字节
+		////第一次读取，读取数据头
+		headData := make([]byte, 9)
+		if _, err := io.ReadFull(s.conn, headData); err != nil {
+			fmt.Println(err)
+			return
+		}
+		pkgHead, err := s.Unpack(headData)
+		if err != nil {
+			fmt.Println("Unpack", err)
+			return
+		}
+		//data
+		if pkgHead.LenOfBody > 0 {
+			pkgHead.Value = make([]byte, pkgHead.LenOfBody)
+			if _, err := io.ReadFull(s.conn, pkgHead.Value); err != nil {
+				return
+			}
+		}
+
+		//处理服务器回执业务
+		fmt.Println("=================================")
+		fmt.Println("StartSign:", pkgHead.StartSign)
+		fmt.Println("MsgType:", pkgHead.MsgType)
+		fmt.Println("TimeStamp:", pkgHead.TimeStamp)
+		fmt.Println("LenOfBody:", pkgHead.LenOfBody)
+		fmt.Println("Value:", string(pkgHead.Value))
+		fmt.Println("=================================")
+		time.Sleep(10 * time.Microsecond)
+	}
+}
+
+func (s *TcpClient) Start() {
+	//登录
+	// data := "reqLoginAlarm;user=audit;key=omc@password;type=ftp"
+	// s.SendMsg(1, []byte(data))
+
+	//发送同步告警信息
+	//data = "reqSyncAlarmMsg;reqId=33;alarmSeq=1"
+	//this.SendMsg(omc.ReqSyncAlarmMsg, []byte(data))
+
+	//发送文件同步告警
+	//data = "reqSyncAlarmFile;reqId=35;alarmSeq=2000;syncSource=1"
+	//data = "reqSyncAlarmFile;reqId=33;startTime=2023-01-08 00:00:00;syncSource=0"
+	// data = "reqSyncAlarmFile;reqId=34;startTime=2023-01-08 16:07:00;endTime=2023-07-19 23:59:59;syncSource=1"
+
+	// 心跳
+	data := "reqHeartBeat;reqId=33"
+	s.SendMsg(8, []byte(data))
+	go s.Receive()
+
+}
+
+func NewTcpClient(ip string, port int) *TcpClient {
+	addrStr := fmt.Sprintf("%s:%d", ip, port)
+	conn, err := net.Dial("tcp", addrStr)
+	if err != nil {
+		fmt.Println("net.Dial err: ", err)
+		panic(err)
+	}
+
+	client := &TcpClient{
+		conn:     conn,
+		PID:      0,
+		isOnline: make(chan bool),
+	}
+
+	fmt.Printf("conn: %+v. Connected to server...", conn)
+
+	return client
+}
+
+func TestFile(t *testing.T) {
+	client := NewTcpClient("127.0.0.1", 31232)
+	client.Start()
+	//防止进程退出，等待中断信号
+	select {}
+}

readme.md

@@ -0,0 +1,43 @@
+
+项目交付文件：
+fpt://192.168.0.229/home/guodeng/bin
+
+服务启动：
+./nb_alarm_agent  -c /home/guodeng/omc/conf/nb_alarm_agent.json
+
+配置文件详情：
+{
+  /* 通道配置项*/
+  "channel": [
+    {
+      "tcp_port": 31232,    //通道TCP监听端口
+      "bind_flag": "SMF#SZ_01",  //通道Bind网元（格式为 ne_type#ne_id）
+      "province": "BJ",    //网元所在省份
+      "device_code": "0001"   //网元主机编码
+    },
+    {
+      "tcp_port": 31233,  //通道TCP监听端口
+      "bind_flag": "UDM#SZ_03", //通道Bind网元（格式为 ne_type#ne_id）
+      "province": "BJ",    //网元所在省份
+      "device_code": "0002"  //网元主机编码
+    }
+  ],
+
+  // 数据库配置
+  "mysql": "root:1000omc@kp!@tcp(192.168.0.229:33066)/omc_db?charset=utf8mb4&parseTime=True&loc=Local",
+
+  // FTP服务器根目录
+  "ftp_root": "ftp:192.168.0.229/data/ftp",
+
+  //服务名称
+  "mame":"north agent",
+
+  // 心跳保活时间， 如果服务器检查client 在超过心跳时间没有数据（心跳数据或者业务数据）发送，则断开连接
+  "heartbeat_max": 180,
+
+  // 日志存放目录
+  "log_dir": "./omc_log",
+
+  // 日志文件名称
+  "log_file":"omc.log"
+}

router/router.go

@@ -0,0 +1,65 @@
+package router
+
+import (
+	"fmt"
+	"omc/conf"
+	"omc/core/decoder"
+	"omc/core/dpack"
+	heartbeat "omc/core/heart_beat"
+	"omc/core/manage"
+	"omc/handle/api"
+	"time"
+
+	"github.com/aceld/zinx/zconf"
+	"github.com/aceld/zinx/ziface"
+	"github.com/aceld/zinx/znet"
+)
+
+func LoadServer(cg conf.Channel) ziface.IServer {
+	serverName := fmt.Sprintf("%s:%d", cg.BindFlag, cg.TCPPort)
+
+	//new 一个TCP服务
+	s := znet.NewUserConfServer(&zconf.Config{
+		TCPPort:        cg.TCPPort,
+		Name:           serverName,
+		Host:           "0.0.0.0",
+		MaxConn:        conf.OmcConf.MaxConn,
+		WorkerPoolSize: uint32(conf.OmcConf.WorkerPoolSize),
+		HeartbeatMax:   conf.OmcConf.HeartbeatMax,
+		LogDir:         conf.OmcConf.LogDir,
+		LogFile:        conf.OmcConf.LogFile,
+	})
+
+	//注册客户端连接建立和丢失函数
+	s.SetOnConnStart(manage.OnConnectionAdd)
+	s.SetOnConnStop(manage.OnConnectionLost)
+
+	//添加LTV数据格式Decoder
+	s.SetDecoder(decoder.NewOmcDecoder())
+	//添加LTV数据格式的Pack封包Encoder
+	s.SetPacket(dpack.NewDataPack())
+
+	// (启动心跳检测)
+	s.StartHeartBeatWithOption(60*time.Second, &ziface.HeartBeatOption{
+		MakeMsg:          heartbeat.MyHeartBeatMsg,
+		OnRemoteNotAlive: heartbeat.MyOnRemoteNotAlive,
+		Router:           &api.ReqHeartBeat{},
+		HeadBeatMsgID:    api.ReqHeartBeatMsgID,
+	})
+	// 设置默认的心跳发送函数
+	heart := s.GetHeartBeat()
+	heart.SetHeartbeatFunc(heartbeat.MyHeartBeat)
+
+	// 添加自定义拦截器 debug调试输出信息用
+	// s.AddInterceptor(&interceptor.GlobalInterceptor{})
+
+	//注册路由 ================
+	s.AddRouter(api.ReqLoginAlarmMsgID, &api.ReqLoginAlarm{})
+	s.AddRouter(api.ReqCMCALoginSeqMsgID, &api.ReqCMCALoginSeq{})
+	s.AddRouter(api.ReqCMCALoginAlarmMsgID, &api.ReqCMCALoginAlarm{})
+	s.AddRouter(api.ReqSyncAlarmMsgID, &api.ReqSyncAlarmMsg{})
+	s.AddRouter(api.ReqSyncAlarmFileID, &api.ReqSyncAlarmFile{})
+	s.AddRouter(api.CloseConnAlarmMsgID, &api.CloseConnAlarm{})
+
+	return s
+}