Only initial MaxConcurrentIOSPushes once. (#591)
This commit is contained in:
		
							parent
							
								
									ab8b1991ab
								
							
						
					
					
						commit
						73ff554b19
					
				| 
						 | 
				
			
			@ -57,6 +57,7 @@ type RequestPush struct {
 | 
			
		|||
type PushNotification struct {
 | 
			
		||||
	Wg  *sync.WaitGroup
 | 
			
		||||
	Log *[]logx.LogPushEntry
 | 
			
		||||
	Cfg config.ConfYaml
 | 
			
		||||
 | 
			
		||||
	// Common
 | 
			
		||||
	ID               string      `json:"notif_id,omitempty"`
 | 
			
		||||
| 
						 | 
				
			
			@ -238,17 +239,17 @@ func CheckPushConf(cfg config.ConfYaml) error {
 | 
			
		|||
}
 | 
			
		||||
 | 
			
		||||
// SendNotification send notification
 | 
			
		||||
func SendNotification(cfg config.ConfYaml, req PushNotification) {
 | 
			
		||||
func SendNotification(req PushNotification) {
 | 
			
		||||
	defer func() {
 | 
			
		||||
		req.WaitDone()
 | 
			
		||||
	}()
 | 
			
		||||
 | 
			
		||||
	switch req.Platform {
 | 
			
		||||
	case core.PlatFormIos:
 | 
			
		||||
		PushToIOS(cfg, req)
 | 
			
		||||
		PushToIOS(req)
 | 
			
		||||
	case core.PlatFormAndroid:
 | 
			
		||||
		PushToAndroid(cfg, req)
 | 
			
		||||
		PushToAndroid(req)
 | 
			
		||||
	case core.PlatFormHuawei:
 | 
			
		||||
		PushToHuawei(cfg, req)
 | 
			
		||||
		PushToHuawei(req)
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -31,6 +31,8 @@ var (
 | 
			
		|||
	tcpKeepAlive    = 60 * time.Second
 | 
			
		||||
)
 | 
			
		||||
 | 
			
		||||
var doOnce sync.Once
 | 
			
		||||
 | 
			
		||||
// DialTLS is the default dial function for creating TLS connections for
 | 
			
		||||
// non-proxied HTTPS requests.
 | 
			
		||||
var DialTLS = func(cfg *tls.Config) func(network, addr string) (net.Conn, error) {
 | 
			
		||||
| 
						 | 
				
			
			@ -132,7 +134,9 @@ func InitAPNSClient(cfg config.ConfYaml) error {
 | 
			
		|||
			return err
 | 
			
		||||
		}
 | 
			
		||||
 | 
			
		||||
		MaxConcurrentIOSPushes = make(chan struct{}, cfg.Ios.MaxConcurrentPushes)
 | 
			
		||||
		doOnce.Do(func() {
 | 
			
		||||
			MaxConcurrentIOSPushes = make(chan struct{}, cfg.Ios.MaxConcurrentPushes)
 | 
			
		||||
		})
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	return nil
 | 
			
		||||
| 
						 | 
				
			
			@ -384,12 +388,12 @@ func getApnsClient(cfg config.ConfYaml, req PushNotification) (client *apns2.Cli
 | 
			
		|||
}
 | 
			
		||||
 | 
			
		||||
// PushToIOS provide send notification to APNs server.
 | 
			
		||||
func PushToIOS(cfg config.ConfYaml, req PushNotification) {
 | 
			
		||||
func PushToIOS(req PushNotification) {
 | 
			
		||||
	logx.LogAccess.Debug("Start push notification for iOS")
 | 
			
		||||
 | 
			
		||||
	var (
 | 
			
		||||
		retryCount = 0
 | 
			
		||||
		maxRetry   = cfg.Ios.MaxRetry
 | 
			
		||||
		maxRetry   = req.Cfg.Ios.MaxRetry
 | 
			
		||||
	)
 | 
			
		||||
 | 
			
		||||
	if req.Retry > 0 && req.Retry < maxRetry {
 | 
			
		||||
| 
						 | 
				
			
			@ -400,7 +404,7 @@ Retry:
 | 
			
		|||
	var newTokens []string
 | 
			
		||||
 | 
			
		||||
	notification := GetIOSNotification(req)
 | 
			
		||||
	client := getApnsClient(cfg, req)
 | 
			
		||||
	client := getApnsClient(req.Cfg, req)
 | 
			
		||||
 | 
			
		||||
	var wg sync.WaitGroup
 | 
			
		||||
	for _, token := range req.Tokens {
 | 
			
		||||
| 
						 | 
				
			
			@ -419,17 +423,17 @@ Retry:
 | 
			
		|||
					err = errors.New(res.Reason)
 | 
			
		||||
				}
 | 
			
		||||
				// apns server error
 | 
			
		||||
				logPush(cfg, core.FailedPush, token, req, err)
 | 
			
		||||
				logPush(req.Cfg, core.FailedPush, token, req, err)
 | 
			
		||||
 | 
			
		||||
				if cfg.Core.Sync {
 | 
			
		||||
					req.AddLog(createLogPushEntry(cfg, core.FailedPush, token, req, err))
 | 
			
		||||
				} else if cfg.Core.FeedbackURL != "" {
 | 
			
		||||
				if req.Cfg.Core.Sync {
 | 
			
		||||
					req.AddLog(createLogPushEntry(req.Cfg, core.FailedPush, token, req, err))
 | 
			
		||||
				} else if req.Cfg.Core.FeedbackURL != "" {
 | 
			
		||||
					go func(logger *logrus.Logger, log logx.LogPushEntry, url string, timeout int64) {
 | 
			
		||||
						err := DispatchFeedback(log, url, timeout)
 | 
			
		||||
						if err != nil {
 | 
			
		||||
							logger.Error(err)
 | 
			
		||||
						}
 | 
			
		||||
					}(logx.LogError, createLogPushEntry(cfg, core.FailedPush, token, req, err), cfg.Core.FeedbackURL, cfg.Core.FeedbackTimeout)
 | 
			
		||||
					}(logx.LogError, createLogPushEntry(req.Cfg, core.FailedPush, token, req, err), req.Cfg.Core.FeedbackURL, req.Cfg.Core.FeedbackTimeout)
 | 
			
		||||
				}
 | 
			
		||||
 | 
			
		||||
				status.StatStorage.AddIosError(1)
 | 
			
		||||
| 
						 | 
				
			
			@ -441,14 +445,16 @@ Retry:
 | 
			
		|||
			}
 | 
			
		||||
 | 
			
		||||
			if res != nil && res.Sent() {
 | 
			
		||||
				logPush(cfg, core.SucceededPush, token, req, nil)
 | 
			
		||||
				logPush(req.Cfg, core.SucceededPush, token, req, nil)
 | 
			
		||||
				status.StatStorage.AddIosSuccess(1)
 | 
			
		||||
			}
 | 
			
		||||
 | 
			
		||||
			// free push slot
 | 
			
		||||
			<-MaxConcurrentIOSPushes
 | 
			
		||||
			wg.Done()
 | 
			
		||||
		}(*notification, token)
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	wg.Wait()
 | 
			
		||||
 | 
			
		||||
	if len(newTokens) > 0 && retryCount < maxRetry {
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -727,13 +727,14 @@ func TestPushToIOS(t *testing.T) {
 | 
			
		|||
	assert.Nil(t, err)
 | 
			
		||||
 | 
			
		||||
	req := PushNotification{
 | 
			
		||||
		Cfg:      cfg,
 | 
			
		||||
		Tokens:   []string{"11aa01229f15f0f0c52029d8cf8cd0aeaf2365fe4cebc4af26cd6d76b7919ef7", "11aa01229f15f0f0c52029d8cf8cd0aeaf2365fe4cebc4af26cd6d76b7919ef1"},
 | 
			
		||||
		Platform: 1,
 | 
			
		||||
		Message:  "Welcome",
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	// send fail
 | 
			
		||||
	PushToIOS(cfg, req)
 | 
			
		||||
	PushToIOS(req)
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
func TestApnsHostFromRequest(t *testing.T) {
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -106,13 +106,13 @@ func GetAndroidNotification(req PushNotification) *fcm.Message {
 | 
			
		|||
}
 | 
			
		||||
 | 
			
		||||
// PushToAndroid provide send notification to Android server.
 | 
			
		||||
func PushToAndroid(cfg config.ConfYaml, req PushNotification) {
 | 
			
		||||
func PushToAndroid(req PushNotification) {
 | 
			
		||||
	logx.LogAccess.Debug("Start push notification for Android")
 | 
			
		||||
 | 
			
		||||
	var (
 | 
			
		||||
		client     *fcm.Client
 | 
			
		||||
		retryCount = 0
 | 
			
		||||
		maxRetry   = cfg.Android.MaxRetry
 | 
			
		||||
		maxRetry   = req.Cfg.Android.MaxRetry
 | 
			
		||||
	)
 | 
			
		||||
 | 
			
		||||
	if req.Retry > 0 && req.Retry < maxRetry {
 | 
			
		||||
| 
						 | 
				
			
			@ -130,9 +130,9 @@ Retry:
 | 
			
		|||
	notification := GetAndroidNotification(req)
 | 
			
		||||
 | 
			
		||||
	if req.APIKey != "" {
 | 
			
		||||
		client, err = InitFCMClient(cfg, req.APIKey)
 | 
			
		||||
		client, err = InitFCMClient(req.Cfg, req.APIKey)
 | 
			
		||||
	} else {
 | 
			
		||||
		client, err = InitFCMClient(cfg, cfg.Android.APIKey)
 | 
			
		||||
		client, err = InitFCMClient(req.Cfg, req.Cfg.Android.APIKey)
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	if err != nil {
 | 
			
		||||
| 
						 | 
				
			
			@ -147,28 +147,28 @@ Retry:
 | 
			
		|||
		logx.LogError.Error("FCM server send message error: " + err.Error())
 | 
			
		||||
 | 
			
		||||
		if req.IsTopic() {
 | 
			
		||||
			if cfg.Core.Sync {
 | 
			
		||||
				req.AddLog(createLogPushEntry(cfg, core.FailedPush, req.To, req, err))
 | 
			
		||||
			} else if cfg.Core.FeedbackURL != "" {
 | 
			
		||||
			if req.Cfg.Core.Sync {
 | 
			
		||||
				req.AddLog(createLogPushEntry(req.Cfg, core.FailedPush, req.To, req, err))
 | 
			
		||||
			} else if req.Cfg.Core.FeedbackURL != "" {
 | 
			
		||||
				go func(logger *logrus.Logger, log logx.LogPushEntry, url string, timeout int64) {
 | 
			
		||||
					err := DispatchFeedback(log, url, timeout)
 | 
			
		||||
					if err != nil {
 | 
			
		||||
						logger.Error(err)
 | 
			
		||||
					}
 | 
			
		||||
				}(logx.LogError, createLogPushEntry(cfg, core.FailedPush, req.To, req, err), cfg.Core.FeedbackURL, cfg.Core.FeedbackTimeout)
 | 
			
		||||
				}(logx.LogError, createLogPushEntry(req.Cfg, core.FailedPush, req.To, req, err), req.Cfg.Core.FeedbackURL, req.Cfg.Core.FeedbackTimeout)
 | 
			
		||||
			}
 | 
			
		||||
			status.StatStorage.AddAndroidError(1)
 | 
			
		||||
		} else {
 | 
			
		||||
			for _, token := range req.Tokens {
 | 
			
		||||
				if cfg.Core.Sync {
 | 
			
		||||
					req.AddLog(createLogPushEntry(cfg, core.FailedPush, token, req, err))
 | 
			
		||||
				} else if cfg.Core.FeedbackURL != "" {
 | 
			
		||||
				if req.Cfg.Core.Sync {
 | 
			
		||||
					req.AddLog(createLogPushEntry(req.Cfg, core.FailedPush, token, req, err))
 | 
			
		||||
				} else if req.Cfg.Core.FeedbackURL != "" {
 | 
			
		||||
					go func(logger *logrus.Logger, log logx.LogPushEntry, url string, timeout int64) {
 | 
			
		||||
						err := DispatchFeedback(log, url, timeout)
 | 
			
		||||
						if err != nil {
 | 
			
		||||
							logger.Error(err)
 | 
			
		||||
						}
 | 
			
		||||
					}(logx.LogError, createLogPushEntry(cfg, core.FailedPush, token, req, err), cfg.Core.FeedbackURL, cfg.Core.FeedbackTimeout)
 | 
			
		||||
					}(logx.LogError, createLogPushEntry(req.Cfg, core.FailedPush, token, req, err), req.Cfg.Core.FeedbackURL, req.Cfg.Core.FeedbackTimeout)
 | 
			
		||||
				}
 | 
			
		||||
			}
 | 
			
		||||
			status.StatStorage.AddAndroidError(int64(len(req.Tokens)))
 | 
			
		||||
| 
						 | 
				
			
			@ -200,21 +200,21 @@ Retry:
 | 
			
		|||
				newTokens = append(newTokens, to)
 | 
			
		||||
			}
 | 
			
		||||
 | 
			
		||||
			logPush(cfg, core.FailedPush, to, req, result.Error)
 | 
			
		||||
			if cfg.Core.Sync {
 | 
			
		||||
				req.AddLog(createLogPushEntry(cfg, core.FailedPush, to, req, result.Error))
 | 
			
		||||
			} else if cfg.Core.FeedbackURL != "" {
 | 
			
		||||
			logPush(req.Cfg, core.FailedPush, to, req, result.Error)
 | 
			
		||||
			if req.Cfg.Core.Sync {
 | 
			
		||||
				req.AddLog(createLogPushEntry(req.Cfg, core.FailedPush, to, req, result.Error))
 | 
			
		||||
			} else if req.Cfg.Core.FeedbackURL != "" {
 | 
			
		||||
				go func(logger *logrus.Logger, log logx.LogPushEntry, url string, timeout int64) {
 | 
			
		||||
					err := DispatchFeedback(log, url, timeout)
 | 
			
		||||
					if err != nil {
 | 
			
		||||
						logger.Error(err)
 | 
			
		||||
					}
 | 
			
		||||
				}(logx.LogError, createLogPushEntry(cfg, core.FailedPush, to, req, result.Error), cfg.Core.FeedbackURL, cfg.Core.FeedbackTimeout)
 | 
			
		||||
				}(logx.LogError, createLogPushEntry(req.Cfg, core.FailedPush, to, req, result.Error), req.Cfg.Core.FeedbackURL, req.Cfg.Core.FeedbackTimeout)
 | 
			
		||||
			}
 | 
			
		||||
			continue
 | 
			
		||||
		}
 | 
			
		||||
 | 
			
		||||
		logPush(cfg, core.SucceededPush, to, req, nil)
 | 
			
		||||
		logPush(req.Cfg, core.SucceededPush, to, req, nil)
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	// result from Send messages to topics
 | 
			
		||||
| 
						 | 
				
			
			@ -228,12 +228,12 @@ Retry:
 | 
			
		|||
		logx.LogAccess.Debug("Send Topic Message: ", to)
 | 
			
		||||
		// Success
 | 
			
		||||
		if res.MessageID != 0 {
 | 
			
		||||
			logPush(cfg, core.SucceededPush, to, req, nil)
 | 
			
		||||
			logPush(req.Cfg, core.SucceededPush, to, req, nil)
 | 
			
		||||
		} else {
 | 
			
		||||
			// failure
 | 
			
		||||
			logPush(cfg, core.FailedPush, to, req, res.Error)
 | 
			
		||||
			if cfg.Core.Sync {
 | 
			
		||||
				req.AddLog(createLogPushEntry(cfg, core.FailedPush, to, req, res.Error))
 | 
			
		||||
			logPush(req.Cfg, core.FailedPush, to, req, res.Error)
 | 
			
		||||
			if req.Cfg.Core.Sync {
 | 
			
		||||
				req.AddLog(createLogPushEntry(req.Cfg, core.FailedPush, to, req, res.Error))
 | 
			
		||||
			}
 | 
			
		||||
		}
 | 
			
		||||
	}
 | 
			
		||||
| 
						 | 
				
			
			@ -242,9 +242,9 @@ Retry:
 | 
			
		|||
	if len(res.FailedRegistrationIDs) > 0 {
 | 
			
		||||
		newTokens = append(newTokens, res.FailedRegistrationIDs...)
 | 
			
		||||
 | 
			
		||||
		logPush(cfg, core.FailedPush, notification.To, req, errors.New("device group: partial success or all fails"))
 | 
			
		||||
		if cfg.Core.Sync {
 | 
			
		||||
			req.AddLog(createLogPushEntry(cfg, core.FailedPush, notification.To, req, errors.New("device group: partial success or all fails")))
 | 
			
		||||
		logPush(req.Cfg, core.FailedPush, notification.To, req, errors.New("device group: partial success or all fails"))
 | 
			
		||||
		if req.Cfg.Core.Sync {
 | 
			
		||||
			req.AddLog(createLogPushEntry(req.Cfg, core.FailedPush, notification.To, req, errors.New("device group: partial success or all fails")))
 | 
			
		||||
		}
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -41,13 +41,14 @@ func TestPushToAndroidWrongToken(t *testing.T) {
 | 
			
		|||
	cfg.Android.APIKey = os.Getenv("ANDROID_API_KEY")
 | 
			
		||||
 | 
			
		||||
	req := PushNotification{
 | 
			
		||||
		Cfg:      cfg,
 | 
			
		||||
		Tokens:   []string{"aaaaaa", "bbbbb"},
 | 
			
		||||
		Platform: core.PlatFormAndroid,
 | 
			
		||||
		Message:  "Welcome",
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	// Android Success count: 0, Failure count: 2
 | 
			
		||||
	PushToAndroid(cfg, req)
 | 
			
		||||
	PushToAndroid(req)
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
func TestPushToAndroidRightTokenForJSONLog(t *testing.T) {
 | 
			
		||||
| 
						 | 
				
			
			@ -61,12 +62,13 @@ func TestPushToAndroidRightTokenForJSONLog(t *testing.T) {
 | 
			
		|||
	androidToken := os.Getenv("ANDROID_TEST_TOKEN")
 | 
			
		||||
 | 
			
		||||
	req := PushNotification{
 | 
			
		||||
		Cfg:      cfg,
 | 
			
		||||
		Tokens:   []string{androidToken},
 | 
			
		||||
		Platform: core.PlatFormAndroid,
 | 
			
		||||
		Message:  "Welcome",
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	PushToAndroid(cfg, req)
 | 
			
		||||
	PushToAndroid(req)
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
func TestPushToAndroidRightTokenForStringLog(t *testing.T) {
 | 
			
		||||
| 
						 | 
				
			
			@ -78,12 +80,13 @@ func TestPushToAndroidRightTokenForStringLog(t *testing.T) {
 | 
			
		|||
	androidToken := os.Getenv("ANDROID_TEST_TOKEN")
 | 
			
		||||
 | 
			
		||||
	req := PushNotification{
 | 
			
		||||
		Cfg:      cfg,
 | 
			
		||||
		Tokens:   []string{androidToken},
 | 
			
		||||
		Platform: core.PlatFormAndroid,
 | 
			
		||||
		Message:  "Welcome",
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	PushToAndroid(cfg, req)
 | 
			
		||||
	PushToAndroid(req)
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
func TestOverwriteAndroidAPIKey(t *testing.T) {
 | 
			
		||||
| 
						 | 
				
			
			@ -96,6 +99,7 @@ func TestOverwriteAndroidAPIKey(t *testing.T) {
 | 
			
		|||
	androidToken := os.Getenv("ANDROID_TEST_TOKEN")
 | 
			
		||||
 | 
			
		||||
	req := PushNotification{
 | 
			
		||||
		Cfg:      cfg,
 | 
			
		||||
		Tokens:   []string{androidToken, "bbbbb"},
 | 
			
		||||
		Platform: core.PlatFormAndroid,
 | 
			
		||||
		Message:  "Welcome",
 | 
			
		||||
| 
						 | 
				
			
			@ -106,7 +110,7 @@ func TestOverwriteAndroidAPIKey(t *testing.T) {
 | 
			
		|||
	}
 | 
			
		||||
 | 
			
		||||
	// FCM server error: 401 error: 401 Unauthorized (Wrong API Key)
 | 
			
		||||
	PushToAndroid(cfg, req)
 | 
			
		||||
	PushToAndroid(req)
 | 
			
		||||
 | 
			
		||||
	assert.Len(t, *req.Log, 2)
 | 
			
		||||
}
 | 
			
		||||
| 
						 | 
				
			
			@ -197,13 +201,14 @@ func TestCheckAndroidMessage(t *testing.T) {
 | 
			
		|||
 | 
			
		||||
	timeToLive := uint(2419201)
 | 
			
		||||
	req := PushNotification{
 | 
			
		||||
		Cfg:        cfg,
 | 
			
		||||
		Tokens:     []string{"aaaaaa", "bbbbb"},
 | 
			
		||||
		Platform:   core.PlatFormAndroid,
 | 
			
		||||
		Message:    "Welcome",
 | 
			
		||||
		TimeToLive: &timeToLive,
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	PushToAndroid(cfg, req)
 | 
			
		||||
	PushToAndroid(req)
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
func TestAndroidNotificationStructure(t *testing.T) {
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -166,8 +166,9 @@ func GetHuaweiNotification(req PushNotification) (*model.MessageRequest, error)
 | 
			
		|||
}
 | 
			
		||||
 | 
			
		||||
// PushToHuawei provide send notification to Android server.
 | 
			
		||||
func PushToHuawei(cfg config.ConfYaml, req PushNotification) bool {
 | 
			
		||||
func PushToHuawei(req PushNotification) bool {
 | 
			
		||||
	logx.LogAccess.Debug("Start push notification for Huawei")
 | 
			
		||||
	cfg := req.Cfg
 | 
			
		||||
 | 
			
		||||
	var (
 | 
			
		||||
		client     *core.HMSClient
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
							
								
								
									
										13
									
								
								main.go
								
								
								
								
							
							
						
						
									
										13
									
								
								main.go
								
								
								
								
							| 
						 | 
				
			
			@ -19,6 +19,7 @@ import (
 | 
			
		|||
	"github.com/appleboy/gorush/gorush"
 | 
			
		||||
	"github.com/appleboy/gorush/logx"
 | 
			
		||||
	"github.com/appleboy/gorush/queue"
 | 
			
		||||
	"github.com/appleboy/gorush/queue/simple"
 | 
			
		||||
	"github.com/appleboy/gorush/router"
 | 
			
		||||
	"github.com/appleboy/gorush/rpc"
 | 
			
		||||
	"github.com/appleboy/gorush/status"
 | 
			
		||||
| 
						 | 
				
			
			@ -192,6 +193,7 @@ func main() {
 | 
			
		|||
	if opts.Android.Enabled {
 | 
			
		||||
		cfg.Android.Enabled = opts.Android.Enabled
 | 
			
		||||
		req := gorush.PushNotification{
 | 
			
		||||
			Cfg:      cfg,
 | 
			
		||||
			Platform: core.PlatFormAndroid,
 | 
			
		||||
			Message:  message,
 | 
			
		||||
			Title:    title,
 | 
			
		||||
| 
						 | 
				
			
			@ -216,7 +218,7 @@ func main() {
 | 
			
		|||
			return
 | 
			
		||||
		}
 | 
			
		||||
 | 
			
		||||
		gorush.PushToAndroid(cfg, req)
 | 
			
		||||
		gorush.PushToAndroid(req)
 | 
			
		||||
 | 
			
		||||
		return
 | 
			
		||||
	}
 | 
			
		||||
| 
						 | 
				
			
			@ -225,6 +227,7 @@ func main() {
 | 
			
		|||
	if opts.Huawei.Enabled {
 | 
			
		||||
		cfg.Huawei.Enabled = opts.Huawei.Enabled
 | 
			
		||||
		req := gorush.PushNotification{
 | 
			
		||||
			Cfg:      cfg,
 | 
			
		||||
			Platform: core.PlatFormHuawei,
 | 
			
		||||
			Message:  message,
 | 
			
		||||
			Title:    title,
 | 
			
		||||
| 
						 | 
				
			
			@ -249,7 +252,7 @@ func main() {
 | 
			
		|||
			return
 | 
			
		||||
		}
 | 
			
		||||
 | 
			
		||||
		gorush.PushToHuawei(cfg, req)
 | 
			
		||||
		gorush.PushToHuawei(req)
 | 
			
		||||
 | 
			
		||||
		return
 | 
			
		||||
	}
 | 
			
		||||
| 
						 | 
				
			
			@ -262,6 +265,7 @@ func main() {
 | 
			
		|||
 | 
			
		||||
		cfg.Ios.Enabled = opts.Ios.Enabled
 | 
			
		||||
		req := gorush.PushNotification{
 | 
			
		||||
			Cfg:      cfg,
 | 
			
		||||
			Platform: core.PlatFormIos,
 | 
			
		||||
			Message:  message,
 | 
			
		||||
			Title:    title,
 | 
			
		||||
| 
						 | 
				
			
			@ -289,7 +293,7 @@ func main() {
 | 
			
		|||
		if err := gorush.InitAPNSClient(cfg); err != nil {
 | 
			
		||||
			return
 | 
			
		||||
		}
 | 
			
		||||
		gorush.PushToIOS(cfg, req)
 | 
			
		||||
		gorush.PushToIOS(req)
 | 
			
		||||
 | 
			
		||||
		return
 | 
			
		||||
	}
 | 
			
		||||
| 
						 | 
				
			
			@ -312,7 +316,8 @@ func main() {
 | 
			
		|||
		logx.LogError.Fatal(err)
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	q := queue.NewQueue(cfg)
 | 
			
		||||
	w := simple.NewWorker(int(cfg.Core.QueueNum))
 | 
			
		||||
	q := queue.NewQueue(w)
 | 
			
		||||
	q.Start()
 | 
			
		||||
 | 
			
		||||
	finished := make(chan struct{})
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -3,9 +3,7 @@ package queue
 | 
			
		|||
import (
 | 
			
		||||
	"runtime"
 | 
			
		||||
 | 
			
		||||
	"github.com/appleboy/gorush/config"
 | 
			
		||||
	"github.com/appleboy/gorush/logx"
 | 
			
		||||
	"github.com/appleboy/gorush/queue/simple"
 | 
			
		||||
)
 | 
			
		||||
 | 
			
		||||
type (
 | 
			
		||||
| 
						 | 
				
			
			@ -20,21 +18,13 @@ type (
 | 
			
		|||
)
 | 
			
		||||
 | 
			
		||||
// NewQueue returns a Queue.
 | 
			
		||||
func NewQueue(cfg config.ConfYaml) *Queue {
 | 
			
		||||
func NewQueue(w Worker) *Queue {
 | 
			
		||||
	q := &Queue{
 | 
			
		||||
		workerCount:  int(cfg.Core.WorkerNum),
 | 
			
		||||
		queueCount:   int(cfg.Core.QueueNum),
 | 
			
		||||
		workerCount:  runtime.NumCPU(),
 | 
			
		||||
		queueCount:   runtime.NumCPU() << 1,
 | 
			
		||||
		routineGroup: newRoutineGroup(),
 | 
			
		||||
		quit:         make(chan struct{}),
 | 
			
		||||
		worker:       simple.NewWorker(cfg),
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	if q.workerCount != 0 {
 | 
			
		||||
		q.workerCount = runtime.NumCPU()
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	if q.queueCount == 0 {
 | 
			
		||||
		q.queueCount = runtime.NumCPU() << 1
 | 
			
		||||
		worker:       w,
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	return q
 | 
			
		||||
| 
						 | 
				
			
			@ -50,11 +40,6 @@ func (q *Queue) Usage() int {
 | 
			
		|||
	return q.worker.Usage()
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// Config update current config
 | 
			
		||||
func (q *Queue) Config(cfg config.ConfYaml) {
 | 
			
		||||
	q.worker.Config(cfg)
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// Start to enable all worker
 | 
			
		||||
func (q *Queue) Start() {
 | 
			
		||||
	q.startWorker()
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -3,20 +3,18 @@ package simple
 | 
			
		|||
import (
 | 
			
		||||
	"errors"
 | 
			
		||||
 | 
			
		||||
	"github.com/appleboy/gorush/config"
 | 
			
		||||
	"github.com/appleboy/gorush/gorush"
 | 
			
		||||
)
 | 
			
		||||
 | 
			
		||||
// Worker for simple queue using channel
 | 
			
		||||
type Worker struct {
 | 
			
		||||
	cfg               config.ConfYaml
 | 
			
		||||
	queueNotification chan gorush.PushNotification
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// Run start the worker
 | 
			
		||||
func (s *Worker) Run(_ chan struct{}) {
 | 
			
		||||
	for notification := range s.queueNotification {
 | 
			
		||||
		gorush.SendNotification(s.cfg, notification)
 | 
			
		||||
		gorush.SendNotification(notification)
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -45,15 +43,9 @@ func (s *Worker) Enqueue(job interface{}) error {
 | 
			
		|||
	}
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// Config update current config
 | 
			
		||||
func (s *Worker) Config(cfg config.ConfYaml) {
 | 
			
		||||
	s.cfg = cfg
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// NewWorker for struct
 | 
			
		||||
func NewWorker(cfg config.ConfYaml) *Worker {
 | 
			
		||||
func NewWorker(num int) *Worker {
 | 
			
		||||
	return &Worker{
 | 
			
		||||
		cfg:               cfg,
 | 
			
		||||
		queueNotification: make(chan gorush.PushNotification, cfg.Core.QueueNum),
 | 
			
		||||
		queueNotification: make(chan gorush.PushNotification, num),
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -1,7 +1,5 @@
 | 
			
		|||
package queue
 | 
			
		||||
 | 
			
		||||
import "github.com/appleboy/gorush/config"
 | 
			
		||||
 | 
			
		||||
// Worker interface
 | 
			
		||||
type Worker interface {
 | 
			
		||||
	Run(chan struct{})
 | 
			
		||||
| 
						 | 
				
			
			@ -9,5 +7,4 @@ type Worker interface {
 | 
			
		|||
	Enqueue(job interface{}) error
 | 
			
		||||
	Capacity() int
 | 
			
		||||
	Usage() int
 | 
			
		||||
	Config(config.ConfYaml)
 | 
			
		||||
}
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -266,6 +266,7 @@ func handleNotification(ctx context.Context, cfg config.ConfYaml, req gorush.Req
 | 
			
		|||
				continue
 | 
			
		||||
			}
 | 
			
		||||
		}
 | 
			
		||||
		notification.Cfg = cfg
 | 
			
		||||
		newNotification = append(newNotification, notification)
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -16,6 +16,7 @@ import (
 | 
			
		|||
	"github.com/appleboy/gorush/gorush"
 | 
			
		||||
	"github.com/appleboy/gorush/logx"
 | 
			
		||||
	"github.com/appleboy/gorush/queue"
 | 
			
		||||
	"github.com/appleboy/gorush/queue/simple"
 | 
			
		||||
	"github.com/appleboy/gorush/status"
 | 
			
		||||
 | 
			
		||||
	"github.com/appleboy/gofight/v2"
 | 
			
		||||
| 
						 | 
				
			
			@ -27,6 +28,7 @@ import (
 | 
			
		|||
var (
 | 
			
		||||
	goVersion = runtime.Version()
 | 
			
		||||
	q         *queue.Queue
 | 
			
		||||
	w         queue.Worker
 | 
			
		||||
)
 | 
			
		||||
 | 
			
		||||
func TestMain(m *testing.M) {
 | 
			
		||||
| 
						 | 
				
			
			@ -44,9 +46,14 @@ func TestMain(m *testing.M) {
 | 
			
		|||
		log.Fatal(err)
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	q = queue.NewQueue(cfg)
 | 
			
		||||
	w = simple.NewWorker(int(cfg.Core.QueueNum))
 | 
			
		||||
	q = queue.NewQueue(w)
 | 
			
		||||
	q.Start()
 | 
			
		||||
	defer q.Stop()
 | 
			
		||||
	defer func() {
 | 
			
		||||
		q.Stop()
 | 
			
		||||
		q.Wait()
 | 
			
		||||
	}()
 | 
			
		||||
 | 
			
		||||
	m.Run()
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -310,7 +317,7 @@ func TestMutableContent(t *testing.T) {
 | 
			
		|||
				{
 | 
			
		||||
					"tokens":          []string{"aaaaa", "bbbbb"},
 | 
			
		||||
					"platform":        core.PlatFormAndroid,
 | 
			
		||||
					"message":         "Welcome",
 | 
			
		||||
					"message":         "Welcome From API",
 | 
			
		||||
					"mutable_content": 1,
 | 
			
		||||
					"topic":           "test",
 | 
			
		||||
					"badge":           1,
 | 
			
		||||
| 
						 | 
				
			
			@ -341,12 +348,12 @@ func TestOutOfRangeMaxNotifications(t *testing.T) {
 | 
			
		|||
				{
 | 
			
		||||
					"tokens":   []string{"aaaaa", "bbbbb"},
 | 
			
		||||
					"platform": core.PlatFormAndroid,
 | 
			
		||||
					"message":  "Welcome",
 | 
			
		||||
					"message":  "Welcome API From Android",
 | 
			
		||||
				},
 | 
			
		||||
				{
 | 
			
		||||
					"tokens":   []string{"aaaaa", "bbbbb"},
 | 
			
		||||
					"platform": core.PlatFormAndroid,
 | 
			
		||||
					"message":  "Welcome",
 | 
			
		||||
					"message":  "Welcome API From Android",
 | 
			
		||||
				},
 | 
			
		||||
			},
 | 
			
		||||
		}).
 | 
			
		||||
| 
						 | 
				
			
			@ -372,7 +379,7 @@ func TestSuccessPushHandler(t *testing.T) {
 | 
			
		|||
				{
 | 
			
		||||
					"tokens":   []string{androidToken, "bbbbb"},
 | 
			
		||||
					"platform": core.PlatFormAndroid,
 | 
			
		||||
					"message":  "Welcome",
 | 
			
		||||
					"message":  "Welcome Android",
 | 
			
		||||
				},
 | 
			
		||||
			},
 | 
			
		||||
		}).
 | 
			
		||||
| 
						 | 
				
			
			@ -462,7 +469,6 @@ func TestSenMultipleNotifications(t *testing.T) {
 | 
			
		|||
 | 
			
		||||
	cfg.Android.Enabled = true
 | 
			
		||||
	cfg.Android.APIKey = os.Getenv("ANDROID_API_KEY")
 | 
			
		||||
	q.Config(cfg)
 | 
			
		||||
 | 
			
		||||
	androidToken := os.Getenv("ANDROID_TEST_TOKEN")
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -472,13 +478,13 @@ func TestSenMultipleNotifications(t *testing.T) {
 | 
			
		|||
			{
 | 
			
		||||
				Tokens:   []string{"11aa01229f15f0f0c52029d8cf8cd0aeaf2365fe4cebc4af26cd6d76b7919ef7"},
 | 
			
		||||
				Platform: core.PlatFormIos,
 | 
			
		||||
				Message:  "Welcome",
 | 
			
		||||
				Message:  "Welcome iOS",
 | 
			
		||||
			},
 | 
			
		||||
			// android
 | 
			
		||||
			{
 | 
			
		||||
				Tokens:   []string{androidToken, "bbbbb"},
 | 
			
		||||
				Platform: core.PlatFormAndroid,
 | 
			
		||||
				Message:  "Welcome",
 | 
			
		||||
				Message:  "Welcome Android",
 | 
			
		||||
			},
 | 
			
		||||
		},
 | 
			
		||||
	}
 | 
			
		||||
| 
						 | 
				
			
			@ -499,7 +505,6 @@ func TestDisabledAndroidNotifications(t *testing.T) {
 | 
			
		|||
 | 
			
		||||
	cfg.Android.Enabled = false
 | 
			
		||||
	cfg.Android.APIKey = os.Getenv("ANDROID_API_KEY")
 | 
			
		||||
	q.Config(cfg)
 | 
			
		||||
 | 
			
		||||
	androidToken := os.Getenv("ANDROID_TEST_TOKEN")
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -507,15 +512,15 @@ func TestDisabledAndroidNotifications(t *testing.T) {
 | 
			
		|||
		Notifications: []gorush.PushNotification{
 | 
			
		||||
			// ios
 | 
			
		||||
			{
 | 
			
		||||
				Tokens:   []string{"11aa01229f15f0f0c52029d8cf8cd0aeaf2365fe4cebc4af26cd6d76b7919ef7"},
 | 
			
		||||
				Tokens:   []string{"11aa01229f15f0f0c5209d8cf8cd0aeaf2365fe4cebc4af26cd6d76b7919ef7"},
 | 
			
		||||
				Platform: core.PlatFormIos,
 | 
			
		||||
				Message:  "Welcome",
 | 
			
		||||
				Message:  "Welcome iOS",
 | 
			
		||||
			},
 | 
			
		||||
			// android
 | 
			
		||||
			{
 | 
			
		||||
				Tokens:   []string{androidToken, "bbbbb"},
 | 
			
		||||
				Platform: core.PlatFormAndroid,
 | 
			
		||||
				Message:  "Welcome",
 | 
			
		||||
				Message:  "Welcome Android",
 | 
			
		||||
			},
 | 
			
		||||
		},
 | 
			
		||||
	}
 | 
			
		||||
| 
						 | 
				
			
			@ -539,7 +544,6 @@ func TestSyncModeForNotifications(t *testing.T) {
 | 
			
		|||
 | 
			
		||||
	// enable sync mode
 | 
			
		||||
	cfg.Core.Sync = true
 | 
			
		||||
	q.Config(cfg)
 | 
			
		||||
 | 
			
		||||
	androidToken := os.Getenv("ANDROID_TEST_TOKEN")
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -547,15 +551,17 @@ func TestSyncModeForNotifications(t *testing.T) {
 | 
			
		|||
		Notifications: []gorush.PushNotification{
 | 
			
		||||
			// ios
 | 
			
		||||
			{
 | 
			
		||||
				Tokens:   []string{"11aa01229f15f0f0c52029d8cf8cd0aeaf2365fe4cebc4af26cd6d76b7919ef7"},
 | 
			
		||||
				Tokens: []string{
 | 
			
		||||
					"11aa01229f15f0f0c12029d8c111d1ae1f2365f14cebc4af26cd6d76b7919ef7",
 | 
			
		||||
				},
 | 
			
		||||
				Platform: core.PlatFormIos,
 | 
			
		||||
				Message:  "Welcome",
 | 
			
		||||
				Message:  "Welcome iOS Sync",
 | 
			
		||||
			},
 | 
			
		||||
			// android
 | 
			
		||||
			{
 | 
			
		||||
				Tokens:   []string{androidToken, "bbbbb"},
 | 
			
		||||
				Platform: core.PlatFormAndroid,
 | 
			
		||||
				Message:  "Welcome",
 | 
			
		||||
				Message:  "Welcome Android Sync",
 | 
			
		||||
			},
 | 
			
		||||
		},
 | 
			
		||||
	}
 | 
			
		||||
| 
						 | 
				
			
			@ -575,7 +581,6 @@ func TestSyncModeForTopicNotification(t *testing.T) {
 | 
			
		|||
 | 
			
		||||
	// enable sync mode
 | 
			
		||||
	cfg.Core.Sync = true
 | 
			
		||||
	q.Config(cfg)
 | 
			
		||||
 | 
			
		||||
	req := gorush.RequestPush{
 | 
			
		||||
		Notifications: []gorush.PushNotification{
 | 
			
		||||
| 
						 | 
				
			
			@ -597,6 +602,7 @@ func TestSyncModeForTopicNotification(t *testing.T) {
 | 
			
		|||
			// android
 | 
			
		||||
			{
 | 
			
		||||
				// success
 | 
			
		||||
				Cfg:       cfg,
 | 
			
		||||
				Condition: "'dogs' in topics || 'cats' in topics",
 | 
			
		||||
				Platform:  core.PlatFormAndroid,
 | 
			
		||||
				Message:   "This is a Firebase Cloud Messaging Topic Message!",
 | 
			
		||||
| 
						 | 
				
			
			@ -619,7 +625,6 @@ func TestSyncModeForDeviceGroupNotification(t *testing.T) {
 | 
			
		|||
 | 
			
		||||
	// enable sync mode
 | 
			
		||||
	cfg.Core.Sync = true
 | 
			
		||||
	q.Config(cfg)
 | 
			
		||||
 | 
			
		||||
	req := gorush.RequestPush{
 | 
			
		||||
		Notifications: []gorush.PushNotification{
 | 
			
		||||
| 
						 | 
				
			
			@ -648,7 +653,6 @@ func TestDisabledIosNotifications(t *testing.T) {
 | 
			
		|||
 | 
			
		||||
	cfg.Android.Enabled = true
 | 
			
		||||
	cfg.Android.APIKey = os.Getenv("ANDROID_API_KEY")
 | 
			
		||||
	q.Config(cfg)
 | 
			
		||||
 | 
			
		||||
	androidToken := os.Getenv("ANDROID_TEST_TOKEN")
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -656,15 +660,15 @@ func TestDisabledIosNotifications(t *testing.T) {
 | 
			
		|||
		Notifications: []gorush.PushNotification{
 | 
			
		||||
			// ios
 | 
			
		||||
			{
 | 
			
		||||
				Tokens:   []string{"11aa01229f15f0f0c52029d8cf8cd0aeaf2365fe4cebc4af26cd6d76b7919ef7"},
 | 
			
		||||
				Tokens:   []string{"11aa01229f15f0f0c52021d8cf3cd0ae1f2365fe4cebc4af26cd6d76b7919ef7"},
 | 
			
		||||
				Platform: core.PlatFormIos,
 | 
			
		||||
				Message:  "Welcome",
 | 
			
		||||
				Message:  "Welcome iOS platform",
 | 
			
		||||
			},
 | 
			
		||||
			// android
 | 
			
		||||
			{
 | 
			
		||||
				Tokens:   []string{androidToken, androidToken + "_"},
 | 
			
		||||
				Platform: core.PlatFormAndroid,
 | 
			
		||||
				Message:  "Welcome",
 | 
			
		||||
				Message:  "Welcome Android platform",
 | 
			
		||||
			},
 | 
			
		||||
		},
 | 
			
		||||
	}
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -56,6 +56,7 @@ func (s *Server) Check(ctx context.Context, in *proto.HealthCheckRequest) (*prot
 | 
			
		|||
func (s *Server) Send(ctx context.Context, in *proto.NotificationRequest) (*proto.NotificationReply, error) {
 | 
			
		||||
	badge := int(in.Badge)
 | 
			
		||||
	notification := gorush.PushNotification{
 | 
			
		||||
		Cfg:              s.cfg,
 | 
			
		||||
		Platform:         int(in.Platform),
 | 
			
		||||
		Tokens:           in.Tokens,
 | 
			
		||||
		Message:          in.Message,
 | 
			
		||||
| 
						 | 
				
			
			@ -101,7 +102,7 @@ func (s *Server) Send(ctx context.Context, in *proto.NotificationRequest) (*prot
 | 
			
		|||
		}
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	go gorush.SendNotification(s.cfg, notification)
 | 
			
		||||
	go gorush.SendNotification(notification)
 | 
			
		||||
 | 
			
		||||
	return &proto.NotificationReply{
 | 
			
		||||
		Success: true,
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
		Loading…
	
		Reference in New Issue