Send multiple notifications concurrently (#54)

* example of sending notifications concurrently

* push: build a worker pool into Service

* push: remove version of Push that does serialization

and update docs

* push: reorganize

* push: separate queue type for async

and improve handling of command line flags

* test queue push

(not sure how useful this test is)

README.md

@@ -56,43 +56,83 @@ I am still looking for feedback on the API so it may change. Please copy Buford
 package main
 
 import (
-	"log"
+	"encoding/json"
+	"fmt"
 
 	"github.com/RobotsAndPencils/buford/certificate"
 	"github.com/RobotsAndPencils/buford/payload"
 	"github.com/RobotsAndPencils/buford/payload/badge"
 	"github.com/RobotsAndPencils/buford/push"
 )
 
-func main() {
-	// set these variables appropriately
-	filename := "/path/to/certificate.p12"
-	password := ""
-	deviceToken := "c2732227a1d8021cfaf781d71fb2f908c61f5861079a00954a5453f1d0281433"
+// set these variables appropriately
+const (
+	filename = "/path/to/certificate.p12"
+	password = ""
+	host = push.Development
+	deviceToken = "c2732227a1d8021cfaf781d71fb2f908c61f5861079a00954a5453f1d0281433"
+)
 
+func main() {
+	// load a certificate and use it to connect to the APN service:
 	cert, err := certificate.Load(filename, password)
-	if err != nil {
-		log.Fatal(err)
-	}
+	exitOnError(err)
 
-	service, err := push.NewService(push.Development, cert)
-	if err != nil {
-		log.Fatal(err)
-	}
+	client, err := push.NewClient(cert)
+	exitOnError(err)
+
+	service := push.NewService(client, host)
 
+	// construct a payload to send to the device:
 	p := payload.APS{
 		Alert: payload.Alert{Body: "Hello HTTP/2"},
 		Badge: badge.New(42),
 	}
+	b, err := json.Marshal(p)
+	exitOnError(err)
+
+	// push the notification:
+	id, err := service.Push(deviceToken, nil, b)
+	exitOnError(err)
+
+	fmt.Println("apns-id:", id)
+}
+```
+
+See `example/push` for the complete listing.
+
+#### Concurrent use
+
+HTTP/2 can send multiple requests over a single connection, but `service.Push` waits for a response before returning. Instead, you can wrap a `Service` in a queue to handle responses independently, allowing you to send multiple notifications at once.
+
+```go
+queue := push.NewQueue(service, workers)
 
-	id, err := service.Push(deviceToken, nil, p)
-	if err != nil {
-		log.Fatal(err)
+// process responses
+go func() {
+	for {
+		// Response blocks until a response is available
+		log.Println(queue.Response())
 	}
-	log.Println("apns-id:", id)
+}()
+
+// send the notifications
+for i := 0; i < number; i++ {
+	queue.Push(deviceToken, nil, b)
 }
+
+// done sending notifications, wait for all responses
+queue.Wait()
 ```
 
+It's important to set up a goroutine to handle responses before sending any notifications, otherwise Push will block waiting for room to return a Response.
+
+You can configure the number of workers used to send notifications concurrently, but be aware that a larger number isn't necessarily better, as Apple limits the number of concurrent streams. From the Apple Push Notification documentation:
+
+> "The APNs server allows multiple concurrent streams for each connection. The exact number of streams is based on server load, so do not assume a specific number of streams."
+
+See `example/concurrent/` for a complete listing.
+
 #### Headers
 
 You can specify an ID, expiration, priority, and other parameters via the Headers struct.
@@ -104,7 +144,7 @@ headers := &push.Headers{
 	LowPriority: true,
 }
 
-id, err := service.Push(deviceToken, headers, p)
+id, err := service.Push(deviceToken, headers, b)
 ```
 
 If no ID is specified APNS will generate and return a unique ID. When an expiration is specified, APNS will store and retry sending the notification until that time, otherwise APNS will not store or retry the notification. LowPriority should always be set when sending a ContentAvailable payload.
@@ -120,29 +160,17 @@ p := payload.APS{
 pm := p.Map()
 pm["acme2"] = []string{"bang", "whiz"}
 
-id, err := service.Push(deviceToken, nil, pm)
-```
-
-The Push method will use json.Marshal to serialize whatever you send it.
-
-#### Resend the same payload
-
-Use json.Marshal to serialize your payload once and then send it to multiple device tokens with PushBytes.
-
-```go
-b, err := json.Marshal(p)
+b, err := json.Marshal(pm)
 if err != nil {
-	log.Fatal(err)
+	log.Fatal(b)
 }
 
-id, err := service.PushBytes(deviceToken, nil, b)
+service.Push(deviceToken, nil, b)
 ```
 
-Whether you use Push or PushBytes, the underlying HTTP/2 connection to APNS will be reused.
-
 #### Error responses
 
-Push and PushBytes may return an `error`. It could be an error the JSON encoding or HTTP request, or it could be a `push.Error` which contains the response from Apple. To access the Reason and HTTP Status code, you must convert the `error` to a `push.Error` as follows:
+If `service.Push` or `queue.Response` returns an error, it could be an HTTP error, or it could be an error response from Apple. To access the Reason and HTTP Status code, you must convert the `error` to a `push.Error` as follows:
 
 ```go
 if e, ok := err.(*push.Error); ok {

example/concurrent/main.go

@@ -0,0 +1,102 @@
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"time"
+
+	"github.com/RobotsAndPencils/buford/certificate"
+	"github.com/RobotsAndPencils/buford/payload"
+	"github.com/RobotsAndPencils/buford/push"
+)
+
+func main() {
+	log.SetFlags(log.Ltime | log.Lmicroseconds)
+
+	var deviceToken, filename, password, environment, host string
+	var workers uint
+	var number int
+
+	flag.StringVar(&deviceToken, "d", "", "Device token")
+	flag.StringVar(&filename, "c", "", "Path to p12 certificate file")
+	flag.StringVar(&password, "p", "", "Password for p12 file")
+	flag.StringVar(&environment, "e", "development", "Environment")
+	flag.UintVar(&workers, "w", 20, "Workers to send notifications")
+	flag.IntVar(&number, "n", 100, "Number of notifications to send")
+	flag.Parse()
+
+	// ensure required flags are set:
+	halt := false
+	if deviceToken == "" {
+		fmt.Println("Device token is required.")
+		halt = true
+	}
+	if filename == "" {
+		fmt.Println("Path to .p12 certificate file is required.")
+		halt = true
+	}
+	switch environment {
+	case "development":
+		host = push.Development
+	case "production":
+		host = push.Production
+	default:
+		fmt.Println("Environment can be development or production.")
+		halt = true
+	}
+	if halt {
+		flag.Usage()
+		os.Exit(2)
+	}
+
+	// load a certificate and use it to connect to the APN service:
+	cert, err := certificate.Load(filename, password)
+	exitOnError(err)
+
+	client, err := push.NewClient(cert)
+	exitOnError(err)
+	service := push.NewService(client, host)
+	queue := push.NewQueue(service, workers)
+
+	// process responses
+	go func() {
+		count := 1
+		for {
+			id, device, err := queue.Response()
+			if err != nil {
+				log.Printf("(%d) device: %s, error: %v", count, device, err)
+			} else {
+				log.Printf("(%d) device: %s, apns-id: %s", count, device, id)
+			}
+			count++
+		}
+	}()
+
+	// prepare notification(s) to send
+	p := payload.APS{
+		Alert: payload.Alert{Body: "Hello HTTP/2"},
+	}
+	b, err := json.Marshal(p)
+	exitOnError(err)
+
+	// send notifications:
+	start := time.Now()
+	for i := 0; i < number; i++ {
+		queue.Push(deviceToken, nil, b)
+	}
+	// done sending notifications, wait for all responses:
+	queue.Wait()
+	elapsed := time.Since(start)
+
+	log.Printf("Time for %d responses: %s (%s ea.)", number, elapsed, elapsed/time.Duration(number))
+}
+
+func exitOnError(err error) {
+	if err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+}

example/push/main.go

@@ -1,8 +1,10 @@
 package main
 
 import (
+	"encoding/json"
 	"flag"
-	"log"
+	"fmt"
+	"os"
 
 	"github.com/RobotsAndPencils/buford/certificate"
 	"github.com/RobotsAndPencils/buford/payload"
@@ -11,35 +13,65 @@ import (
 )
 
 func main() {
-	var deviceToken, filename, password, environment string
+	var deviceToken, filename, password, environment, host string
 
 	flag.StringVar(&deviceToken, "d", "", "Device token")
-	flag.StringVar(&filename, "c", "", "Path to p12 certificate file")
-	flag.StringVar(&password, "p", "", "Password for p12 file.")
+	flag.StringVar(&filename, "c", "", "Path to .p12 certificate file.")
+	flag.StringVar(&password, "p", "", "Password for .p12 file.")
 	flag.StringVar(&environment, "e", "development", "Environment")
 	flag.Parse()
 
-	cert, err := certificate.Load(filename, password)
-	if err != nil {
-		log.Fatal(err)
+	// ensure required flags are set:
+	halt := false
+	if deviceToken == "" {
+		fmt.Println("Device token is required.")
+		halt = true
 	}
-
-	service, err := push.NewService(push.Development, cert)
-	if err != nil {
-		log.Fatal(err)
+	if filename == "" {
+		fmt.Println("Path to .p12 certificate file is required.")
+		halt = true
+	}
+	switch environment {
+	case "development":
+		host = push.Development
+	case "production":
+		host = push.Production
+	default:
+		fmt.Println("Environment can be development or production.")
+		halt = true
 	}
-	if environment == "production" {
-		service.Host = push.Production
+	if halt {
+		flag.Usage()
+		os.Exit(2)
 	}
 
+	// load a certificate and use it to connect to the APN service:
+	cert, err := certificate.Load(filename, password)
+	exitOnError(err)
+
+	client, err := push.NewClient(cert)
+	exitOnError(err)
+
+	service := push.NewService(client, host)
+
+	// construct a payload to send to the device:
 	p := payload.APS{
 		Alert: payload.Alert{Body: "Hello HTTP/2"},
 		Badge: badge.New(42),
 	}
+	b, err := json.Marshal(p)
+	exitOnError(err)
+
+	// push the notification:
+	id, err := service.Push(deviceToken, nil, b)
+	exitOnError(err)
+
+	fmt.Println("apns-id:", id)
+}
 
-	id, err := service.Push(deviceToken, &push.Headers{}, p)
+func exitOnError(err error) {
 	if err != nil {
-		log.Fatal(err)
+		fmt.Println(err)
+		os.Exit(1)
 	}
-	log.Println("apns-id:", id)
 }

example/website/main.go

@@ -55,8 +55,12 @@ func pushHandler(w http.ResponseWriter, r *http.Request) {
 		// URLArgs must match placeholders in URLFormatString
 		URLArgs: []string{"hello"},
 	}
+	b, err := json.Marshal(p)
+	if err != nil {
+		log.Fatal(err)
+	}
 
-	id, err := service.Push(deviceToken, nil, p)
+	id, err := service.Push(deviceToken, nil, b)
 	if err != nil {
 		log.Println(err)
 		return
@@ -140,11 +144,13 @@ func main() {
 		log.Fatal(err)
 	}
 
-	service, err = push.NewService(push.Production, cert)
+	client, err := push.NewClient(cert)
 	if err != nil {
 		log.Fatal(err)
 	}
 
+	service = push.NewService(client, push.Production)
+
 	r := mux.NewRouter()
 	r.HandleFunc("/", indexHandler).Methods("GET")
 	r.HandleFunc("/request", requestPermissionHandler)