Merge pull request #16 from Jinnrry/hotfix-async

v2.1.2

server/config/config.go

@@ -35,7 +35,7 @@ type Config struct {
 //go:embed tables/*
 var tableConfig embed.FS
 
-const Version = "2.1.1"
+const Version = "2.1.2"
 
 const DBTypeMySQL = "mysql"
 const DBTypeSQLite = "sqlite"

server/controllers/email/send.go

@@ -128,14 +128,16 @@ func Send(ctx *dto.Context, w http.ResponseWriter, req *http.Request) {
 
 	}
 
+	as := async.New(ctx)
 	for _, hook := range hooks.HookList {
 		if hook == nil {
 			continue
 		}
-		async.New(ctx).Process(func() {
-			hook.SendBefore(ctx, e)
-		})
+		as.WaitProcess(func(hk any) {
+			hk.(hooks.EmailHook).SendBefore(ctx, e)
+		}, hook)
 	}
+	as.Wait()
 
 	// 邮件落库
 	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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
@@ -166,18 +168,20 @@ func Send(ctx *dto.Context, w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	async.New(ctx).Process(func() {
+	async.New(ctx).Process(func(p any) {
 		errMsg := ""
 		err, sendErr := smtp_server.Send(ctx, e)
 
+		as2 := async.New(ctx)
 		for _, hook := range hooks.HookList {
 			if hook == nil {
 				continue
 			}
-			async.New(ctx).Process(func() {
-				hook.SendAfter(ctx, e, sendErr)
-			})
+			as2.WaitProcess(func(hk any) {
+				hk.(hooks.EmailHook).SendAfter(ctx, e, sendErr)
+			}, hook)
 		}
+		as2.Wait()
 
 		if err != nil {
 			errMsg = err.Error()
@@ -192,7 +196,7 @@ func Send(ctx *dto.Context, w http.ResponseWriter, req *http.Request) {
 			}
 		}
 
-	})
+	}, nil)
 
 	response.NewSuccessResponse(i18n.GetText(ctx.Lang, "succ")).FPrint(w)
 }

server/smtp_server/read_content.go

@@ -23,14 +23,16 @@ func (s *Session) Data(r io.Reader) error {
 		return err
 	}
 
+	as1 := async.New(nil)
 	for _, hook := range hooks.HookList {
 		if hook == nil {
 			continue
 		}
-		async.New(nil).Process(func() {
-			hook.ReceiveParseBefore(emailData)
-		})
+		as1.WaitProcess(func(hk any) {
+			hk.(hooks.EmailHook).ReceiveParseBefore(emailData)
+		}, hook)
 	}
+	as1.Wait()
 
 	log.Infof("邮件原始内容: %s", emailData)
 
@@ -55,14 +57,16 @@ func (s *Session) Data(r io.Reader) error {
 		spfV = 1
 	}
 
+	as2 := async.New(nil)
 	for _, hook := range hooks.HookList {
 		if hook == nil {
 			continue
 		}
-		async.New(nil).Process(func() {
-			hook.ReceiveParseAfter(email)
-		})
+		as2.WaitProcess(func(hk any) {
+			hk.(hooks.EmailHook).ReceiveParseAfter(email)
+		}, hook)
 	}
+	as2.Wait()
 
 	sql := "INSERT INTO email (send_date, subject, reply_to, from_name, from_address, `to`, bcc, cc, text, html, sender, attachments,spf_check, dkim_check, create_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
 	_, err = db.Instance.Exec(sql,

server/smtp_server/send.go

@@ -61,7 +61,7 @@ func Send(ctx *dto.Context, e *parsemail.Email) (error, map[string]error) {
 	for domain, tos := range toByDomain {
 		domain := domain
 		tos := tos
-		as.WaitProcess(func() {
+		as.WaitProcess(func(p any) {
 
 			err := smtp.SendMail("", domain.mxHost+":25", nil, e.From.EmailAddress, buildAddress(tos), b)
 
@@ -85,7 +85,7 @@ func Send(ctx *dto.Context, e *parsemail.Email) (error, map[string]error) {
 				}
 			}
 			errMap[domain.domain] = err
-		})
+		}, nil)
 	}
 	as.Wait()
 

server/utils/async/async.go

@@ -9,7 +9,7 @@ import (
 	"sync"
 )
 
-type Callback func()
+type Callback func(params any)
 
 type Async struct {
 	wg        *sync.WaitGroup
@@ -27,25 +27,25 @@ func (as *Async) LastError() error {
 	return as.lastError
 }
 
-func (as *Async) WaitProcess(callback Callback) {
+func (as *Async) WaitProcess(callback Callback, params any) {
 	if as.wg == nil {
 		as.wg = &sync.WaitGroup{}
 	}
 	as.wg.Add(1)
-	as.Process(func() {
+	as.Process(func(params any) {
 		defer as.wg.Done()
-		callback()
-	})
+		callback(params)
+	}, params)
 }
 
-func (as *Async) Process(callback Callback) {
+func (as *Async) Process(callback Callback, params any) {
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
 				as.lastError = as.HandleErrRecover(err)
 			}
 		}()
-		callback()
+		callback(params)
 	}()
 }