diff --git a/config/config.go b/config/config.go index 00e9743..ff8f4a4 100644 --- a/config/config.go +++ b/config/config.go @@ -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") diff --git a/config/mongo.go b/config/mongo.go deleted file mode 100644 index 42c8f70..0000000 --- a/config/mongo.go +++ /dev/null @@ -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" -} diff --git a/config/rabbit.go b/config/rabbit.go deleted file mode 100644 index 44ef6b0..0000000 --- a/config/rabbit.go +++ /dev/null @@ -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" -} diff --git a/configs/cl104.json b/configs/cl104.json index 4ce0806..7a73a41 100644 --- a/configs/cl104.json +++ b/configs/cl104.json @@ -1,5 +1,2 @@ { - "station000":{ - "127.0.0.1:8899":[0] - } } \ No newline at end of file diff --git a/configs/mongo.json b/configs/mongo.json deleted file mode 100644 index 55487c7..0000000 --- a/configs/mongo.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "default":{ - "addrs":["127.0.0.1:27017"], - "username":"admin", - "password":"password", - "authsource":"admin", - "authmechanism":"SCRAM-SHA-256" - } -} \ No newline at end of file diff --git a/configs/rabbit.json b/configs/rabbit.json deleted file mode 100644 index b1e8bb1..0000000 --- a/configs/rabbit.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "default": [ - { - "broker": "127.0.0.1:5672", - "username": "rabbitmq", - "password": "password" - } - ] -} \ No newline at end of file diff --git a/data/data.go b/data/data.go index a1a750e..7adb1be 100644 --- a/data/data.go +++ b/data/data.go @@ -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() } diff --git a/data/mongo/alarm.go b/data/mongo/alarm.go deleted file mode 100644 index 3231938..0000000 --- a/data/mongo/alarm.go +++ /dev/null @@ -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 -} diff --git a/data/mongo/event.go b/data/mongo/event.go deleted file mode 100644 index 51e0bb0..0000000 --- a/data/mongo/event.go +++ /dev/null @@ -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 -} diff --git a/data/mongo/mongo.go b/data/mongo/mongo.go deleted file mode 100644 index 0de0696..0000000 --- a/data/mongo/mongo.go +++ /dev/null @@ -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) -} diff --git a/data/publish_event.go b/data/publish_event.go deleted file mode 100644 index 11ddf8e..0000000 --- a/data/publish_event.go +++ /dev/null @@ -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) - } -} diff --git a/data/rabbit/client.go b/data/rabbit/client.go deleted file mode 100644 index 04599d5..0000000 --- a/data/rabbit/client.go +++ /dev/null @@ -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) -} diff --git a/data/rabbit/consume.go b/data/rabbit/consume.go deleted file mode 100644 index 7b5a0c7..0000000 --- a/data/rabbit/consume.go +++ /dev/null @@ -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 - } - } -} diff --git a/data/rabbit/manage.go b/data/rabbit/manage.go deleted file mode 100644 index d948863..0000000 --- a/data/rabbit/manage.go +++ /dev/null @@ -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 -} diff --git a/data/rabbit/publish.go b/data/rabbit/publish.go deleted file mode 100644 index c05c6ca..0000000 --- a/data/rabbit/publish.go +++ /dev/null @@ -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 - } - } -} diff --git a/data/rabbit/rabbit.go b/data/rabbit/rabbit.go deleted file mode 100644 index 12edcbf..0000000 --- a/data/rabbit/rabbit.go +++ /dev/null @@ -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 -} diff --git a/go.mod b/go.mod index 91cf436..01caea8 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 04f4197..8a769bf 100644 --- a/go.sum +++ b/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= diff --git a/route/api/alarm.go b/route/api/alarm.go deleted file mode 100644 index 3a7fcfe..0000000 --- a/route/api/alarm.go +++ /dev/null @@ -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 -} diff --git a/route/api/event.go b/route/api/event.go deleted file mode 100644 index 8c0e1c1..0000000 --- a/route/api/event.go +++ /dev/null @@ -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 -} diff --git a/route/route.go b/route/route.go index 110ef12..4c48f8f 100644 --- a/route/route.go +++ b/route/route.go @@ -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) }