This commit is contained in:
jinnrry 2024-03-04 11:47:28 +08:00
parent de2d68a41c
commit 004fc91586
9 changed files with 54 additions and 19 deletions

View File

@ -129,6 +129,7 @@ func Send(ctx *context.Context, w http.ResponseWriter, req *http.Request) {
} }
log.WithContext(ctx).Debugf("插件执行--SendBefore")
as := async.New(ctx) as := async.New(ctx)
for _, hook := range hooks.HookList { for _, hook := range hooks.HookList {
if hook == nil { if hook == nil {
@ -139,6 +140,7 @@ func Send(ctx *context.Context, w http.ResponseWriter, req *http.Request) {
}, hook) }, hook)
} }
as.Wait() as.Wait()
log.WithContext(ctx).Debugf("插件执行--SendBefore End")
// 邮件落库 // 邮件落库
sql := "INSERT INTO email (type,subject, reply_to, from_name, from_address, `to`, bcc, cc, text, html, sender, attachments,spf_check, dkim_check, create_time,send_user_id,error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" sql := "INSERT INTO email (type,subject, reply_to, from_name, from_address, `to`, bcc, cc, text, html, sender, attachments,spf_check, dkim_check, create_time,send_user_id,error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
@ -175,6 +177,8 @@ func Send(ctx *context.Context, w http.ResponseWriter, req *http.Request) {
errMsg := "" errMsg := ""
err, sendErr := send.Send(ctx, e) err, sendErr := send.Send(ctx, e)
log.WithContext(ctx).Debugf("插件执行--SendAfter")
as2 := async.New(ctx) as2 := async.New(ctx)
for _, hook := range hooks.HookList { for _, hook := range hooks.HookList {
if hook == nil { if hook == nil {
@ -185,6 +189,7 @@ func Send(ctx *context.Context, w http.ResponseWriter, req *http.Request) {
}, hook) }, hook)
} }
as2.Wait() as2.Wait()
log.WithContext(ctx).Debugf("插件执行--SendAfter")
if err != nil { if err != nil {
errMsg = err.Error() errMsg = err.Error()

View File

@ -27,6 +27,7 @@ type HookSender struct {
} }
func (h *HookSender) SendBefore(ctx *context.Context, email *parsemail.Email) { func (h *HookSender) SendBefore(ctx *context.Context, email *parsemail.Email) {
log.WithContext(ctx).Debugf("[%s]Plugin SendBefore Start", h.name)
dto := framework.HookDTO{ dto := framework.HookDTO{
Ctx: ctx, Ctx: ctx,
@ -45,9 +46,12 @@ func (h *HookSender) SendBefore(ctx *context.Context, email *parsemail.Email) {
ctx = dto.Ctx ctx = dto.Ctx
email = dto.Email email = dto.Email
log.WithContext(ctx).Debugf("[%s]Plugin SendBefore End", h.name)
} }
func (h *HookSender) SendAfter(ctx *context.Context, email *parsemail.Email, err map[string]error) { func (h *HookSender) SendAfter(ctx *context.Context, email *parsemail.Email, err map[string]error) {
log.WithContext(ctx).Debugf("[%s]Plugin SendAfter Start", h.name)
dto := framework.HookDTO{ dto := framework.HookDTO{
Ctx: ctx, Ctx: ctx,
Email: email, Email: email,
@ -67,9 +71,13 @@ func (h *HookSender) SendAfter(ctx *context.Context, email *parsemail.Email, err
ctx = dto.Ctx ctx = dto.Ctx
email = dto.Email email = dto.Email
err = dto.ErrMap err = dto.ErrMap
log.WithContext(ctx).Debugf("[%s]Plugin SendAfter End", h.name)
} }
func (h *HookSender) ReceiveParseBefore(ctx *context.Context, email *[]byte) { func (h *HookSender) ReceiveParseBefore(ctx *context.Context, email *[]byte) {
log.WithContext(ctx).Debugf("[%s]Plugin ReceiveParseBefore Start", h.name)
dto := framework.HookDTO{ dto := framework.HookDTO{
Ctx: ctx, Ctx: ctx,
EmailByte: email, EmailByte: email,
@ -87,9 +95,13 @@ func (h *HookSender) ReceiveParseBefore(ctx *context.Context, email *[]byte) {
ctx = dto.Ctx ctx = dto.Ctx
email = dto.EmailByte email = dto.EmailByte
log.WithContext(ctx).Debugf("[%s]Plugin ReceiveParseBefore End", h.name)
} }
func (h *HookSender) ReceiveParseAfter(ctx *context.Context, email *parsemail.Email) { func (h *HookSender) ReceiveParseAfter(ctx *context.Context, email *parsemail.Email) {
log.WithContext(ctx).Debugf("[%s]Plugin ReceiveParseAfter Start", h.name)
dto := framework.HookDTO{ dto := framework.HookDTO{
Ctx: ctx, Ctx: ctx,
Email: email, Email: email,
@ -107,9 +119,11 @@ func (h *HookSender) ReceiveParseAfter(ctx *context.Context, email *parsemail.Em
ctx = dto.Ctx ctx = dto.Ctx
email = dto.Email email = dto.Email
log.WithContext(ctx).Debugf("[%s]Plugin ReceiveParseAfter End", h.name)
} }
func NewHookSender(socketPath string, name string) *HookSender { func NewHookSender(socketPath string, name string, serverVersion string) *HookSender {
httpc := http.Client{ httpc := http.Client{
Timeout: time.Second * 10, Timeout: time.Second * 10,
Transport: &http.Transport{ Transport: &http.Transport{
@ -126,7 +140,7 @@ func NewHookSender(socketPath string, name string) *HookSender {
} }
// Init 注册hook对象 // Init 注册hook对象
func Init() { func Init(serverVersion string) {
env := os.Environ() env := os.Environ()
procAttr := &os.ProcAttr{ procAttr := &os.ProcAttr{
@ -161,7 +175,7 @@ func Init() {
pluginNo++ pluginNo++
HookList = append(HookList, NewHookSender(socketPath, info.Name())) HookList = append(HookList, NewHookSender(socketPath, info.Name(), serverVersion))
go func() { go func() {
stat, err := p.Wait() stat, err := p.Wait()

View File

@ -10,6 +10,7 @@ import (
"path/filepath" "path/filepath"
"pmail/dto/parsemail" "pmail/dto/parsemail"
"pmail/utils/context" "pmail/utils/context"
"time"
) )
type EmailHook interface { type EmailHook interface {
@ -23,19 +24,23 @@ type EmailHook interface {
ReceiveParseAfter(ctx *context.Context, email *parsemail.Email) ReceiveParseAfter(ctx *context.Context, email *parsemail.Email)
} }
// HookDTO PMail 主程序和插件通信的结构体
type HookDTO struct { type HookDTO struct {
Ctx *context.Context ServerVersion string // 服务端程序版本
Email *parsemail.Email Ctx *context.Context // 上下文
EmailByte *[]byte Email *parsemail.Email // 邮件内容
ErrMap map[string]error EmailByte *[]byte // 未解析前的文件内容
ErrMap map[string]error // 错误信息
} }
type Plugin struct { type Plugin struct {
name string
hook EmailHook hook EmailHook
} }
func CreatePlugin(hook EmailHook) *Plugin { func CreatePlugin(name string, hook EmailHook) *Plugin {
return &Plugin{ return &Plugin{
name: name,
hook: hook, hook: hook,
} }
} }
@ -47,6 +52,7 @@ func (p *Plugin) Run() {
mux := http.NewServeMux() mux := http.NewServeMux()
mux.HandleFunc("/SendBefore", func(writer http.ResponseWriter, request *http.Request) { mux.HandleFunc("/SendBefore", func(writer http.ResponseWriter, request *http.Request) {
log.Debugf("[%s] SendBefore Start", p.name)
var hookDTO HookDTO var hookDTO HookDTO
body, _ := io.ReadAll(request.Body) body, _ := io.ReadAll(request.Body)
err := json.Unmarshal(body, &hookDTO) err := json.Unmarshal(body, &hookDTO)
@ -57,8 +63,10 @@ func (p *Plugin) Run() {
p.hook.SendBefore(hookDTO.Ctx, hookDTO.Email) p.hook.SendBefore(hookDTO.Ctx, hookDTO.Email)
body, _ = json.Marshal(hookDTO) body, _ = json.Marshal(hookDTO)
writer.Write(body) writer.Write(body)
log.Debugf("[%s] SendBefore End", p.name)
}) })
mux.HandleFunc("/SendAfter", func(writer http.ResponseWriter, request *http.Request) { mux.HandleFunc("/SendAfter", func(writer http.ResponseWriter, request *http.Request) {
log.Debugf("[%s] SendAfter Start", p.name)
var hookDTO HookDTO var hookDTO HookDTO
body, _ := io.ReadAll(request.Body) body, _ := io.ReadAll(request.Body)
@ -70,9 +78,11 @@ func (p *Plugin) Run() {
p.hook.SendAfter(hookDTO.Ctx, hookDTO.Email, hookDTO.ErrMap) p.hook.SendAfter(hookDTO.Ctx, hookDTO.Email, hookDTO.ErrMap)
body, _ = json.Marshal(hookDTO) body, _ = json.Marshal(hookDTO)
writer.Write(body) writer.Write(body)
log.Debugf("[%s] SendAfter End", p.name)
}) })
mux.HandleFunc("/ReceiveParseBefore", func(writer http.ResponseWriter, request *http.Request) { mux.HandleFunc("/ReceiveParseBefore", func(writer http.ResponseWriter, request *http.Request) {
log.Debugf("[%s] ReceiveParseBefore Start", p.name)
var hookDTO HookDTO var hookDTO HookDTO
body, _ := io.ReadAll(request.Body) body, _ := io.ReadAll(request.Body)
err := json.Unmarshal(body, &hookDTO) err := json.Unmarshal(body, &hookDTO)
@ -83,9 +93,10 @@ func (p *Plugin) Run() {
p.hook.ReceiveParseBefore(hookDTO.Ctx, hookDTO.EmailByte) p.hook.ReceiveParseBefore(hookDTO.Ctx, hookDTO.EmailByte)
body, _ = json.Marshal(hookDTO) body, _ = json.Marshal(hookDTO)
writer.Write(body) writer.Write(body)
log.Debugf("[%s] ReceiveParseBefore End", p.name)
}) })
mux.HandleFunc("/ReceiveParseAfter", func(writer http.ResponseWriter, request *http.Request) { mux.HandleFunc("/ReceiveParseAfter", func(writer http.ResponseWriter, request *http.Request) {
log.Debugf("[%s] ReceiveParseAfter Start", p.name)
var hookDTO HookDTO var hookDTO HookDTO
body, _ := io.ReadAll(request.Body) body, _ := io.ReadAll(request.Body)
err := json.Unmarshal(body, &hookDTO) err := json.Unmarshal(body, &hookDTO)
@ -96,10 +107,13 @@ func (p *Plugin) Run() {
p.hook.ReceiveParseAfter(hookDTO.Ctx, hookDTO.Email) p.hook.ReceiveParseAfter(hookDTO.Ctx, hookDTO.Email)
body, _ = json.Marshal(hookDTO) body, _ = json.Marshal(hookDTO)
writer.Write(body) writer.Write(body)
log.Debugf("[%s] ReceiveParseAfter End", p.name)
}) })
server := http.Server{ server := http.Server{
Handler: mux, ReadTimeout: 5 * time.Second,
WriteTimeout: 5 * time.Second,
Handler: mux,
} }
unixListener, err := net.Listen("unix", getExePath()+"/"+os.Args[1]) unixListener, err := net.Listen("unix", getExePath()+"/"+os.Args[1])

View File

@ -140,5 +140,5 @@ func NewTelegramPushHook() *TelegramPushHook {
} }
func main() { func main() {
framework.CreatePlugin(NewTelegramPushHook()).Run() framework.CreatePlugin("telegram_push", NewTelegramPushHook()).Run()
} }

View File

@ -129,5 +129,5 @@ func NewWebPushHook() *WebPushHook {
} }
func main() { func main() {
framework.CreatePlugin(NewWebPushHook()).Run() framework.CreatePlugin("web_push", NewWebPushHook()).Run()
} }

View File

@ -176,5 +176,5 @@ func NewWechatPushHook() *WeChatPushHook {
// 插件将以独立进程运行,因此需要主函数。 // 插件将以独立进程运行,因此需要主函数。
func main() { func main() {
framework.CreatePlugin(NewWechatPushHook()).Run() framework.CreatePlugin("wechat_push", NewWechatPushHook()).Run()
} }

View File

@ -89,7 +89,7 @@ func main() {
go cron_server.Start() go cron_server.Start()
// 核心服务启动 // 核心服务启动
res_init.Init() res_init.Init(version)
s := make(chan bool) s := make(chan bool)
<-s <-s

View File

@ -17,7 +17,7 @@ import (
"pmail/utils/file" "pmail/utils/file"
) )
func Init() { func Init(serverVersion string) {
if !config.IsInit { if !config.IsInit {
dirInit() dirInit()
@ -38,7 +38,7 @@ func Init() {
panic(err) panic(err)
} }
session.Init() session.Init()
hooks.Init() hooks.Init(serverVersion)
// smtp server start // smtp server start
go smtp_server.Start() go smtp_server.Start()
go smtp_server.StartWithTLS() go smtp_server.StartWithTLS()

View File

@ -32,7 +32,7 @@ func (s *Session) Data(r io.Reader) error {
log.WithContext(ctx).Error("邮件内容无法读取", err) log.WithContext(ctx).Error("邮件内容无法读取", err)
return err return err
} }
log.WithContext(ctx).Debugf("开始执行插件ReceiveParseBefore!")
as1 := async.New(ctx) as1 := async.New(ctx)
for _, hook := range hooks.HookList { for _, hook := range hooks.HookList {
if hook == nil { if hook == nil {
@ -43,6 +43,7 @@ func (s *Session) Data(r io.Reader) error {
}, hook) }, hook)
} }
as1.Wait() as1.Wait()
log.WithContext(ctx).Debugf("开始执行插件ReceiveParseBefore End!")
log.WithContext(ctx).Infof("邮件原始内容: %s", emailData) log.WithContext(ctx).Infof("邮件原始内容: %s", emailData)
@ -87,7 +88,7 @@ func (s *Session) Data(r io.Reader) error {
saveEmail(ctx, email, 0, SPFStatus, dkimStatus) saveEmail(ctx, email, 0, SPFStatus, dkimStatus)
log.WithContext(ctx).Debugf("开始执行插件!") log.WithContext(ctx).Debugf("开始执行插件ReceiveParseAfter!")
as2 := async.New(ctx) as2 := async.New(ctx)
for _, hook := range hooks.HookList { for _, hook := range hooks.HookList {
@ -99,6 +100,7 @@ func (s *Session) Data(r io.Reader) error {
}, hook) }, hook)
} }
as2.Wait() as2.Wait()
log.WithContext(ctx).Debugf("开始执行插件ReceiveParseAfter!End")
log.WithContext(ctx).Debugf("开始执行邮件规则!") log.WithContext(ctx).Debugf("开始执行邮件规则!")
// 执行邮件规则 // 执行邮件规则