重构 event, 日志改用 stderr

Author: cupen

Makefile

@@ -8,13 +8,13 @@ install:
 
 
 test-integration:
-	sudo supervisorctl stop supervisor-event-listener
 	go build 
+	sudo supervisorctl stop supervisor-event-listener
 	sudo cp ./supervisor-event-listener /usr/local/bin/
 	sudo cp ./tests/supervisor-app.ini /etc/supervisor.d/
-	sudo supervisorctl start  supervisor-event-listener
-	sudo supervisorctl update
-	sudo supervisorctl start sleep-then-exit
+	sudo supervisorctl remove supervisor-event-listener
+	sudo supervisorctl update supervisor-event-listener
+	sudo supervisorctl tail -f supervisor-event-listener stderr
 
 
 clean:

config/config.go

@@ -6,7 +6,7 @@ import (
 	"strings"
 
 	"github.com/ouqiang/supervisor-event-listener/utils"
-	"github.com/ouqiang/supervisor-event-listener/utils/tmpfslog"
+	"github.com/ouqiang/supervisor-event-listener/utils/errlog"
 	"gopkg.in/ini.v1"
 )
 
@@ -66,7 +66,7 @@ func ParseConfig(configFile string) *Config {
 	config := &Config{}
 	config.NotifyType = notifyType
 
-	tmpfslog.Info("notifyType: %+v\n", config.NotifyType)
+	errlog.Info("notifyType: %+v\n", config.NotifyType)
 	switch notifyType {
 	case "mail":
 		config.MailServer = parseMailServer(section)

event/errs.go

@@ -0,0 +1,11 @@
+package event
+
+import (
+	"errors"
+)
+
+var (
+	ErrParseHeader   = errors.New("parse header failed")
+	ErrParsePayload  = errors.New("parse payload failed")
+	ErrPayloadLength = errors.New("invalid payload length")
+)

event/event.go

@@ -0,0 +1,31 @@
+package event
+
+import "strconv"
+
+type Header struct {
+	Ver        string
+	Server     string
+	Serial     int
+	Pool       string
+	PoolSerial int
+	EventName  string // 事件名称
+	Len        int    // Payload长度
+}
+
+func ParseHeader(header string) (*Header, error) {
+	h := &Header{}
+	fields := parseFields(header)
+	if len(fields) == 0 {
+		return h, ErrParseHeader
+	}
+
+	h.Ver = fields["ver"]
+	h.Server = fields["server"]
+	h.Serial, _ = strconv.Atoi(fields["serial"])
+	h.Pool = fields["pool"]
+	h.PoolSerial, _ = strconv.Atoi(fields["poolserial"])
+	h.EventName = fields["eventname"]
+	h.Len, _ = strconv.Atoi(fields["len"])
+
+	return h, nil
+}

event/header.go

@@ -0,0 +1,115 @@
+package event
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/ouqiang/supervisor-event-listener/utils/errlog"
+	"github.com/pkg/errors"
+)
+
+type Message struct {
+	TS      time.Time
+	Header  *Header
+	Payload *Payload
+}
+
+func NewMessage(h *Header, p *Payload) Message {
+	return Message{
+		TS:      time.Now(),
+		Header:  h,
+		Payload: p,
+	}
+}
+
+func ReadMessage(reader *bufio.Reader) (Message, error) {
+	header, err := readHeader(reader)
+	if err != nil {
+		errlog.Error("header:%+v err:%+v", header, err)
+		return Message{}, err
+	}
+	payload, err := readPayload(reader, header.Len)
+	if err != nil {
+		errlog.Error("payload:%+v err:%+v", payload, err)
+		return Message{}, err
+	}
+	return NewMessage(header, payload), nil
+}
+
+// 读取header
+func readHeader(reader *bufio.Reader) (*Header, error) {
+	// 读取Header
+	data, err := reader.ReadString('\n')
+	if err != nil {
+		return nil, err
+	}
+	// 解析Header
+	header, err := ParseHeader(data)
+	if err != nil {
+		return nil, err
+	}
+	return header, nil
+}
+
+// 读取payload
+func readPayload(reader *bufio.Reader, payloadLen int) (*Payload, error) {
+	// 读取payload
+	buf := make([]byte, payloadLen)
+	length, err := reader.Read(buf)
+	if err != nil {
+		return nil, err
+	}
+	if payloadLen != length {
+		err := ErrPayloadLength
+		err = errors.Wrapf(err, " payloadLen:%d != length:%d", payloadLen, length)
+		return nil, err
+	}
+	// 解析payload
+	payload, err := ParsePayload(string(buf))
+	if err != nil {
+		return nil, err
+	}
+	return payload, nil
+}
+
+func (msg *Message) String() string {
+	tmpl := `Proc:  %s
+Host:  %s
+State: %s
+PID:   %d
+Date:  %s`
+	return fmt.Sprintf(tmpl,
+		msg.Payload.ProcessName,
+		msg.Payload.IP,
+		msg.Payload.FromState,
+		msg.Payload.PID,
+		msg.TS.Format(time.RFC3339),
+	)
+}
+
+func (msg *Message) ToJson(indent ...int) string {
+	realIndent := 0
+	if len(indent) > 0 {
+		realIndent = indent[0]
+	}
+	t := ""
+	switch realIndent {
+	case 0:
+	case 1:
+		t = " "
+	case 2:
+		t = "  "
+	case 3:
+		t = "   "
+	case 4:
+		t = "    "
+	default:
+		t = "    "
+	}
+	_bytes, _ := json.MarshalIndent(msg, "", t)
+	return string(_bytes)
+}
+
+// Header Supervisord触发事件时会先发送Header，根据Header中len字段去读取Payload