[APNS] Make pushQueue and responseChannel buffered (#57)

* feat(apns): make pushQ and responseChan buffered

The pushQueue and responseChannels are used by multiple goroutines
and are not buffered, thus deadlocks might happen and this would
throttle whoever is feeding or consuming from it. Thus, make them
buffered and configurable

* chore(apns): log retries as info

They are important for us to check that this event is happening
and check if there's any throttling/deadlock happening on channel
consumption

config/default.yaml

@@ -3,6 +3,8 @@ gracefulShutdownTimeout: 30
 apns:
   concurrentWorkers: 300
   connectionPoolSize: 1
+  pushQueueSize: 100
+  responseChannelSize: 100
   logStatsInterval: 10000
   apps: "game"
   certs:

extensions/apns_message_handler.go

@@ -323,7 +323,7 @@ func (a *APNSMessageHandler) handleAPNSResponse(responseWithMetadata *structs.Re
 				"sendAttempts": sendAttempts,
 				"maxRetries":   a.maxRetryAttempts,
 				"apnsID":       responseWithMetadata.ApnsID,
-			}).Debug("retrying notification")
+			}).Info("retrying notification")
 			inFlightNotificationInstance.sendAttempts.Add(1)
 			if a.pendingMessagesWG != nil {
 				a.pendingMessagesWG.Add(1)

extensions/apns_push_queue.go

@@ -64,6 +64,12 @@ func NewAPNSPushQueue(
 	}
 }
 
+func (p *APNSPushQueue) loadConfigDefault() {
+	p.Config.SetDefault("apns.connectionPoolSize", 1)
+	p.Config.SetDefault("apns.pushQueueSize", 100)
+	p.Config.SetDefault("apns.responseChannelSize", 100)
+}
+
 // Configure configures queues and token
 func (p *APNSPushQueue) Configure() error {
 	l := p.Logger.WithFields(log.Fields{
@@ -75,7 +81,10 @@ func (p *APNSPushQueue) Configure() error {
 		return err
 	}
 	p.Closed = false
+	p.loadConfigDefault()
 	connectionPoolSize := p.Config.GetInt("apns.connectionPoolSize")
+	pushQueueSize := p.Config.GetInt("apns.pushQueueSize")
+	respChannelSize := p.Config.GetInt("apns.responseChannelSize")
 	p.clients = make(chan *apns2.Client, connectionPoolSize)
 	for i := 0; i < connectionPoolSize; i++ {
 		client := apns2.NewTokenClient(p.token)
@@ -89,8 +98,8 @@ func (p *APNSPushQueue) Configure() error {
 		p.clients <- client
 	}
 	l.Debug("clients configured")
-	p.pushChannel = make(chan *apns2.Notification)
-	p.responseChannel = make(chan *structs.ResponseWithMetadata)
+	p.pushChannel = make(chan *apns2.Notification, pushQueueSize)
+	p.responseChannel = make(chan *structs.ResponseWithMetadata, respChannelSize)
 
 	for i := 0; i < p.Config.GetInt("apns.concurrentWorkers"); i++ {
 		go p.pushWorker()

extensions/apns_push_queue_test.go

@@ -75,6 +75,13 @@ var _ = Describe("APNS Push Queue", func() {
 				Expect(err).To(HaveOccurred())
 				Expect(err.Error()).To(Equal("open ./invalid-certficate.pem: no such file or directory"))
 			})
+
+			It("should load default configs", func() {
+				err := queue.Configure()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(cap(queue.pushChannel)).To(Equal(100))
+				Expect(cap(queue.responseChannel)).To(Equal(100))
+			})
 		})
 
 		Describe("Configuring Certificate", func() {