zeaslity/ProjectOctopus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[message pusher] - standlone gin service

Browse Source
This commit is contained in:
zeaslity
2024-04-23 11:45:32 +08:00
parent 30d83f3c61
commit 16541183ef
41 changed files with 1481 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
agent-common/message_pusher/client.go
View File

@@ -1,165 +0,0 @@
package message_pusher
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"regexp"
"strings"
"wdd.io/agent-common/logger"
)
var (
topicRegex = regexp.MustCompile(`^[-_A-Za-z0-9]{1,64}$`) // Same as in server/server.go
log = logger.Log
)
const (
maxResponseBytes = 4096
)
type Client struct {
config *Config
}
// Message is a struct that represents a ntfy message
type Message struct { // TODO combine with server.message
ID string
Event string
Time int64
Topic string
Message string
Title string
Priority int
Tags []string
Click string
Icon string
Attachment *Attachment
// Additional fields
TopicURL string
SubscriptionID string
Raw string
}
// Attachment represents a message attachment
type Attachment struct {
Name string `json:"name"`
Type string `json:"type,omitempty"`
Size int64 `json:"size,omitempty"`
Expires int64 `json:"expires,omitempty"`
URL string `json:"url"`
Owner string `json:"-"` // IP address of uploader, used for rate limiting
}
// New creates a new Client using a given Config
func New(config *Config) *Client {
return &Client{
config: config,
}
}
func NewDefaultClient() *Client {
defaultConfig := NewDefaultConfig()
return New(defaultConfig)
}
func (c *Client) PublishDefault(message bytes.Buffer, options []PublishOption) (*Message, error) {
if c.config.DefaultTopic == "" {
return nil, errors.New("[PublishDefault] - topic empty")
}
// parse default
options = c.parseConfigToOption(options)
return c.PublishReader(c.config.DefaultTopic, bytes.NewReader(message.Bytes()), options)
}
// Publish sends a message to a specific topic, optionally using options.
// See PublishReader for details.
func (c *Client) Publish(topic, message string, options []PublishOption) (*Message, error) {
return c.PublishReader(topic, strings.NewReader(message), options)
}
// PublishReader sends a message to a specific topic, optionally using options.
//
// A topic can be either a full URL (e.g. https://myhost.lan/mytopic), a short URL which is then prepended https://
// (e.g. myhost.lan -> https://myhost.lan), or a short name which is expanded using the default host in the
// config (e.g. mytopic -> https://ntfy.sh/mytopic).
//
// To pass title, priority and tags, check out WithTitle, WithPriority, WithTagsList, WithDelay, WithNoCache,
// WithNoFirebase, and the generic WithHeader.
func (c *Client) PublishReader(topic string, body io.Reader, options []PublishOption) (*Message, error) {
topicURL, err := c.expandTopicURL(topic)
if err != nil {
return nil, err
}
req, err := http.NewRequest("POST", topicURL, body)
if err != nil {
return nil, err
}
for _, option := range options {
if err := option(req); err != nil {
return nil, err
}
}
log.DebugF("%s Publishing message with headers %s", topicURL, req.Header)
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
b, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
return nil, errors.New(strings.TrimSpace(string(b)))
}
m, err := toMessage(string(b), topicURL, "")
if err != nil {
return nil, err
}
return m, nil
}
func (c *Client) expandTopicURL(topic string) (string, error) {
if strings.HasPrefix(topic, "http://") || strings.HasPrefix(topic, "https://") {
return topic, nil
} else if strings.Contains(topic, "/") {
return fmt.Sprintf("https://%s", topic), nil
}
if !topicRegex.MatchString(topic) {
return "", fmt.Errorf("invalid topic name: %s", topic)
}
return fmt.Sprintf("%s/%s", c.config.DefaultHost, topic), nil
}
func (c *Client) parseConfigToOption(options []PublishOption) []PublishOption {
config := c.config
if config.DefaultToken != "" {
options = append(options, WithBearerAuth(config.DefaultToken))
} else if config.DefaultUser != "" {
if *config.DefaultPassword != "" {
options = append(options, WithBasicAuth(config.DefaultUser, *config.DefaultPassword))
} else {
log.ErrorF("[parseConfigToOption] - default password is empty!")
}
}
return options
}
func toMessage(s, topicURL, subscriptionID string) (*Message, error) {
var m *Message
if err := json.NewDecoder(strings.NewReader(s)).Decode(&m); err != nil {
return nil, err
}
m.TopicURL = topicURL
m.SubscriptionID = subscriptionID
m.Raw = s
return m, nil
}

34
agent-common/message_pusher/client_test.go
View File

@@ -1,34 +0,0 @@
package message_pusher
import (
"testing"
"wdd.io/agent-common/utils"
)
func TestClient_Publish(t *testing.T) {
client := NewDefaultClient()
optionList := []PublishOption{
WithTitle("测试内容"),
WithPriority("5"),
WithMarkdown(),
}
deployPush := DeployPush{
Namespace: "uavcloud-dev",
AppName: "cmii-uav-platform",
Replicas: "1",
DeployStatus: false,
}
deployPush.ParseCmiiDeployTemplate()
message, err := client.PublishDefault(deployPush.ParseCmiiDeployTemplate(), optionList)
if err != nil {
return
}
utils.BeautifulPrint(message)
}

62
agent-common/message_pusher/config.go
View File

@@ -1,62 +0,0 @@
package message_pusher
import (
"gopkg.in/yaml.v3"
"os"
)
const (
// DefaultBaseURL is the base URL used to expand short topic names
DefaultBaseURL = "https://push.107421.xyz"
DefaultBaseToken = "tk_zvdb67fwj1hrjivkq3ga9z7u63av5"
DefaultTopic = "cmii"
)
// Config is the config struct for a Client
type Config struct {
DefaultHost string `yaml:"default-host"`
DefaultUser string `yaml:"default-user"`
DefaultPassword *string `yaml:"default-password"`
DefaultToken string `yaml:"default-token"`
DefaultCommand string `yaml:"default-command"`
DefaultTopic string `yaml:"default-topic"`
Subscribe []Subscribe `yaml:"subscribe"`
}
// Subscribe is the struct for a Subscription within Config
type Subscribe struct {
Topic string `yaml:"topic"`
User *string `yaml:"user"`
Password *string `yaml:"password"`
Token *string `yaml:"token"`
Command string `yaml:"command"`
If map[string]string `yaml:"if"`
}
// NewDefaultConfig creates a new Config struct for a Client
func NewDefaultConfig() *Config {
return &Config{
DefaultHost: DefaultBaseURL,
DefaultUser: "",
DefaultPassword: nil,
DefaultToken: DefaultBaseToken,
DefaultTopic: DefaultTopic,
DefaultCommand: "",
Subscribe: nil,
}
}
// LoadConfig loads the Client config from a yaml file
func LoadConfig(filename string) (*Config, error) {
b, err := os.ReadFile(filename)
if err != nil {
return nil, err
}
c := NewDefaultConfig()
if err := yaml.Unmarshal(b, c); err != nil {
return nil, err
}
return c, nil
}

204
agent-common/message_pusher/options.go
View File

@@ -1,204 +0,0 @@
package message_pusher
import (
"encoding/base64"
"fmt"
"net/http"
"strings"
"time"
)
// RequestOption is a generic request option that can be added to Client calls
type RequestOption = func(r *http.Request) error
// PublishOption is an option that can be passed to the Client.Publish call
type PublishOption = RequestOption
// SubscribeOption is an option that can be passed to a Client.Subscribe or Client.Poll call
type SubscribeOption = RequestOption
// WithMessage sets the notification message. This is an alternative way to passing the message body.
func WithMessage(message string) PublishOption {
return WithHeader("X-Message", message)
}
// WithTitle adds a title to a message
func WithTitle(title string) PublishOption {
return WithHeader("X-Title", title)
}
// WithPriority adds a priority to a message. The priority can be either a number (1=min, 5=max),
// or the corresponding names (see util.ParsePriority).
func WithPriority(priority string) PublishOption {
return WithHeader("X-Priority", priority)
}
// WithTagsList adds a list of tags to a message. The tags parameter must be a comma-separated list
// of tags. To use a slice, use WithTags instead
func WithTagsList(tags string) PublishOption {
return WithHeader("X-Tags", tags)
}
// WithTags adds a list of a tags to a message
func WithTags(tags []string) PublishOption {
return WithTagsList(strings.Join(tags, ","))
}
// WithDelay instructs the server to send the message at a later date. The delay parameter can be a
// Unix timestamp, a duration string or a natural langage string. See https://ntfy.sh/docs/publish/#scheduled-delivery
// for details.
func WithDelay(delay string) PublishOption {
return WithHeader("X-Delay", delay)
}
// WithClick makes the notification action open the given URL as opposed to entering the detail view
func WithClick(url string) PublishOption {
return WithHeader("X-Click", url)
}
// WithIcon makes the notification use the given URL as its icon
func WithIcon(icon string) PublishOption {
return WithHeader("X-Icon", icon)
}
// WithActions adds custom user actions to the notification. The value can be either a JSON array or the
// simple format definition. See https://ntfy.sh/docs/publish/#action-buttons for details.
func WithActions(value string) PublishOption {
return WithHeader("X-Actions", value)
}
// WithAttach sets a URL that will be used by the client to download an attachment
func WithAttach(attach string) PublishOption {
return WithHeader("X-Attach", attach)
}
// WithMarkdown instructs the server to interpret the message body as Markdown
func WithMarkdown() PublishOption {
return WithHeader("X-Markdown", "yes")
}
// WithFilename sets a filename for the attachment, and/or forces the HTTP body to interpreted as an attachment
func WithFilename(filename string) PublishOption {
return WithHeader("X-Filename", filename)
}
// WithEmail instructs the server to also send the message to the given e-mail address
func WithEmail(email string) PublishOption {
return WithHeader("X-Email", email)
}
// WithBasicAuth adds the Authorization header for basic auth to the request
func WithBasicAuth(user, pass string) PublishOption {
return WithHeader("Authorization", fmt.Sprintf("Basic %s", base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%s:%s", user, pass)))))
}
// WithBearerAuth adds the Authorization header for Bearer auth to the request
func WithBearerAuth(token string) PublishOption {
return WithHeader("Authorization", fmt.Sprintf("Bearer %s", token))
}
// WithEmptyAuth clears the Authorization header
func WithEmptyAuth() PublishOption {
return RemoveHeader("Authorization")
}
// WithNoCache instructs the server not to cache the message server-side
func WithNoCache() PublishOption {
return WithHeader("X-Cache", "no")
}
// WithNoFirebase instructs the server not to forward the message to Firebase
func WithNoFirebase() PublishOption {
return WithHeader("X-Firebase", "no")
}
// WithSince limits the number of messages returned from the server. The parameter since can be a Unix
// timestamp (see WithSinceUnixTime), a duration (WithSinceDuration) the word "all" (see WithSinceAll).
func WithSince(since string) SubscribeOption {
return WithQueryParam("since", since)
}
// WithSinceAll instructs the server to return all messages for the given topic from the server
func WithSinceAll() SubscribeOption {
return WithSince("all")
}
// WithSinceDuration instructs the server to return all messages since the given duration ago
func WithSinceDuration(since time.Duration) SubscribeOption {
return WithSinceUnixTime(time.Now().Add(-1 * since).Unix())
}
// WithSinceUnixTime instructs the server to return only messages newer or equal to the given timestamp
func WithSinceUnixTime(since int64) SubscribeOption {
return WithSince(fmt.Sprintf("%d", since))
}
// WithPoll instructs the server to close the connection after messages have been returned. Don't use this option
// directly. Use Client.Poll instead.
func WithPoll() SubscribeOption {
return WithQueryParam("poll", "1")
}
// WithScheduled instructs the server to also return messages that have not been sent yet, i.e. delayed/scheduled
// messages (see WithDelay). The messages will have a future date.
func WithScheduled() SubscribeOption {
return WithQueryParam("scheduled", "1")
}
// WithFilter is a generic subscribe option meant to be used to filter for certain messages only
func WithFilter(param, value string) SubscribeOption {
return WithQueryParam(param, value)
}
// WithMessageFilter instructs the server to only return messages that match the exact message
func WithMessageFilter(message string) SubscribeOption {
return WithQueryParam("message", message)
}
// WithTitleFilter instructs the server to only return messages with a title that match the exact string
func WithTitleFilter(title string) SubscribeOption {
return WithQueryParam("title", title)
}
// WithPriorityFilter instructs the server to only return messages with the matching priority. Not that messages
// without priority also implicitly match priority 3.
func WithPriorityFilter(priority int) SubscribeOption {
return WithQueryParam("priority", fmt.Sprintf("%d", priority))
}
// WithTagsFilter instructs the server to only return messages that contain all of the given tags
func WithTagsFilter(tags []string) SubscribeOption {
return WithQueryParam("tags", strings.Join(tags, ","))
}
// WithHeader is a generic option to add headers to a request
func WithHeader(header, value string) RequestOption {
return func(r *http.Request) error {
if value != "" {
r.Header.Set(header, value)
}
return nil
}
}
// WithQueryParam is a generic option to add query parameters to a request
func WithQueryParam(param, value string) RequestOption {
return func(r *http.Request) error {
if value != "" {
q := r.URL.Query()
q.Add(param, value)
r.URL.RawQuery = q.Encode()
}
return nil
}
}
// RemoveHeader is a generic option to remove a header from a request
func RemoveHeader(header string) RequestOption {
return func(r *http.Request) error {
if header != "" {
delete(r.Header, header)
}
return nil
}
}

27
agent-common/message_pusher/publish.go
View File

@@ -1,27 +0,0 @@
package message_pusher
var DefaultClientOp = NewDefaultClient()
var CmiiUpdatePushOptions = []PublishOption{
WithTitle("更新应用"),
WithPriority("3"),
}
func PushCmiiUpdateMessage(cmiiEnv, appName, newTag string, updateStatus bool) {
deployPush := DeployPush{
Namespace: cmiiEnv,
AppName: appName,
DeployStatus: updateStatus,
ToTag: newTag,
}
deployPush.ParseCmiiDeployTemplate()
_, err := DefaultClientOp.PublishDefault(deployPush.ParseCmiiDeployTemplate(), CmiiUpdatePushOptions)
if err != nil {
log.ErrorF("[PushCmiiUpdateMessage] - message push error ！ %s", err.Error())
return
}
}

43
agent-common/message_pusher/push_template.go
View File

@@ -1,43 +0,0 @@
package message_pusher
import (
"bytes"
"text/template"
)
const cmiiDeployTemplate = `
{{if .DeployStatus}}
部署状态: 成功😍
{{- else }}
部署状态: 失败👿👿👿
{{- end}}
命名空间: {{.Namespace}}
应用名称: {{.AppName}}
副本数量: {{.Replicas}}
目标版本: {{.ToTag}}
`
type DeployPush struct {
Namespace string
AppName string
Replicas string
DeployStatus bool
ToTag string
}
func (d DeployPush) ParseCmiiDeployTemplate() bytes.Buffer {
// 解析模板
tmpl, err := template.New("cmiiDeployTemplate").Parse(cmiiDeployTemplate)
if err != nil {
panic(err)
}
// 应用数据并打印结果
var result bytes.Buffer
err = tmpl.Execute(&result, d)
if err != nil {
panic(err)
}
return result
}

20
agent-common/message_pusher/push_template_test.go
View File

@@ -1,20 +0,0 @@
package message_pusher
import (
"fmt"
"testing"
)
func TestDeployPush_ParseCmiiDeployTemplate(t *testing.T) {
deployPush := DeployPush{
Namespace: "casc",
AppName: "sdasdas",
Replicas: "dasdasd",
DeployStatus: false,
ToTag: "5.4.0",
}
template := deployPush.ParseCmiiDeployTemplate()
fmt.Println(template.String())
}