⚡ Directly use multiple goroutines to speed up delivery speed

pkg/internal/grpc/postman.go

@@ -4,37 +4,50 @@ import (
 	"context"
 	"git.solsynth.dev/hydrogen/dealer/pkg/internal/services"
 	"git.solsynth.dev/hydrogen/dealer/pkg/proto"
+	"sync"
 )
 
 func (v *Server) DeliverNotification(ctx context.Context, request *proto.DeliverNotificationRequest) (*proto.DeliverResponse, error) {
-	services.PublishDeliveryTask(request)
+	services.DealDeliveryTask(request)
 	return &proto.DeliverResponse{}, nil
 }
 
 func (v *Server) DeliverNotificationBatch(ctx context.Context, request *proto.DeliverNotificationBatchRequest) (*proto.DeliverResponse, error) {
+	var wg sync.WaitGroup
 	for idx, provider := range request.GetProviders() {
 		token := request.GetDeviceTokens()[idx]
-		services.PublishDeliveryTask(&proto.DeliverNotificationRequest{
+		provider := provider
+		go func() {
+			wg.Add(1)
+			services.DealDeliveryTask(&proto.DeliverNotificationRequest{
 				Provider:    provider,
 				DeviceToken: token,
 				Notify:      request.GetNotify(),
 			})
+			wg.Done()
+		}()
 	}
 
 	return &proto.DeliverResponse{}, nil
 }
 
 func (v *Server) DeliverEmail(ctx context.Context, request *proto.DeliverEmailRequest) (*proto.DeliverResponse, error) {
-	services.PublishDeliveryTask(request)
+	services.DealDeliveryTask(request)
 	return &proto.DeliverResponse{}, nil
 }
 
 func (v *Server) DeliverEmailBatch(ctx context.Context, request *proto.DeliverEmailBatchRequest) (*proto.DeliverResponse, error) {
+	var wg sync.WaitGroup
 	for _, to := range request.GetTo() {
-		services.PublishDeliveryTask(&proto.DeliverEmailRequest{
+		to := to
+		go func() {
+			wg.Add(1)
+			services.DealDeliveryTask(&proto.DeliverEmailRequest{
 				To:    to,
 				Email: request.GetEmail(),
 			})
+			wg.Done()
+		}()
 	}
 
 	return &proto.DeliverResponse{}, nil

pkg/internal/services/postman.go

@@ -17,15 +17,7 @@ import (
 	"time"
 )
 
-var deliveryTaskQueue = make(chan any, 256)
-
-func PublishDeliveryTask(task any) {
-	deliveryTaskQueue <- task
-}
-
-func ConsumeDeliveryTasks() {
-	for {
-		task := <-deliveryTaskQueue
+func DealDeliveryTask(task any) {
 	switch tk := task.(type) {
 	case *proto.DeliverEmailRequest:
 		if tk.GetEmail().HtmlBody != nil {
@@ -42,7 +34,6 @@ func ConsumeDeliveryTasks() {
 		}
 	}
 }
-}
 
 func PushFirebaseNotify(token string, in *proto.NotifyRequest) error {
 	if ExtFire == nil {

pkg/main.go

@@ -41,11 +41,6 @@ func main() {
 		log.Warn().Err(err).Msg("An error occurred when setup APNs, apple notification push is unavailable...")
 	}
 
-	// Set up tasks queue consumers
-	for idx := 0; idx < max(1, viper.GetInt("performance.notification_deliver.worker_count")); idx++ {
-		go services.ConsumeDeliveryTasks()
-	}
-
 	// Server
 	go server.NewServer().Listen()
 

settings.toml

@@ -16,9 +16,6 @@ firebase_credentials = ""
 
 use_registration_magic_token = false
 
-[performance]
-notification_deliver.worker_count = 4
-
 [dealer]
 addr = "127.0.0.1:8442"