Squashed 'src/github.com/uniqush/uniqush-push/' changes from fe9fdc1..aa8d278

aa8d278 min. wait time is 1 sec
0d7b936 randomly wait to prevent message storm
21f4735 version
0ad2647 To work with CL 67010043
a4c9fc4 version
ff309e0 wrong waiting time in apns.
0320967 Try to fix #45
525f18a fix #44
3944e04 try to recover on temporary error.

git-subtree-dir: src/github.com/uniqush/uniqush-push
git-subtree-split: aa8d278e0fad38bbb00cb84111f5586e2e3e7d6a

Author: mindreframer

db/pushredisdb.go

@@ -20,10 +20,11 @@ package db
 import (
 	"errors"
 	"fmt"
-	redis "github.com/monnand/goredis"
-	. "github.com/uniqush/uniqush-push/push"
 	"strconv"
 	"strings"
+
+	redis "github.com/monnand/goredis"
+	. "github.com/uniqush/uniqush-push/push"
 )
 
 type PushRedisDB struct {
@@ -75,22 +76,22 @@ func newPushRedisDB(c *DatabaseConfig) (*PushRedisDB, error) {
 	return ret, nil
 }
 
-func (r *PushRedisDB) keyValueToDeliveryPoint(name string, value []byte) *DeliveryPoint {
+func (r *PushRedisDB) keyValueToDeliveryPoint(name string, value []byte) (dp *DeliveryPoint, err error) {
 	psm := r.psm
-	dp, err := psm.BuildDeliveryPointFromBytes(value)
+	dp, err = psm.BuildDeliveryPointFromBytes(value)
 	if err != nil {
-		return nil
+		dp = nil
 	}
-	return dp
+	return
 }
 
-func (r *PushRedisDB) keyValueToPushServiceProvider(name string, value []byte) *PushServiceProvider {
+func (r *PushRedisDB) keyValueToPushServiceProvider(name string, value []byte) (psp *PushServiceProvider, err error) {
 	psm := r.psm
-	psp, err := psm.BuildPushServiceProviderFromBytes(value)
+	psp, err = psm.BuildPushServiceProviderFromBytes(value)
 	if err != nil {
-		return nil
+		psp = nil
 	}
-	return psp
+	return
 }
 
 func deliveryPointToValue(dp *DeliveryPoint) []byte {
@@ -109,7 +110,7 @@ func (r *PushRedisDB) GetDeliveryPoint(name string) (*DeliveryPoint, error) {
 	if b == nil {
 		return nil, nil
 	}
-	return r.keyValueToDeliveryPoint(name, b), nil
+	return r.keyValueToDeliveryPoint(name, b)
 }
 
 func (r *PushRedisDB) SetDeliveryPoint(dp *DeliveryPoint) error {
@@ -125,7 +126,7 @@ func (r *PushRedisDB) GetPushServiceProvider(name string) (*PushServiceProvider,
 	if b == nil {
 		return nil, nil
 	}
-	return r.keyValueToPushServiceProvider(name, b), nil
+	return r.keyValueToPushServiceProvider(name, b)
 }
 
 func (r *PushRedisDB) SetPushServiceProvider(psp *PushServiceProvider) error {

main.go

@@ -20,15 +20,15 @@ package main
 import (
 	"flag"
 	"fmt"
-	. "github.com/uniqush/uniqush-push/srv"
 	"os"
 	"runtime"
+	. "github.com/uniqush/uniqush-push/srv"
 )
 
 var uniqushPushConfFlags = flag.String("config", "/etc/uniqush/uniqush-push.conf", "Config file path")
 var uniqushPushShowVersionFlag = flag.Bool("version", false, "Version info")
 
-var uniqushPushVersion = "uniqush-push 1.5.1"
+var uniqushPushVersion = "uniqush-push 1.5.2"
 
 func installPushSrvices() {
 	InstallGCM()

push/errors.go

@@ -47,12 +47,26 @@ type RetryError struct {
 	Provider    *PushServiceProvider
 	Destination *DeliveryPoint
 	Content     *Notification
+	Reason      error
 }
 
 func (e *RetryError) Error() string {
+	if e.Reason != nil {
+		return fmt.Sprintf("Retry (%v)", e.Reason)
+	}
 	return fmt.Sprintf("Retry")
 }
 
+func NewRetryErrorWithReason(psp *PushServiceProvider, dp *DeliveryPoint, notif *Notification, after time.Duration, reason error) error {
+	return &RetryError{
+		After:       after,
+		Provider:    psp,
+		Destination: dp,
+		Content:     notif,
+		Reason:      reason,
+	}
+}
+
 func NewRetryError(psp *PushServiceProvider, dp *DeliveryPoint, notif *Notification, after time.Duration) error {
 	return &RetryError{
 		After:       after,

pushbackend.go

@@ -18,11 +18,11 @@
 package main
 
 import (
+	"sync"
+	"time"
 	. "github.com/uniqush/log"
 	. "github.com/uniqush/uniqush-push/db"
 	. "github.com/uniqush/uniqush-push/push"
-	"sync"
-	"time"
 )
 
 type PushBackEnd struct {
@@ -111,7 +111,7 @@ func (self *PushBackEnd) fixError(reqId string, event error, logger Logger, afte
 		if sub, ok = err.Destination.FixedData["subscriber"]; !ok {
 			return nil
 		}
-		if after <= 0*time.Second {
+		if after <= 1*time.Second {
 			after = 5 * time.Second
 		}
 		if after > 1*time.Minute {

restapi.go

@@ -20,18 +20,17 @@ package main
 import (
 	"crypto/rand"
 	"encoding/base64"
-	"time"
-
 	"fmt"
-
-	"github.com/uniqush/log"
-	. "github.com/uniqush/uniqush-push/push"
 	"io"
 	"net/http"
 	"regexp"
 	"strconv"
 	"strings"
 	"sync"
+	"time"
+
+	"github.com/uniqush/log"
+	. "github.com/uniqush/uniqush-push/push"
 )
 
 type RestAPI struct {
@@ -115,7 +114,13 @@ func getSubscribersFromMap(kv map[string]string, validate bool) (subs []string,
 			return
 		}
 	}
-	subs = strings.Split(v, ",")
+	s := strings.Split(v, ",")
+	subs = make([]string, 0, len(s))
+	for _, sub := range s {
+		if len(sub) > 0 {
+			subs = append(subs, sub)
+		}
+	}
 	if validate {
 		err = validateSubscribers(subs)
 		if err != nil {
@@ -247,7 +252,7 @@ func (self *RestAPI) pushNotification(reqId string, kv map[string]string, perdp
 		return
 	}
 
-	logger.Infof("RequestId=%v From=%v Service=%v Subscribers=\"%v\"", reqId, remoteAddr, service, subs)
+	logger.Infof("RequestId=%v From=%v Service=%v NrSubscribers=%v Subscribers=\"%+v\"", reqId, remoteAddr, service, len(subs), subs)
 
 	self.backend.Push(reqId, service, subs, notif, perdp, logger)
 	return

srv/apns.go

@@ -25,9 +25,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/uniqush/cache"
-	"github.com/uniqush/connpool"
-	. "github.com/uniqush/uniqush-push/push"
 	"io"
 	"math/rand"
 	"net"
@@ -36,6 +33,10 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/uniqush/cache"
+	"github.com/uniqush/connpool"
+	. "github.com/uniqush/uniqush-push/push"
 )
 
 const (
@@ -161,6 +162,10 @@ func (p *apnsPushService) BuildDeliveryPointFromMap(kv map[string]string, dp *De
 		return errors.New("NoSubscriber")
 	}
 	if devtoken, ok := kv["devtoken"]; ok && len(devtoken) > 0 {
+		_, err := hex.DecodeString(devtoken)
+		if err != nil {
+			return fmt.Errorf("Invalid delivery point: bad device token. %v", err)
+		}
 		dp.FixedData["devtoken"] = devtoken
 	} else {
 		return errors.New("NoDevToken")
@@ -405,12 +410,12 @@ func newAPNSConnManager(psp *PushServiceProvider, resultChan chan *apnsResult) *
 		InsecureSkipVerify: false,
 	}
 
+	manager.addr = psp.VolatileData["addr"]
 	if skip, ok := psp.VolatileData["skipverify"]; ok {
 		if skip == "true" {
 			manager.conf.InsecureSkipVerify = true
 		}
 	}
-	manager.addr = psp.VolatileData["addr"]
 	manager.resultChan = resultChan
 	return manager
 }
@@ -420,16 +425,23 @@ func (self *apnsConnManager) NewConn() (net.Conn, error) {
 		return nil, self.err
 	}
 
-	conn, err := net.DialTimeout("tcp", self.addr, time.Duration(maxWaitTime)*time.Second)
-	if err != nil {
-		return nil, err
-	}
+	/*
+		conn, err := net.DialTimeout("tcp", self.addr, time.Duration(maxWaitTime)*time.Second)
+		if err != nil {
+			return nil, err
+		}
 
-	if c, ok := conn.(*net.TCPConn); ok {
-		c.SetKeepAlive(true)
-	}
-	tlsconn := tls.Client(conn, self.conf)
-	err = tlsconn.Handshake()
+		if c, ok := conn.(*net.TCPConn); ok {
+			c.SetKeepAlive(true)
+		}
+		tlsconn := tls.Client(conn, self.conf)
+		err = tlsconn.Handshake()
+		if err != nil {
+			return nil, err
+		}
+	*/
+
+	tlsconn, err := tls.Dial("tcp", self.addr, self.conf)
 	if err != nil {
 		return nil, err
 	}
@@ -491,7 +503,8 @@ func (self *apnsPushService) singlePush(payload, token []byte, expiry uint32, mi
 		conn.Close()
 
 		time.Sleep(sleepTime)
-		sleepTime *= sleepTime
+		// randomly wait more time
+		sleepTime += time.Duration(rand.Int63n(int64(sleepTime)))
 		// Let's try another connection to see if we can recover this error
 		conn, err = pool.Get()
 

srv/gcm.go

@@ -23,11 +23,12 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	. "github.com/uniqush/uniqush-push/push"
 	"io/ioutil"
+	"net"
 	"net/http"
 	"strconv"
 	"time"
+	. "github.com/uniqush/uniqush-push/push"
 )
 
 const (
@@ -211,8 +212,21 @@ func (self *gcmPushService) multicast(psp *PushServiceProvider, dpList []*Delive
 			res.Provider = psp
 			res.Content = notif
 
-			res.Err = e2
 			res.Destination = dp
+			if err, ok := e2.(net.Error); ok {
+				// Temporary error. Try to recover
+				if err.Temporary() {
+					after := 3 * time.Second
+					res.Err = NewRetryErrorWithReason(psp, dp, notif, after, err)
+				}
+			} else if err, ok := e2.(*net.DNSError); ok {
+				// DNS error, try to recover it by retry
+				after := 3 * time.Second
+				res.Err = NewRetryErrorWithReason(psp, dp, notif, after, err)
+
+			} else {
+				res.Err = e2
+			}
 			resQueue <- res
 		}
 		return