sparrow/pkg/ruleEngine/nodes/dingtalk_robot_node.go

package nodes

import (
	"encoding/json"
	"github.com/gogf/gf/encoding/gjson"
	"github.com/gogf/gf/os/grpool"
	"net/http"
	"sparrow/pkg/protocol"
	"sparrow/pkg/ruleEngine"
	"sparrow/pkg/server"
	"sparrow/pkg/utils"
	"sync"
)

type DingTalkRobotNode struct {
	pool    *grpool.Pool
	config  *DingTalkRobotNodeConfig
	client  *utils.HttpClient
	bufPool sync.Pool
}

type DingTalkRobotNodeConfig struct {
	WebHook string `json:"webhook"` // webhook
}

func (d *DingTalkRobotNode) Init(ctx ruleEngine.Context, config string) error {
	d.pool = grpool.New(10)
	if config == "" {
		d.config = &DingTalkRobotNodeConfig{
			WebHook: "http://localhost:8899/test",
		}
	} else {
		c := new(DingTalkRobotNodeConfig)
		err := json.Unmarshal([]byte(config), c)
		if err != nil {
			return err
		}
		d.config = c
	}
	client := utils.NewHttpClient()
	client.SetLogger(server.Log)
	d.client = client
	return nil
}

func (d *DingTalkRobotNode) OnMessage(ctx ruleEngine.Context, message *protocol.Message) error {
	body, err := d.newBody(message)
	if body == nil {
		return nil
	} else if err != nil {
		return err
	}
	req, err := d.client.Post(d.config.WebHook, "application/json", gjson.New(body))

	return d.pool.Add(func() {
		if err != nil {
			server.Log.Errorf("请求出错%s", err.Error())
			return
		}
		if req != nil && req.Response() != nil {
			defer req.Close()
			if req.Response().StatusCode == http.StatusOK {
				ctx.TellSuccess(message)
			}
		}
	})
}

func (d *DingTalkRobotNode) newBody(message *protocol.Message) (body map[string]interface{}, err error) {
	msg := map[string]interface{}{
		"content": message.AlarmMessage,
	}
	body = map[string]interface{}{
		"msgtype": "text",
		"text":    msg,
	}
	return body, nil
}