remove unused
This commit is contained in:
parent
535f565519
commit
e6e70113b3
|
|
@ -12,8 +12,6 @@ type config struct {
|
|||
postgresConf map[string]*postgresConfig
|
||||
influxConf map[string]*influxConfig
|
||||
redisConf map[string]*redisConfig
|
||||
mongoConf map[string]*mongoConfig
|
||||
rabbitConf map[string][]*rabbitConfig
|
||||
cl104Conf map[string]map[string][]int
|
||||
}
|
||||
|
||||
|
|
@ -46,14 +44,6 @@ func init() {
|
|||
redisConf := confDir + string(os.PathSeparator) + redisConfigName()
|
||||
conf.unmarshalJsonFile(redisConf, &conf.redisConf)
|
||||
|
||||
conf.mongoConf = make(map[string]*mongoConfig)
|
||||
mongoConf := confDir + string(os.PathSeparator) + mongoConfigName()
|
||||
conf.unmarshalJsonFile(mongoConf, &conf.mongoConf)
|
||||
|
||||
conf.rabbitConf = make(map[string][]*rabbitConfig)
|
||||
rabbitConf := confDir + string(os.PathSeparator) + rabbitConfigName()
|
||||
conf.unmarshalJsonFile(rabbitConf, &conf.rabbitConf)
|
||||
|
||||
conf.cl104Conf = make(map[string]map[string][]int)
|
||||
cl104Conf := confDir + string(os.PathSeparator) + cl104ConfigName()
|
||||
conf.unmarshalJsonFile(cl104Conf, &conf.cl104Conf)
|
||||
|
|
@ -98,20 +88,6 @@ func (c *config) RedisConf(tag string) *redisConfig {
|
|||
return c.redisConf[tag]
|
||||
}
|
||||
|
||||
func (c *config) MongoConf(tag string) *mongoConfig {
|
||||
if c == nil || c.mongoConf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
return c.mongoConf[tag]
|
||||
}
|
||||
|
||||
func (c *config) RabbitConf(tag string) []*rabbitConfig {
|
||||
if c == nil || c.rabbitConf == nil {
|
||||
panic("rabbit config is nil")
|
||||
}
|
||||
return c.rabbitConf[tag]
|
||||
}
|
||||
|
||||
func (c *config) CL104ConfAll() map[string]map[string][]int {
|
||||
if c == nil {
|
||||
panic("config is nil")
|
||||
|
|
|
|||
102
config/mongo.go
102
config/mongo.go
|
|
@ -1,102 +0,0 @@
|
|||
package config
|
||||
|
||||
type mongoConfig struct {
|
||||
Addrs []string `json:"addrs" yaml:"addrs"`
|
||||
Username string `json:"username" yaml:"username"`
|
||||
Password string `json:"password" yaml:"password"`
|
||||
AuthSource string `josn:"authsource" yaml:"authsource"`
|
||||
AuthMechanism string `json:"authmechanism" yaml:"authmechanism"`
|
||||
}
|
||||
|
||||
func NewMongoConfig() *mongoConfig {
|
||||
return new(mongoConfig)
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) GetAddrs() []string {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
|
||||
return conf.Addrs
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) SetAddrs(addrs []string) *mongoConfig {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
conf.Addrs = addrs
|
||||
|
||||
return conf
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) GetUsername() string {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
|
||||
return conf.Username
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) SetUsername(username string) *mongoConfig {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
conf.Username = username
|
||||
|
||||
return conf
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) GetPassword() string {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
|
||||
return conf.Password
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) SetPassword(password string) *mongoConfig {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
conf.Password = password
|
||||
|
||||
return conf
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) GetAuthSource() string {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
|
||||
return conf.AuthSource
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) SetAuthSource(authSource string) *mongoConfig {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
conf.AuthSource = authSource
|
||||
|
||||
return conf
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) GetAuthMechanism() string {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
|
||||
return conf.AuthMechanism
|
||||
}
|
||||
|
||||
func (conf *mongoConfig) SetAuthMechanism(authMechanism string) *mongoConfig {
|
||||
if conf == nil {
|
||||
panic("mongo config is nil")
|
||||
}
|
||||
conf.AuthMechanism = authMechanism
|
||||
|
||||
return conf
|
||||
}
|
||||
|
||||
func mongoConfigName() string {
|
||||
return "mongo.json"
|
||||
}
|
||||
|
|
@ -1,80 +0,0 @@
|
|||
package config
|
||||
|
||||
type rabbitConfig struct {
|
||||
Broker string `json:"broker" yaml:"broker"`
|
||||
Username string `json:"username" yaml:"username"`
|
||||
Password string `json:"password" yaml:"password"`
|
||||
}
|
||||
|
||||
func NewRabbitConfig() *rabbitConfig {
|
||||
return new(rabbitConfig)
|
||||
}
|
||||
|
||||
func (conf *rabbitConfig) GenAddress(tls bool) string {
|
||||
if conf == nil {
|
||||
panic("rabbit config is nil")
|
||||
}
|
||||
|
||||
address := "amqp://"
|
||||
if conf.GetUsername() != "" && conf.GetPassword() != "" {
|
||||
address += conf.GetUsername() + ":" + conf.GetPassword() + "@"
|
||||
}
|
||||
address += conf.GetBroker() + "/"
|
||||
|
||||
return address
|
||||
}
|
||||
|
||||
func (conf *rabbitConfig) GetBroker() string {
|
||||
if conf == nil {
|
||||
panic("rabbit config is nil")
|
||||
}
|
||||
|
||||
return conf.Broker
|
||||
}
|
||||
|
||||
func (conf *rabbitConfig) SetBroker(broker string) *rabbitConfig {
|
||||
if conf == nil {
|
||||
panic("rabbit config is nil")
|
||||
}
|
||||
conf.Broker = broker
|
||||
|
||||
return conf
|
||||
}
|
||||
|
||||
func (conf *rabbitConfig) GetUsername() string {
|
||||
if conf == nil {
|
||||
panic("rabbit config is nil")
|
||||
}
|
||||
|
||||
return conf.Username
|
||||
}
|
||||
|
||||
func (conf *rabbitConfig) SetUsername(username string) *rabbitConfig {
|
||||
if conf == nil {
|
||||
panic("rabbit config is nil")
|
||||
}
|
||||
conf.Username = username
|
||||
|
||||
return conf
|
||||
}
|
||||
|
||||
func (conf *rabbitConfig) GetPassword() string {
|
||||
if conf == nil {
|
||||
panic("rabbit config is nil")
|
||||
}
|
||||
|
||||
return conf.Password
|
||||
}
|
||||
|
||||
func (conf *rabbitConfig) SetPassword(password string) *rabbitConfig {
|
||||
if conf == nil {
|
||||
panic("rabbit config is nil")
|
||||
}
|
||||
conf.Password = password
|
||||
|
||||
return conf
|
||||
}
|
||||
|
||||
func rabbitConfigName() string {
|
||||
return "rabbit.json"
|
||||
}
|
||||
|
|
@ -1,5 +1,2 @@
|
|||
{
|
||||
"station000":{
|
||||
"127.0.0.1:8899":[0]
|
||||
}
|
||||
}
|
||||
|
|
@ -1,9 +0,0 @@
|
|||
{
|
||||
"default":{
|
||||
"addrs":["127.0.0.1:27017"],
|
||||
"username":"admin",
|
||||
"password":"password",
|
||||
"authsource":"admin",
|
||||
"authmechanism":"SCRAM-SHA-256"
|
||||
}
|
||||
}
|
||||
|
|
@ -1,9 +0,0 @@
|
|||
{
|
||||
"default": [
|
||||
{
|
||||
"broker": "127.0.0.1:5672",
|
||||
"username": "rabbitmq",
|
||||
"password": "password"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -4,9 +4,7 @@ import (
|
|||
"context"
|
||||
"datart/data/cl104"
|
||||
"datart/data/influx"
|
||||
"datart/data/mongo"
|
||||
"datart/data/postgres"
|
||||
"datart/data/rabbit"
|
||||
"datart/data/redis"
|
||||
"datart/log"
|
||||
)
|
||||
|
|
@ -36,11 +34,7 @@ func (p *Processes) StartDataProcessing() {
|
|||
func (p *Processes) Cancel(ctx context.Context) {
|
||||
p.cancel()
|
||||
|
||||
eventNotifyPublisher.Close(ctx)
|
||||
|
||||
influx.CloseDefault()
|
||||
mongo.CloseDefault(ctx)
|
||||
postgres.CloseDefault()
|
||||
rabbit.CloseDefault(ctx)
|
||||
redis.CloseDefault()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,131 +0,0 @@
|
|||
package mongo
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
const (
|
||||
_ = iota
|
||||
almCodeCommmExcept
|
||||
almCodeADFault
|
||||
almCodePPSExcept
|
||||
almCodeReserve1
|
||||
almCodeUnitInit
|
||||
almCodeReadParamErr
|
||||
almCodeReserve2
|
||||
almCodeStartSample
|
||||
almCodeOverSample
|
||||
almCodeUnderSample
|
||||
)
|
||||
|
||||
type Alarm struct {
|
||||
DriverName string `bson:"driver_name" json:"driver_name"`
|
||||
DeviceNo string `bson:"device_no" json:"device_no"`
|
||||
AlarmCode int `bson:"alarm_code" json:"alarm_code"`
|
||||
AlarmTime int64 `bson:"alarm_time" json:"alarm_time"` // ms
|
||||
AlarmStatus int `bson:"alarm_status" json:"alarm_status"` // 0 "复位", 1 "动作/产生/告警"
|
||||
}
|
||||
|
||||
var almCode2Name = []string{
|
||||
almCodeCommmExcept: "通信异常",
|
||||
almCodeADFault: "AD故障",
|
||||
almCodePPSExcept: "同步秒脉冲异常",
|
||||
almCodeReserve1: "备用",
|
||||
almCodeUnitInit: "单元初始化",
|
||||
almCodeReadParamErr: "读参数错",
|
||||
almCodeReserve2: "备用",
|
||||
almCodeStartSample: "启动采样-内部转换信号",
|
||||
almCodeOverSample: "秒内采样点数过量",
|
||||
almCodeUnderSample: "秒内采样点数欠量",
|
||||
}
|
||||
|
||||
func (a *Alarm) GetName() string {
|
||||
return almCode2Name[a.AlarmCode]
|
||||
}
|
||||
|
||||
func (a *Alarm) GetType() int {
|
||||
switch a.AlarmCode {
|
||||
case almCodeReserve1,
|
||||
almCodeReserve2,
|
||||
almCodeUnitInit,
|
||||
almCodeStartSample:
|
||||
return genEventType(0, 0)
|
||||
|
||||
case almCodeOverSample,
|
||||
almCodeUnderSample:
|
||||
return genEventType(0, 1)
|
||||
|
||||
case almCodeCommmExcept,
|
||||
almCodeADFault,
|
||||
almCodePPSExcept,
|
||||
almCodeReadParamErr:
|
||||
return genEventType(0, 2)
|
||||
}
|
||||
|
||||
return -1
|
||||
}
|
||||
|
||||
func (a *Alarm) GetPriority() int {
|
||||
switch a.AlarmCode {
|
||||
case almCodeReserve1,
|
||||
almCodeReserve2,
|
||||
almCodeUnitInit,
|
||||
almCodeStartSample:
|
||||
return 1
|
||||
|
||||
case almCodeOverSample,
|
||||
almCodeUnderSample:
|
||||
return 4
|
||||
|
||||
case almCodeCommmExcept,
|
||||
almCodeADFault,
|
||||
almCodePPSExcept,
|
||||
almCodeReadParamErr:
|
||||
return 7
|
||||
}
|
||||
|
||||
return -1
|
||||
}
|
||||
|
||||
func GenEvent(alarm []byte, ip string) (*Event, error) {
|
||||
a, err := UnmarshallToAlarm(alarm)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return a.ConvertToEvent(ip)
|
||||
}
|
||||
|
||||
func (a *Alarm) ConvertToEvent(ip string) (*Event, error) {
|
||||
e := new(Event)
|
||||
uid, err := uuid.NewV7()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if a != nil {
|
||||
e.Event = a.GetName()
|
||||
e.EventUUID = uid.String()
|
||||
e.Type = a.GetType()
|
||||
e.Priority = a.GetPriority()
|
||||
e.Status = EventStatusHappen
|
||||
e.Timestamp = a.AlarmTime
|
||||
e.From = "station"
|
||||
e.Operations = append(e.Operations, &operation{
|
||||
Action: EventActionHappen,
|
||||
OP: ip,
|
||||
TS: a.AlarmTime,
|
||||
})
|
||||
e.Origin = a
|
||||
}
|
||||
|
||||
return e, nil
|
||||
}
|
||||
|
||||
func UnmarshallToAlarm(data []byte) (*Alarm, error) {
|
||||
alm := new(Alarm)
|
||||
err := json.Unmarshal(data, alm)
|
||||
return alm, err
|
||||
}
|
||||
|
|
@ -1,202 +0,0 @@
|
|||
package mongo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
)
|
||||
|
||||
const (
|
||||
dbevent string = "cl"
|
||||
tbevent string = "events"
|
||||
)
|
||||
|
||||
const (
|
||||
EventStatusHappen = iota
|
||||
EventStatusDataAt
|
||||
EventStatusReport
|
||||
EventStatusConfirm
|
||||
EventStatusPersist
|
||||
EventStatusClose
|
||||
)
|
||||
|
||||
const (
|
||||
EventActionHappen = "happen"
|
||||
EventActionDataAt = "data_attach"
|
||||
EventActionReport = "report"
|
||||
EventActionConfirm = "confirm"
|
||||
EventActionPersist = "persist"
|
||||
EventActionClose = "close"
|
||||
)
|
||||
|
||||
var EventStatusAction = []string{
|
||||
EventStatusHappen: EventActionHappen,
|
||||
EventStatusDataAt: EventActionDataAt,
|
||||
EventStatusReport: EventActionReport,
|
||||
EventStatusConfirm: EventActionConfirm,
|
||||
EventStatusPersist: EventActionPersist,
|
||||
EventStatusClose: EventActionClose,
|
||||
}
|
||||
|
||||
type operation struct {
|
||||
Action string `bson:"action" json:"action"`
|
||||
OP string `bson:"op" json:"op"`
|
||||
TS int64 `bson:"ts" json:"ts"`
|
||||
}
|
||||
|
||||
func GenOperation(action, op string) operation {
|
||||
return operation{
|
||||
Action: action,
|
||||
OP: op,
|
||||
TS: time.Now().UnixMilli(),
|
||||
}
|
||||
}
|
||||
|
||||
type Event struct {
|
||||
Event string `bson:"event" json:"event"`
|
||||
EventUUID string `bson:"event_uuid" json:"event_uuid"`
|
||||
Type int `bson:"type" json:"type"`
|
||||
Priority int `bson:"priority" json:"priority"` // 0~9
|
||||
Status int `bson:"status" json:"status"`
|
||||
Timestamp int64 `bson:"timestamp" json:"timestamp"`
|
||||
From string `bson:"from" json:"from"`
|
||||
Operations []*operation `bson:"operations" json:"operations"`
|
||||
// others
|
||||
Origin *Alarm `bson:"origin" json:"origin"`
|
||||
}
|
||||
|
||||
func (e *Event) Marshall() ([]byte, error) {
|
||||
return json.Marshal(e)
|
||||
}
|
||||
|
||||
func InsertEvent(ctx context.Context, doc any) error {
|
||||
_, err := getCollection(dbevent, tbevent).InsertOne(ctx, doc)
|
||||
return err
|
||||
}
|
||||
|
||||
func DeleteEvent[T bson.M | bson.D](ctx context.Context, filter T) error {
|
||||
_, err := getCollection(dbevent, tbevent).DeleteOne(ctx, filter)
|
||||
return err
|
||||
}
|
||||
|
||||
func UpdateEvent[T bson.M | bson.D](ctx context.Context, filter T, update T) error {
|
||||
_, err := getCollection(dbevent, tbevent).UpdateOne(ctx, filter, update)
|
||||
return err
|
||||
}
|
||||
|
||||
func FindEventsWithPageLimit[T bson.M | bson.D](ctx context.Context, filter T,
|
||||
sort int, page int64, limit int64) ([]*Event, error) {
|
||||
|
||||
opt := options.Find()
|
||||
if sort == 1 || sort == -1 {
|
||||
opt.SetSort(bson.D{{Key: "timestamp", Value: sort}})
|
||||
} else {
|
||||
opt.SetSort(bson.D{{Key: "_id", Value: 1}})
|
||||
}
|
||||
|
||||
if page > 0 && limit > 0 {
|
||||
opt.SetSkip(limit * (page - 1)).SetLimit(limit)
|
||||
}
|
||||
|
||||
cursor, err := getCollection(dbevent, tbevent).Find(ctx, filter, opt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer cursor.Close(ctx)
|
||||
|
||||
var docs []*Event
|
||||
for cursor.Next(ctx) {
|
||||
doc := new(Event)
|
||||
if err := cursor.Decode(doc); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
docs = append(docs, doc)
|
||||
}
|
||||
if err := cursor.Err(); err != nil {
|
||||
return docs, err
|
||||
}
|
||||
|
||||
return docs, nil
|
||||
}
|
||||
|
||||
func BulkWriteEventsWithUUID(ctx context.Context, curd byte, events []map[string]any) ([]string, error) {
|
||||
length := len(events)
|
||||
if length <= 0 {
|
||||
return nil, errors.New("no event")
|
||||
}
|
||||
|
||||
models := make([]mongo.WriteModel, 0, length)
|
||||
idx2UUID := make(map[int]string, length)
|
||||
for i, event := range events {
|
||||
uuid, ok := event["event_uuid"].(string)
|
||||
if !ok || uuid == "" {
|
||||
return nil, fmt.Errorf("invalid uuid at index %d", i)
|
||||
}
|
||||
|
||||
switch curd {
|
||||
case 'c':
|
||||
models = append(models,
|
||||
mongo.NewInsertOneModel().
|
||||
SetDocument(event))
|
||||
|
||||
case 'u':
|
||||
filter := bson.M{"event_uuid": uuid}
|
||||
status, ok := event["status"].(int)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("invalid status at index %d", i)
|
||||
}
|
||||
operation, ok := event["operation"].(operation)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("invalid operation at index %d", i)
|
||||
}
|
||||
update := bson.M{"$set": bson.M{"status": status},
|
||||
"$push": bson.M{"operations": operation}}
|
||||
models = append(models,
|
||||
mongo.NewUpdateOneModel().
|
||||
SetFilter(filter).
|
||||
SetUpdate(update))
|
||||
|
||||
case 'd':
|
||||
filter := bson.M{"event_uuid": uuid}
|
||||
models = append(models,
|
||||
mongo.NewDeleteOneModel().
|
||||
SetFilter(filter))
|
||||
|
||||
default:
|
||||
return nil, fmt.Errorf("invalid curd")
|
||||
}
|
||||
|
||||
idx2UUID[i] = uuid
|
||||
}
|
||||
|
||||
opts := options.BulkWrite().SetOrdered(false)
|
||||
_, err := getCollection(dbevent, tbevent).BulkWrite(ctx, models, opts)
|
||||
|
||||
sUUIDs := []string{}
|
||||
if err != nil {
|
||||
if bulkErr, ok := err.(mongo.BulkWriteException); ok {
|
||||
idxExist := map[int]bool{}
|
||||
for _, writeErr := range bulkErr.WriteErrors {
|
||||
idxExist[writeErr.Index] = true
|
||||
}
|
||||
|
||||
for idx, uuid := range idx2UUID {
|
||||
if !idxExist[idx] {
|
||||
sUUIDs = append(sUUIDs, uuid)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return sUUIDs, err
|
||||
}
|
||||
|
||||
func genEventType(sys int, level int) int {
|
||||
return sys + level*3
|
||||
}
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
package mongo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"datart/config"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
)
|
||||
|
||||
var client *mongo.Client
|
||||
|
||||
func init() {
|
||||
conf := config.Conf().MongoConf("default")
|
||||
uri := "mongodb://" + strings.Join(conf.GetAddrs(), ",")
|
||||
cliOpts := options.Client().
|
||||
ApplyURI(uri).SetTimeout(1 * time.Second).
|
||||
SetAuth(options.Credential{
|
||||
AuthMechanism: conf.GetAuthMechanism(),
|
||||
AuthSource: conf.GetAuthSource(),
|
||||
Username: conf.GetUsername(),
|
||||
Password: conf.GetPassword(),
|
||||
})
|
||||
|
||||
cli, err := mongo.Connect(cliOpts)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
client = cli
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
|
||||
defer cancel()
|
||||
if err := client.Ping(ctx, nil); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
func NewMongoClient(opts ...*options.ClientOptions) (*mongo.Client, error) {
|
||||
return mongo.Connect(opts...)
|
||||
}
|
||||
|
||||
func CloseDefault(ctx context.Context) error {
|
||||
return client.Disconnect(ctx)
|
||||
}
|
||||
|
||||
func GetSession() (*mongo.Session, error) {
|
||||
return client.StartSession()
|
||||
}
|
||||
|
||||
func getCollection(db string, tb string) *mongo.Collection {
|
||||
return client.Database(db).Collection(tb)
|
||||
}
|
||||
|
|
@ -1,73 +0,0 @@
|
|||
package data
|
||||
|
||||
import (
|
||||
"context"
|
||||
"datart/data/rabbit"
|
||||
"datart/log"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
|
||||
)
|
||||
|
||||
var eventNotifyXQK = rabbit.XQK{
|
||||
Exchange: "event_notify_fanout",
|
||||
}
|
||||
|
||||
var eventNotifyPublisher *rmq.Publisher
|
||||
|
||||
func init() {
|
||||
ctx := context.Background()
|
||||
|
||||
m := new(rabbit.Manage)
|
||||
|
||||
rm := rabbit.DefaultManagement()
|
||||
|
||||
rx := &rmq.FanOutExchangeSpecification{
|
||||
Name: eventNotifyXQK.Exchange,
|
||||
}
|
||||
|
||||
m.Init(ctx, rm, rx, nil, nil)
|
||||
|
||||
if err := m.DeclareExchange(ctx); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
publisher, err := rabbit.NewPublisher(ctx, "default", &eventNotifyXQK)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
eventNotifyPublisher = publisher
|
||||
}
|
||||
|
||||
func PublishEvent(ctx context.Context, event any) error {
|
||||
data, err := json.Marshal(event)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
result, err := eventNotifyPublisher.Publish(ctx, rmq.NewMessage(data))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch result.Outcome.(type) {
|
||||
case *rmq.StateAccepted:
|
||||
// "message accepted"
|
||||
case *rmq.StateReleased:
|
||||
// "message released"
|
||||
case *rmq.StateRejected:
|
||||
return errors.New("message rejected")
|
||||
default:
|
||||
return fmt.Errorf("invalid message state: %v", result.Outcome)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func CloseEventPublisher(ctx context.Context) {
|
||||
if err := eventNotifyPublisher.Close(ctx); err != nil {
|
||||
log.Error(err)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,50 +0,0 @@
|
|||
package rabbit
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
|
||||
)
|
||||
|
||||
type rabbitClient struct {
|
||||
env *rmq.Environment
|
||||
conn *rmq.AmqpConnection
|
||||
}
|
||||
|
||||
func NewClient(ctx context.Context, tag string) (*rabbitClient, error) {
|
||||
|
||||
endpoints, err := genEndpoints(tag)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cli := new(rabbitClient)
|
||||
cli.env = rmq.NewClusterEnvironment(endpoints)
|
||||
conn, err := cli.env.NewConnection(context.Background())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
client.conn = conn
|
||||
|
||||
return cli, nil
|
||||
}
|
||||
|
||||
func (c *rabbitClient) Management() *rmq.AmqpManagement {
|
||||
return c.conn.Management()
|
||||
}
|
||||
|
||||
func (c *rabbitClient) NewPublisher(ctx context.Context, destination rmq.ITargetAddress,
|
||||
options rmq.IConsumerOptions) (*rmq.Publisher, error) {
|
||||
|
||||
return c.conn.NewPublisher(ctx, destination, options)
|
||||
}
|
||||
|
||||
func (c *rabbitClient) NewConsumer(ctx context.Context, queueName string,
|
||||
options rmq.IConsumerOptions) (*rmq.Consumer, error) {
|
||||
|
||||
return c.conn.NewConsumer(ctx, queueName, options)
|
||||
}
|
||||
|
||||
func (c *rabbitClient) Close(ctx context.Context) error {
|
||||
return c.env.CloseConnections(ctx)
|
||||
}
|
||||
|
|
@ -1,45 +0,0 @@
|
|||
package rabbit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
|
||||
)
|
||||
|
||||
func NewConsumer(ctx context.Context, tag string, xqk *XQK) (*rmq.Consumer, error) {
|
||||
cli := client
|
||||
if tag != "default" {
|
||||
var err error
|
||||
cli, err = NewClient(ctx, tag)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return cli.conn.NewConsumer(ctx, xqk.Queue, nil)
|
||||
}
|
||||
|
||||
func Consuming(ctx context.Context, consumer *rmq.Consumer, msgChan chan<- []byte) {
|
||||
for {
|
||||
deliCtx, err := consumer.Receive(ctx)
|
||||
if errors.Is(err, context.Canceled) {
|
||||
// The consumer was closed correctly
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
// An error occurred receiving the message
|
||||
return
|
||||
}
|
||||
|
||||
for _, data := range deliCtx.Message().Data {
|
||||
msgChan <- data
|
||||
}
|
||||
|
||||
err = deliCtx.Accept(ctx)
|
||||
if err != nil {
|
||||
// accept error
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,83 +0,0 @@
|
|||
package rabbit
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
|
||||
)
|
||||
|
||||
type Manage struct {
|
||||
m *rmq.AmqpManagement
|
||||
xName string
|
||||
x rmq.IExchangeSpecification
|
||||
qName string
|
||||
q rmq.IQueueSpecification
|
||||
bPath string
|
||||
b rmq.IBindingSpecification
|
||||
}
|
||||
|
||||
func (m *Manage) Init(ctx context.Context, rm *rmq.AmqpManagement,
|
||||
rx rmq.IExchangeSpecification, rq rmq.IQueueSpecification,
|
||||
rb rmq.IBindingSpecification) {
|
||||
|
||||
m.m = rm
|
||||
m.x = rx
|
||||
m.q = rq
|
||||
m.b = rb
|
||||
}
|
||||
|
||||
func (m *Manage) DeclareExchange(ctx context.Context) error {
|
||||
xinfo, err := m.m.DeclareExchange(ctx, m.x)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.xName = xinfo.Name()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manage) DeclareAndBind(ctx context.Context) error {
|
||||
xinfo, err := m.m.DeclareExchange(ctx, m.x)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.xName = xinfo.Name()
|
||||
|
||||
qinfo, err := m.m.DeclareQueue(ctx, m.q)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.qName = qinfo.Name()
|
||||
|
||||
bPath, err := m.m.Bind(ctx, m.b)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.bPath = bPath
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manage) UnbindAndDelete(ctx context.Context) (purged int, err error) {
|
||||
err = m.m.Unbind(ctx, m.bPath)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = m.m.DeleteExchange(ctx, m.xName)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
purged, err = m.m.PurgeQueue(ctx, m.qName)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = m.m.DeleteQueue(ctx, m.qName)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
|
@ -1,52 +0,0 @@
|
|||
package rabbit
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
|
||||
)
|
||||
|
||||
func NewPublisher(ctx context.Context, tag string, xqk *XQK) (*rmq.Publisher, error) {
|
||||
cli := client
|
||||
if tag != "default" {
|
||||
var err error
|
||||
cli, err = NewClient(ctx, tag)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return cli.conn.NewPublisher(context.Background(), &rmq.ExchangeAddress{
|
||||
Exchange: xqk.Exchange,
|
||||
Key: xqk.Key,
|
||||
}, nil)
|
||||
}
|
||||
|
||||
func Publishing(ctx context.Context, publisher *rmq.Publisher, msgChan <-chan []byte) {
|
||||
for {
|
||||
select {
|
||||
case msg := <-msgChan:
|
||||
result, err := publisher.Publish(ctx, rmq.NewMessage(msg))
|
||||
if err != nil {
|
||||
_ = err // publish error
|
||||
continue
|
||||
}
|
||||
|
||||
switch result.Outcome.(type) {
|
||||
case *rmq.StateAccepted:
|
||||
// "message accepted"
|
||||
|
||||
case *rmq.StateReleased:
|
||||
// "message not routed"
|
||||
|
||||
case *rmq.StateRejected:
|
||||
// "message rejected"
|
||||
|
||||
default:
|
||||
// *rmp.StateModified
|
||||
// *rmq.StateReceived
|
||||
}
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
package rabbit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"datart/config"
|
||||
|
||||
rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
|
||||
)
|
||||
|
||||
type XQK struct {
|
||||
Exchange string `json:"exchange" yaml:"exchange"`
|
||||
Queue string `json:"queue" yaml:"queue"`
|
||||
Key string `json:"key" yaml:"key"`
|
||||
QueueCap int64 `json:"queuecap" yaml:"queuecap"`
|
||||
}
|
||||
|
||||
var client *rabbitClient
|
||||
|
||||
func init() {
|
||||
endpoints, err := genEndpoints("default")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
client = new(rabbitClient)
|
||||
client.env = rmq.NewClusterEnvironment(endpoints)
|
||||
conn, err := client.env.NewConnection(context.Background())
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
client.conn = conn
|
||||
}
|
||||
|
||||
func CloseDefault(ctx context.Context) error {
|
||||
return client.Close(ctx)
|
||||
}
|
||||
|
||||
func DefaultManagement() *rmq.AmqpManagement {
|
||||
return client.Management()
|
||||
}
|
||||
|
||||
func genEndpoints(tag string) ([]rmq.Endpoint, error) {
|
||||
confs := config.Conf().RabbitConf(tag)
|
||||
endpoints := make([]rmq.Endpoint, len(confs))
|
||||
|
||||
for i, conf := range confs {
|
||||
var options *rmq.AmqpConnOptions
|
||||
|
||||
endpoints[i].Address = conf.GenAddress(false)
|
||||
endpoints[i].Options = options
|
||||
}
|
||||
|
||||
return endpoints, nil
|
||||
}
|
||||
10
go.mod
10
go.mod
|
|
@ -4,11 +4,8 @@ go 1.24.0
|
|||
|
||||
require (
|
||||
github.com/gin-gonic/gin v1.11.0
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/gorilla/websocket v1.5.3
|
||||
github.com/rabbitmq/rabbitmq-amqp-go-client v0.2.0
|
||||
github.com/redis/go-redis/v9 v9.14.0
|
||||
go.mongodb.org/mongo-driver/v2 v2.3.0
|
||||
go.uber.org/zap v1.27.0
|
||||
gopkg.in/natefinch/lumberjack.v2 v2.2.1
|
||||
gorm.io/driver/postgres v1.6.0
|
||||
|
|
@ -16,7 +13,6 @@ require (
|
|||
)
|
||||
|
||||
require (
|
||||
github.com/Azure/go-amqp v1.5.0 // indirect
|
||||
github.com/bytedance/sonic v1.14.0 // indirect
|
||||
github.com/bytedance/sonic/loader v0.3.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
|
|
@ -29,7 +25,6 @@ require (
|
|||
github.com/go-playground/validator/v10 v10.27.0 // indirect
|
||||
github.com/goccy/go-json v0.10.2 // indirect
|
||||
github.com/goccy/go-yaml v1.18.0 // indirect
|
||||
github.com/golang/snappy v1.0.0 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
|
||||
github.com/jackc/pgx/v5 v5.6.0 // indirect
|
||||
|
|
@ -37,7 +32,6 @@ require (
|
|||
github.com/jinzhu/inflection v1.0.0 // indirect
|
||||
github.com/jinzhu/now v1.1.5 // indirect
|
||||
github.com/json-iterator/go v1.1.12 // indirect
|
||||
github.com/klauspost/compress v1.16.7 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
|
||||
github.com/leodido/go-urn v1.4.0 // indirect
|
||||
github.com/mattn/go-isatty v0.0.20 // indirect
|
||||
|
|
@ -48,10 +42,6 @@ require (
|
|||
github.com/quic-go/quic-go v0.54.0 // indirect
|
||||
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
|
||||
github.com/ugorji/go/codec v1.3.0 // indirect
|
||||
github.com/xdg-go/pbkdf2 v1.0.0 // indirect
|
||||
github.com/xdg-go/scram v1.1.2 // indirect
|
||||
github.com/xdg-go/stringprep v1.0.4 // indirect
|
||||
github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
|
||||
go.uber.org/mock v0.5.0 // indirect
|
||||
go.uber.org/multierr v1.10.0 // indirect
|
||||
golang.org/x/arch v0.20.0 // indirect
|
||||
|
|
|
|||
60
go.sum
60
go.sum
|
|
@ -1,5 +1,3 @@
|
|||
github.com/Azure/go-amqp v1.5.0 h1:GRiQK1VhrNFbyx5VlmI6BsA1FCp27W5rb9kxOZScnTo=
|
||||
github.com/Azure/go-amqp v1.5.0/go.mod h1:vZAogwdrkbyK3Mla8m/CxSc/aKdnTZ4IbPxl51Y5WZE=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
|
||||
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
|
||||
|
|
@ -17,16 +15,12 @@ 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/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
|
||||
github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
|
||||
github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
|
||||
github.com/gabriel-vasile/mimetype v1.4.8 h1:FfZ3gj38NjllZIeJAmMhr+qKL8Wu+nOoI3GqacKw1NM=
|
||||
github.com/gabriel-vasile/mimetype v1.4.8/go.mod h1:ByKUIKGjh1ODkGM1asKUbQZOLGrPjydw3hYPU2YU9t8=
|
||||
github.com/gin-contrib/sse v1.1.0 h1:n0w2GMuUpWDVp7qSpvze6fAu9iRxJY4Hmj6AmBOU05w=
|
||||
github.com/gin-contrib/sse v1.1.0/go.mod h1:hxRZ5gVpWMT7Z0B0gSNYqqsSCNIJMjzvm6fqCz9vjwM=
|
||||
github.com/gin-gonic/gin v1.11.0 h1:OW/6PLjyusp2PPXtyxKHU0RbX6I/l28FTdDlae5ueWk=
|
||||
github.com/gin-gonic/gin v1.11.0/go.mod h1:+iq/FyxlGzII0KHiBGjuNn4UNENUlKbGlNmc+W50Dls=
|
||||
github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
|
||||
github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
|
||||
github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
|
||||
github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
|
||||
github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
|
||||
|
|
@ -35,23 +29,13 @@ github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJn
|
|||
github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
|
||||
github.com/go-playground/validator/v10 v10.27.0 h1:w8+XrWVMhGkxOaaowyKH35gFydVHOvC0/uWoy2Fzwn4=
|
||||
github.com/go-playground/validator/v10 v10.27.0/go.mod h1:I5QpIEbmr8On7W0TktmJAumgzX4CA1XNl4ZmDuVHKKo=
|
||||
github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
|
||||
github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
|
||||
github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
|
||||
github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
|
||||
github.com/goccy/go-yaml v1.18.0 h1:8W7wMFS12Pcas7KU+VVkaiCng+kG8QiFeFwzFb+rwuw=
|
||||
github.com/goccy/go-yaml v1.18.0/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
|
||||
github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8=
|
||||
github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk=
|
||||
github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs=
|
||||
github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
|
||||
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
|
||||
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
|
||||
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
|
||||
github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 h1:BHT72Gu3keYf3ZEu2J0b1vyeLSOYI8bm5wbJM/8yDe8=
|
||||
github.com/google/pprof v0.0.0-20250403155104-27863c87afa6/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
|
||||
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
|
|
@ -68,8 +52,6 @@ github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
|
|||
github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
|
||||
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
|
||||
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
|
||||
github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGCFR9I=
|
||||
github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
|
||||
github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
|
||||
github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
|
||||
github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ=
|
||||
|
|
@ -80,10 +62,6 @@ github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OH
|
|||
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
|
||||
github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
|
||||
github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
|
||||
github.com/onsi/ginkgo/v2 v2.23.4 h1:ktYTpKJAVZnDT4VjxSbiBenUjmlL/5QkBEocaWXiQus=
|
||||
github.com/onsi/ginkgo/v2 v2.23.4/go.mod h1:Bt66ApGPBFzHyR+JO10Zbt0Gsp4uWxu5mIOTusL46e8=
|
||||
github.com/onsi/gomega v1.38.0 h1:c/WX+w8SLAinvuKKQFh77WEucCnPk4j2OTUr7lt7BeY=
|
||||
github.com/onsi/gomega v1.38.0/go.mod h1:OcXcwId0b9QsE7Y49u+BTrL4IdKOBOKnD6VQNTJEB6o=
|
||||
github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
|
||||
github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
|
|
@ -92,8 +70,6 @@ github.com/quic-go/qpack v0.5.1 h1:giqksBPnT/HDtZ6VhtFKgoLOWmlyo9Ei6u9PqzIMbhI=
|
|||
github.com/quic-go/qpack v0.5.1/go.mod h1:+PC4XFrEskIVkcLzpEkbLqq1uCoxPhQuvK5rH1ZgaEg=
|
||||
github.com/quic-go/quic-go v0.54.0 h1:6s1YB9QotYI6Ospeiguknbp2Znb/jZYjZLRXn9kMQBg=
|
||||
github.com/quic-go/quic-go v0.54.0/go.mod h1:e68ZEaCdyviluZmy44P6Iey98v/Wfz6HCjQEm+l8zTY=
|
||||
github.com/rabbitmq/rabbitmq-amqp-go-client v0.2.0 h1:q0zPF8/7Bdm+XwjWevFynB8fNiuE65x4q2vmFxU2cjM=
|
||||
github.com/rabbitmq/rabbitmq-amqp-go-client v0.2.0/go.mod h1:t5oaK/4mJjw9dNpDzwvH6bE7p9XtM1JyObEHszFu3lU=
|
||||
github.com/redis/go-redis/v9 v9.14.0 h1:u4tNCjXOyzfgeLN+vAZaW1xUooqWDqVEsZN0U01jfAE=
|
||||
github.com/redis/go-redis/v9 v9.14.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
|
|
@ -110,19 +86,6 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
|
|||
github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
|
||||
github.com/ugorji/go/codec v1.3.0 h1:Qd2W2sQawAfG8XSvzwhBeoGq71zXOC/Q1E9y/wUcsUA=
|
||||
github.com/ugorji/go/codec v1.3.0/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4=
|
||||
github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
|
||||
github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
|
||||
github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
|
||||
github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
|
||||
github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
|
||||
github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
|
||||
github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM=
|
||||
github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI=
|
||||
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
|
||||
go.mongodb.org/mongo-driver/v2 v2.3.0 h1:sh55yOXA2vUjW1QYw/2tRlHSQViwDyPnW61AwpZ4rtU=
|
||||
go.mongodb.org/mongo-driver/v2 v2.3.0/go.mod h1:jHeEDJHJq7tm6ZF45Issun9dbogjfnPySb1vXA7EeAI=
|
||||
go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
|
||||
go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8=
|
||||
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
|
||||
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
|
||||
go.uber.org/mock v0.5.0 h1:KAMbZvZPyBPWgD14IrIQ38QCyjwpvVVV6K/bHl1IwQU=
|
||||
|
|
@ -133,44 +96,21 @@ go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
|
|||
go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
|
||||
golang.org/x/arch v0.20.0 h1:dx1zTU0MAE98U+TQ8BLl7XsJbgze2WnNKF/8tGp/Q6c=
|
||||
golang.org/x/arch v0.20.0/go.mod h1:bdwinDaKcfZUGpH09BB7ZmOfhalA8lQdzl62l8gGWsk=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
|
||||
golang.org/x/crypto v0.40.0 h1:r4x+VvoG5Fm+eJcxMaY8CQM7Lb0l1lsmjGBQ6s8BfKM=
|
||||
golang.org/x/crypto v0.40.0/go.mod h1:Qr1vMER5WyS2dfPHAlsOj01wgLbsyWtFn/aY+5+ZdxY=
|
||||
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
|
||||
golang.org/x/mod v0.25.0 h1:n7a+ZbQKQA/Ysbyb0/6IbB1H/X41mKgbhfv7AfG/44w=
|
||||
golang.org/x/mod v0.25.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
|
||||
golang.org/x/net v0.42.0 h1:jzkYrhi3YQWD6MLBJcsklgQsoAcw89EcZbJw8Z614hs=
|
||||
golang.org/x/net v0.42.0/go.mod h1:FF1RA5d3u7nAYA4z2TkclSCKh68eSXtiFwcWQpPXdt8=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
|
||||
golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
|
||||
golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
|
||||
golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
|
||||
golang.org/x/text v0.27.0 h1:4fGWRpyh641NLlecmyl4LOe6yDdfaYNrGb2zdfo4JV4=
|
||||
golang.org/x/text v0.27.0/go.mod h1:1D28KMCvyooCX9hBiosv5Tz/+YLxj0j7XhWjpSUF7CU=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
|
||||
golang.org/x/tools v0.34.0 h1:qIpSLOxeCYGg9TrcJokLBG4KFA6d795g0xkBkiESGlo=
|
||||
golang.org/x/tools v0.34.0/go.mod h1:pAP9OwEaY1CAW3HOmg3hLZC5Z0CCmzjAF2UQMSqNARg=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/protobuf v1.36.9 h1:w2gp2mA27hUeUzj9Ex9FBjsBm40zfaDtEWow293U7Iw=
|
||||
google.golang.org/protobuf v1.36.9/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
|
|
|
|||
|
|
@ -1,91 +0,0 @@
|
|||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"datart/data"
|
||||
"datart/data/mongo"
|
||||
"datart/log"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func (a *Api) PostUploadAlarm(ctx *gin.Context) {
|
||||
alarm, ip, err := a.checkAndGenUploadAlarmRequest(ctx)
|
||||
if err != nil {
|
||||
|
||||
log.Error(err)
|
||||
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 1,
|
||||
"msg": "invalid param",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
event, err := alarm.ConvertToEvent(ip)
|
||||
if err != nil {
|
||||
|
||||
log.Error(err, fmt.Sprintf(" params: %v", alarm))
|
||||
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 2,
|
||||
"msg": "convert error",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
err = mongo.InsertEvent(ctx.Request.Context(), event)
|
||||
if err != nil {
|
||||
|
||||
log.Error(err, fmt.Sprintf(" params: %v", event))
|
||||
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 3,
|
||||
"msg": "insert error",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
go func(e *mongo.Event) {
|
||||
if err := data.PublishEvent(context.Background(), e); err != nil {
|
||||
|
||||
log.Error(err, fmt.Sprintf(" params: %v", e))
|
||||
}
|
||||
}(event)
|
||||
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 0,
|
||||
"msg": "success",
|
||||
})
|
||||
}
|
||||
|
||||
func (a *Api) checkAndGenUploadAlarmRequest(ctx *gin.Context) (*mongo.Alarm, string, error) {
|
||||
alarm := new(mongo.Alarm)
|
||||
|
||||
err := ctx.ShouldBindJSON(alarm)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
ok, err := regexp.MatchString(`ssu\d{3}`, alarm.DeviceNo)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
if !ok {
|
||||
return nil, "", errors.New("invalid device_no")
|
||||
}
|
||||
|
||||
if alarm.AlarmCode < 1 || alarm.AlarmCode > 10 {
|
||||
return nil, "", errors.New("invalid alarm_code")
|
||||
}
|
||||
|
||||
if alarm.AlarmStatus < 0 || alarm.AlarmStatus > 1 {
|
||||
return nil, "", errors.New("invalid alarm_status")
|
||||
}
|
||||
|
||||
return alarm, ctx.RemoteIP(), nil
|
||||
}
|
||||
|
|
@ -1,291 +0,0 @@
|
|||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"datart/data"
|
||||
"datart/data/mongo"
|
||||
"datart/log"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/google/uuid"
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
)
|
||||
|
||||
const (
|
||||
pageSizeLimit = 500
|
||||
)
|
||||
|
||||
func (a *Api) GetEvents(ctx *gin.Context) {
|
||||
filter, sort, pageNo, pageSize, err := a.checkAndGenGetEventsRequest(ctx)
|
||||
if err != nil {
|
||||
|
||||
log.Error(err)
|
||||
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 1,
|
||||
"msg": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
events, err := mongo.FindEventsWithPageLimit(ctx.Request.Context(), filter, sort, pageNo, pageSize)
|
||||
if err != nil {
|
||||
|
||||
log.Error(err, fmt.Sprintf(" params: %v, %d, %d, %d", filter, sort, pageNo, pageSize))
|
||||
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 2,
|
||||
"msg": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 0,
|
||||
"msg": "success",
|
||||
"data": events,
|
||||
})
|
||||
}
|
||||
|
||||
func (a *Api) PostUpsertEvents(ctx *gin.Context) {
|
||||
curd, uuids, events, err := a.checkAndGenUpsertEventsRequest(ctx)
|
||||
if err != nil {
|
||||
log.Error(err)
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 1,
|
||||
"msg": err.Error(),
|
||||
"data": map[string]any{
|
||||
"success_uuids": nil,
|
||||
},
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
suuids, err := mongo.BulkWriteEventsWithUUID(ctx.Request.Context(), curd, events)
|
||||
if err != nil {
|
||||
log.Error(err, fmt.Sprintf(" params:%v", events))
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 2,
|
||||
"msg": err.Error(),
|
||||
"data": map[string]any{
|
||||
"success_uuids": suuids,
|
||||
},
|
||||
})
|
||||
} else {
|
||||
suuids = uuids
|
||||
ctx.JSON(200, gin.H{
|
||||
"code": 0,
|
||||
"msg": "success",
|
||||
"data": map[string]any{
|
||||
"success_uuids": uuids,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
uuid2Event := map[string]any{}
|
||||
for i, event := range events {
|
||||
uuid2Event[uuids[i]] = event
|
||||
}
|
||||
|
||||
go func(suuids []string, uuid2Event map[string]any) {
|
||||
workers := 5
|
||||
ch := make(chan any, len(suuids))
|
||||
var wg sync.WaitGroup
|
||||
for range workers {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for e := range ch {
|
||||
if perr := data.PublishEvent(context.Background(), e); perr != nil {
|
||||
log.Error(perr, fmt.Sprintf("publish event failed: %v", e))
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
for _, suuid := range suuids {
|
||||
if e, ok := uuid2Event[suuid]; ok {
|
||||
ch <- e
|
||||
}
|
||||
}
|
||||
close(ch)
|
||||
wg.Wait()
|
||||
}(suuids, uuid2Event)
|
||||
}
|
||||
|
||||
func (a *Api) checkAndGenGetEventsRequest(ctx *gin.Context) (bson.M, int, int64, int64, error) {
|
||||
uuidStr := ctx.Query("uuid")
|
||||
if len(uuidStr) > 0 {
|
||||
|
||||
if uuid.Validate(uuidStr) != nil {
|
||||
return nil, 0, -1, -1, errors.New("invalid uuid")
|
||||
}
|
||||
|
||||
return bson.M{"event_uuid": uuidStr}, 0, -1, -1, nil
|
||||
}
|
||||
|
||||
filter := bson.M{}
|
||||
|
||||
var err error
|
||||
begin, end := int64(-1), int64(-1)
|
||||
beginStr := ctx.Query("begin")
|
||||
if len(beginStr) > 0 {
|
||||
if begin, err = strconv.ParseInt(beginStr, 10, 64); err != nil {
|
||||
return nil, 0, -1, -1, err
|
||||
}
|
||||
}
|
||||
|
||||
endStr := ctx.Query("end")
|
||||
if len(endStr) > 0 {
|
||||
if end, err = strconv.ParseInt(endStr, 10, 64); err != nil {
|
||||
return nil, 0, -1, -1, err
|
||||
}
|
||||
}
|
||||
|
||||
if begin > 0 && end > 0 && begin > end {
|
||||
return nil, 0, -1, -1, errors.New("invalid time")
|
||||
}
|
||||
|
||||
switch {
|
||||
case begin > 0 && end < 0:
|
||||
filter["timestamp"] = bson.M{"$gte": begin}
|
||||
case begin < 0 && end > 0:
|
||||
filter["timestamp"] = bson.M{"$lte": end}
|
||||
case begin > 0 && end > 0:
|
||||
filter["timestamp"] = bson.M{"$gte": begin, "$lte": end}
|
||||
}
|
||||
|
||||
statusStr := ctx.Query("status")
|
||||
if len(statusStr) > 0 {
|
||||
statusStrs := strings.Split(statusStr, ",")
|
||||
statuss := make([]int, len(statusStrs))
|
||||
for i := range statusStrs {
|
||||
s, err := strconv.Atoi(statusStrs[i])
|
||||
if err != nil {
|
||||
return nil, 0, -1, -1, errors.New("invalid status")
|
||||
}
|
||||
statuss[i] = s
|
||||
}
|
||||
filter["status"] = bson.M{"$in": statuss}
|
||||
}
|
||||
|
||||
var sort int
|
||||
sortStr := ctx.Query("sort")
|
||||
if len(sortStr) > 0 {
|
||||
s, err := strconv.Atoi(sortStr)
|
||||
if err != nil {
|
||||
return nil, 0, -1, -1, err
|
||||
}
|
||||
|
||||
if s != 1 && s != -1 {
|
||||
return nil, 0, -1, -1, errors.New("invalid sort")
|
||||
}
|
||||
sort = s
|
||||
}
|
||||
|
||||
pageNo, pageSize := -1, -1
|
||||
pageNoStr := ctx.Query("page_no")
|
||||
pageSizeStr := ctx.Query("page_size")
|
||||
if len(pageNoStr) > 0 && len(pageSizeStr) > 0 {
|
||||
pageNo, err = strconv.Atoi(pageNoStr)
|
||||
if err != nil {
|
||||
return nil, 0, -1, -1, err
|
||||
}
|
||||
|
||||
pageSize, err = strconv.Atoi(pageSizeStr)
|
||||
if err != nil {
|
||||
return nil, 0, -1, -1, err
|
||||
}
|
||||
|
||||
if pageNo <= 0 || pageSize <= 0 {
|
||||
return nil, 0, -1, -1, errors.New("invalid page param")
|
||||
}
|
||||
|
||||
if pageSize > pageSizeLimit {
|
||||
return nil, 0, -1, -1, fmt.Errorf("too many events, max %d", pageSizeLimit)
|
||||
}
|
||||
}
|
||||
|
||||
return filter, sort, int64(pageNo), int64(pageSize), nil
|
||||
}
|
||||
|
||||
func (a *Api) checkAndGenUpsertEventsRequest(ctx *gin.Context) (byte, []string, []map[string]any, error) {
|
||||
insert := true
|
||||
var status int
|
||||
var err error
|
||||
statusStr := ctx.Query("status")
|
||||
if len(statusStr) > 0 {
|
||||
insert = false
|
||||
status, err = strconv.Atoi(statusStr)
|
||||
if err != nil {
|
||||
return 0, nil, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
events := []map[string]any{}
|
||||
uuids := []string{}
|
||||
|
||||
if !insert {
|
||||
err := ctx.ShouldBindJSON(&uuids)
|
||||
if err != nil {
|
||||
return 0, nil, nil, err
|
||||
}
|
||||
if len(uuids) == 0 {
|
||||
return 0, nil, nil, errors.New("no uuid")
|
||||
}
|
||||
if len(uuids) > pageSizeLimit {
|
||||
return 0, nil, nil, fmt.Errorf("too many uuids, max %d", pageSizeLimit)
|
||||
}
|
||||
|
||||
operation := mongo.GenOperation(mongo.EventStatusAction[status], ctx.RemoteIP())
|
||||
for _, uid := range uuids {
|
||||
if uuid.Validate(uid) != nil {
|
||||
return 0, nil, nil, errors.New("invalid uuid")
|
||||
}
|
||||
|
||||
events = append(events,
|
||||
map[string]any{"event_uuid": uid,
|
||||
"status": status,
|
||||
"operation": operation})
|
||||
}
|
||||
|
||||
return 'u', uuids, events, nil
|
||||
}
|
||||
|
||||
err = ctx.ShouldBindJSON(&events)
|
||||
if err != nil {
|
||||
return 0, nil, nil, err
|
||||
}
|
||||
if len(events) == 0 {
|
||||
return 0, nil, nil, errors.New("no event")
|
||||
}
|
||||
if len(events) > pageSizeLimit {
|
||||
return 0, nil, nil, fmt.Errorf("too many events, max %d", pageSizeLimit)
|
||||
}
|
||||
|
||||
for _, event := range events {
|
||||
noUUID := true
|
||||
if eu, ok := event["event_uuid"]; ok {
|
||||
if eID, ok := eu.(string); ok {
|
||||
if uuid.Validate(eID) == nil {
|
||||
uuids = append(uuids, eID)
|
||||
noUUID = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if noUUID {
|
||||
return 0, nil, nil, errors.New("invalid uuid")
|
||||
}
|
||||
|
||||
if len(event) < 2 {
|
||||
return 0, nil, nil, errors.New("invalid event")
|
||||
}
|
||||
}
|
||||
|
||||
return 'c', uuids, events, nil
|
||||
}
|
||||
|
|
@ -11,10 +11,7 @@ func LoadRoute(engine *gin.Engine) {
|
|||
|
||||
a := new(api.Api)
|
||||
ga := engine.Group("api")
|
||||
ga.POST("/alarm", a.PostUploadAlarm)
|
||||
ga.GET("/points", a.GetPoints)
|
||||
ga.GET("/events", a.GetEvents)
|
||||
ga.POST("/events", a.PostUpsertEvents)
|
||||
ga.POST("/command", a.PostExecuteCommand)
|
||||
ga.GET("/104up", a.Cl104Up)
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue