代码拉取完成，页面将自动刷新
package main
import (
"bytes"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"os/exec"
"path/filepath"
"regexp"
"strconv"
"strings"
"time"
"os"
"github.com/Jeffail/gabs"
"github.com/buaazp/fasthttprouter"
"github.com/nu7hatch/gouuid"
logger "github.com/snail007/mini-logger"
"github.com/valyala/fasthttp"
)
var (
apiToken string
apiTimeout = time.Second * 30
)
func apiMessageAdd(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
Name := string(ctx.QueryArgs().Peek("Name"))
Comment := string(ctx.QueryArgs().Peek("Comment"))
DurableS := string(ctx.QueryArgs().Peek("Durable"))
IsNeedTokenS := string(ctx.QueryArgs().Peek("IsNeedToken"))
Mode := string(ctx.QueryArgs().Peek("Mode"))
Token := string(ctx.QueryArgs().Peek("Token"))
if Name == "" || DurableS == "" || IsNeedTokenS == "" {
response(ctx, "", errors.New("args required.10001"))
return
}
if _, _, err := getMessage(Name); err == nil {
response(ctx, "", errors.New("message exists"))
return
}
Durable := false
if DurableS == "1" {
Durable = true
}
IsNeedToken := false
if IsNeedTokenS == "1" {
IsNeedToken = true
}
if IsNeedToken && Token == "" {
response(ctx, "", errors.New("message exists"))
return
}
if Mode != "fanout" && Mode != "topic" && Mode != "direct" {
response(ctx, "", errors.New("args required.10002"))
return
}
m := message{
Name: Name,
Durable: Durable,
Mode: Mode,
IsNeedToken: IsNeedToken,
Token: Token,
Comment: Comment,
Consumers: []consumer{},
}
err := addMessage(m)
if err == nil {
err = writeMessagesToFile(messages, cfg.GetString("consume.DataFile"))
}
response(ctx, err, err)
}
func apiMessageUpdate(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
Name := string(ctx.QueryArgs().Peek("Name"))
Comment := string(ctx.QueryArgs().Peek("Comment"))
DurableS := string(ctx.QueryArgs().Peek("Durable"))
IsNeedTokenS := string(ctx.QueryArgs().Peek("IsNeedToken"))
Mode := string(ctx.QueryArgs().Peek("Mode"))
Token := string(ctx.QueryArgs().Peek("Token"))
if Name == "" || DurableS == "" || IsNeedTokenS == "" {
response(ctx, "", errors.New("args required.10003"))
return
}
if _, _, err := getMessage(Name); err != nil {
response(ctx, "", errors.New("message not found"))
return
}
Durable := false
if DurableS == "1" {
Durable = true
}
IsNeedToken := false
if IsNeedTokenS == "1" {
IsNeedToken = true
if Token == "" {
response(ctx, "", errors.New("args required.10013"))
return
}
}
if Mode != "fanout" && Mode != "topic" && Mode != "direct" {
response(ctx, "", errors.New("args required.10004"))
return
}
m := message{
Name: Name,
Durable: Durable,
Mode: Mode,
IsNeedToken: IsNeedToken,
Token: Token,
Comment: Comment,
}
err := updateMessage(m)
if err == nil {
err = writeMessagesToFile(messages, cfg.GetString("consume.DataFile"))
}
response(ctx, err, err)
}
func apiMessageDelete(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
exchangeName := string(ctx.QueryArgs().Peek("Name"))
msg, _, err := getMessage(exchangeName)
if err != nil {
response(ctx, "", errors.New("message not found"))
return
}
err = deleteMessage(*msg)
if err != nil {
response(ctx, "", err)
return
}
err = writeMessagesToFile(messages, cfg.GetString("consume.DataFile"))
if err != nil {
response(ctx, "", err)
return
}
response(ctx, "", nil)
}
func apiMessageStatus(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
exchangeName := string(ctx.QueryArgs().Peek("Name"))
j, err := statusMessage(exchangeName)
if err != nil {
response(ctx, "", err)
return
}
ctx.WriteString("{\"code\":1,\"data\":" + j + "}")
}
func apiConsumerAdd(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
exchangeName := string(ctx.QueryArgs().Peek("Name"))
IDUUID, _ := uuid.NewV4()
Comment := string(ctx.QueryArgs().Peek("Comment"))
CodeS := string(ctx.QueryArgs().Peek("Code"))
CheckCodeS := string(ctx.QueryArgs().Peek("CheckCode"))
RouteKey := string(ctx.QueryArgs().Peek("RouteKey"))
TimeoutS := string(ctx.QueryArgs().Peek("Timeout"))
URL := string(ctx.QueryArgs().Peek("URL"))
if exchangeName == "" || CodeS == "" || CheckCodeS == "" || TimeoutS == "" || URL == "" {
response(ctx, "", errors.New("args required.10005"))
return
}
msg, _, err := getMessage(exchangeName)
if err != nil {
response(ctx, "", err)
return
}
ID := IDUUID.String()
CheckCode := false
if CheckCodeS == "1" {
CheckCode = true
}
if ok, err := regexp.Match(`[1-9]\d{1,2}`, []byte(CodeS)); !ok || err != nil {
response(ctx, "", errors.New("args required.10006"))
return
}
if ok, err := regexp.Match(`[1-9]\d*`, []byte(TimeoutS)); !ok || err != nil {
response(ctx, "", errors.New("args required.10007"))
return
}
CheckCode = true
codeI, _ := strconv.Atoi(CodeS)
TimeoutI, _ := strconv.Atoi(TimeoutS)
c := consumer{
ID: ID,
Comment: Comment,
CheckCode: CheckCode,
Code: float64(codeI),
Timeout: float64(TimeoutI),
URL: URL,
RouteKey: RouteKey,
}
err = addConsumer(*msg, c)
if err == nil {
err = writeMessagesToFile(messages, cfg.GetString("consume.DataFile"))
}
response(ctx, err, err)
}
func apiConsumerUpdate(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
exchangeName := string(ctx.QueryArgs().Peek("Name"))
ID := string(ctx.QueryArgs().Peek("ID"))
Comment := string(ctx.QueryArgs().Peek("Comment"))
CodeS := string(ctx.QueryArgs().Peek("Code"))
CheckCodeS := string(ctx.QueryArgs().Peek("CheckCode"))
RouteKey := string(ctx.QueryArgs().Peek("RouteKey"))
TimeoutS := string(ctx.QueryArgs().Peek("Timeout"))
URL := string(ctx.QueryArgs().Peek("URL"))
if exchangeName == "" || CodeS == "" || CheckCodeS == "" || TimeoutS == "" || URL == "" {
response(ctx, "", errors.New("args required.10008"))
return
}
msg, _, err := getMessage(exchangeName)
if err != nil {
response(ctx, "", errors.New("message not found"))
return
}
c, _, _, err := getConsumer(exchangeName, ID)
if err != nil {
response(ctx, "", errors.New("consumer not found"))
return
}
CheckCode := false
if CheckCodeS == "1" {
CheckCode = true
}
if ok, err := regexp.Match(`[1-9]\d{1,2}`, []byte(CodeS)); !ok || err != nil {
response(ctx, "", errors.New("args required.10009"))
return
}
if ok, err := regexp.Match(`[1-9]\d*`, []byte(TimeoutS)); !ok || err != nil {
response(ctx, "", errors.New("args required.10010"))
return
}
CheckCode = true
codeI, _ := strconv.Atoi(CodeS)
TimeoutI, _ := strconv.Atoi(TimeoutS)
c0 := consumer{
ID: c.ID,
Comment: Comment,
CheckCode: CheckCode,
Code: float64(codeI),
Timeout: float64(TimeoutI),
URL: URL,
RouteKey: RouteKey,
}
err = updateConsumer(*msg, c0)
if err == nil {
err = writeMessagesToFile(messages, cfg.GetString("consume.DataFile"))
}
response(ctx, err, err)
}
func apiConsumerDelete(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
exchangeName := string(ctx.QueryArgs().Peek("Name"))
ID := string(ctx.QueryArgs().Peek("ID"))
msg, _, err := getMessage(exchangeName)
if err != nil {
response(ctx, "", errors.New("message not found"))
return
}
c, _, _, err := getConsumer(exchangeName, ID)
if err != nil {
response(ctx, "", errors.New("consumer not found"))
return
}
err = deleteConsumer(*msg, *c)
if err != nil {
response(ctx, "", err)
return
}
err = writeMessagesToFile(messages, cfg.GetString("consume.DataFile"))
if err != nil {
response(ctx, "", err)
return
}
response(ctx, "", nil)
}
func apiConsumerStatus(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
exchangeName := string(ctx.QueryArgs().Peek("Name"))
consumerID := string(ctx.QueryArgs().Peek("ID"))
j, e := statusConsumer(exchangeName, consumerID)
d := ""
if e == nil {
d = j.String()
ctx.WriteString("{\"code\":1,\"data\":" + d + "}")
} else {
response(ctx, e, e)
}
}
func apiPublish(ctx *fasthttp.RequestCtx) {
queryString := string(ctx.QueryArgs().QueryString())
exchangeName := ctx.UserValue("name").(string)
msg, _, err := getMessage(exchangeName)
if err != nil {
ctx.Response.SetStatusCode(fasthttp.StatusInternalServerError)
ctx.WriteString(err.Error())
return
}
tokenB := ctx.Request.Header.Peek("Token")
token := string(tokenB)
if msg.IsNeedToken && token != msg.Token {
ctx.Response.SetStatusCode(fasthttp.StatusInternalServerError)
ctx.WriteString("token error")
return
}
routeKeyB := ctx.Request.Header.Peek("RouteKey")
routeKey := string(routeKeyB)
method := strings.ToLower(string(ctx.Request.Header.Method()))
headerMap := make(map[string]string)
ignores := cfg.GetStringSlice("publish.IgnoreHeaders")
ctx.Request.Header.VisitAll(func(k, v []byte) {
var found = false
for _, ignore := range ignores {
k1 := strings.ToLower(strings.TrimSpace(string(k)))
k2 := strings.ToLower(strings.TrimSpace(ignore))
if k1 == k2 {
found = true
break
}
}
if !found {
headerMap[strings.TrimSpace(string(k))] = string(v)
}
})
encodeString := base64.StdEncoding.EncodeToString(ctx.Request.Body())
mqMessage := gabs.New()
a, _ := json.Marshal(headerMap)
mqMessage.Set(string(a), "header")
mqMessage.Set(ctx.RemoteIP(), "ip")
mqMessage.Set(encodeString, "body")
mqMessage.Set(method, "method")
mqMessage.Set(queryString, "args")
err = publish(mqMessage.String(), exchangeName, routeKey, token)
if err == nil {
ctx.Response.SetStatusCode(fasthttp.StatusNoContent)
return
}
ctx.Response.SetStatusCode(fasthttp.StatusInternalServerError)
ctx.WriteString(err.Error())
return
}
func apiReload(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
reload()
response(ctx, "", nil)
}
func apiRestart(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
err := restart()
response(ctx, "", err)
}
func apiConfig(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
//j, e := config()
response(ctx, messages, nil)
}
func apiLogList(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
f, _ := filepath.Abs(cfg.GetString("log.dir"))
fs, _ := filepath.Glob(f + "/*")
var list []string
for _, v := range fs {
list = append(list, filepath.Base(v))
}
response(ctx, list, nil)
}
func apiLogFile(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
filename := string(ctx.QueryArgs().Peek("file"))
if filename == "" || strings.ContainsAny(filename, "/\\") {
response(ctx, "", errors.New("args required.10011"))
return
}
f, _ := filepath.Abs(cfg.GetString("log.dir"))
file := filepath.Join(f, filename)
if _, e := os.Stat(file); os.IsNotExist(e) {
response(ctx, "", errors.New("file not found"))
return
}
ctx.Response.Header.Set("Content-Type", "application/force-download")
ctx.Response.Header.Set("Content-Disposition", "attachment; filename=\""+filename+"\"")
ctx.Response.Header.Set("Content-Transfer-Encoding", "binary")
ctx.SendFile(file)
}
func apiLog(ctx *fasthttp.RequestCtx) {
if !checkRequest(ctx) {
tokenError(ctx)
return
}
keyword := string(ctx.QueryArgs().Peek("keyword"))
logType := string(ctx.QueryArgs().Peek("type"))
if logType == "" {
response(ctx, "", errors.New("args required.10012"))
return
}
file, _ := filepath.Abs(filepath.Join(cfg.GetString("log.dir"), logType) + ".log")
commandStr := ""
if keyword == "" {
commandStr = fmt.Sprintf("tail -n 100 %s", file)
} else {
commandStr = fmt.Sprintf("grep \"%s\" %s |tail -n 100", keyword, file)
}
cmd := exec.Command("bash", "-c", commandStr)
stdout := &bytes.Buffer{}
stderr := &bytes.Buffer{}
cmd.Stdout = stdout
cmd.Stderr = stderr
err := cmd.Run()
response(ctx, string(stdout.Bytes()), err)
}
func tokenError(ctx *fasthttp.RequestCtx) {
ctx.Response.SetBodyString("{code:0,data:\"token error\"}")
}
func checkRequest(ctx *fasthttp.RequestCtx) (ok bool) {
token := ctx.QueryArgs().Peek("api-token")
if token == nil || string(token) != apiToken {
ok = false
ctx.SetStatusCode(fasthttp.StatusInternalServerError)
return
}
ok = true
return
}
func timeoutFactory(h fasthttp.RequestHandler) fasthttp.RequestHandler {
return fasthttp.TimeoutHandler(h, apiTimeout, "timeout")
}
func response(ctx *fasthttp.RequestCtx, data interface{}, err error) {
callback := ctx.QueryArgs().Peek("callback")
callbackFunc := ""
if callback != nil && len(callback) > 0 {
callbackFunc = string(callback)
}
ja := gabs.New()
if err == nil {
ja.Set(1, "code")
ja.Set(data, "data")
} else {
ja.Set(0, "code")
ja.Set(err.Error(), "data")
}
if callbackFunc == "" {
fmt.Fprintf(ctx, ja.String())
} else {
fmt.Fprintf(ctx, callbackFunc+"("+ja.String()+")")
}
}
func serveAPI(listen, token string) (err error) {
ctx := log.With(logger.Fields{"func": "serveAPI"})
apiToken = token
router := fasthttprouter.New()
router.GET("/message/add", timeoutFactory(apiMessageAdd))
router.GET("/message/update", timeoutFactory(apiMessageUpdate))
router.GET("/message/delete", timeoutFactory(apiMessageDelete))
router.GET("/message/status", timeoutFactory(apiMessageStatus))
router.GET("/consumer/add", timeoutFactory(apiConsumerAdd))
router.GET("/consumer/update", timeoutFactory(apiConsumerUpdate))
router.GET("/consumer/delete", timeoutFactory(apiConsumerDelete))
router.GET("/consumer/status", timeoutFactory(apiConsumerStatus))
router.GET("/reload", timeoutFactory(apiReload))
router.GET("/restart", timeoutFactory(apiRestart))
router.GET("/config", timeoutFactory(apiConfig))
router.GET("/log", timeoutFactory(apiLog))
router.GET("/log/file", apiLogFile)
router.GET("/log/list", timeoutFactory(apiLogList))
ctx.Infof("Api service started")
var h = func(ctx *fasthttp.RequestCtx) {
defer access(ctx)
router.Handler(ctx)
}
if fasthttp.ListenAndServe(listen, h) == nil {
ctx.Safe().Fatalf("start api fail:%s", err)
}
return
}
func servePublish(listen string) (err error) {
ctx := log.With(logger.Fields{"func": "servePublish"})
router := fasthttprouter.New()
router.POST("/:name", timeoutFactory(apiPublish))
router.GET("/:name", timeoutFactory(apiPublish))
ctx.Infof("Publish service started")
var h = func(ctx *fasthttp.RequestCtx) {
defer access(ctx)
router.Handler(ctx)
}
if fasthttp.ListenAndServe(listen, h) == nil {
ctx.Safe().Fatalf("start publish fail:%s", err)
}
return
}
func access(ctx *fasthttp.RequestCtx) {
post := ""
if cfg.GetBool("log.post") {
post = string(ctx.Request.Body())
}
fields := logger.Fields{
"code": strconv.Itoa(ctx.Response.StatusCode()),
"uri": string(ctx.RequestURI()),
"remoteAddr": strings.Split(ctx.RemoteAddr().String(), ":")[0],
"method": string(ctx.Method()),
"host": string(ctx.Request.Host()),
"referer": string(ctx.Request.Header.Referer()),
"userAgent": string(ctx.Request.Header.UserAgent()),
"response": string(ctx.Response.Body()),
"post": post,
}
accessLog.With(fields).Info("")
}
此处可能存在不合适展示的内容，页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容，可点击提交进行申诉，我们将尽快为您处理。