pubsub/awssnssqs: aws sqs expose receiver max batch

pubsub/awssnssqs/awssnssqs.go

@@ -217,6 +217,9 @@ const SQSScheme = "awssqs"
 //   - nacklazy (for "awssqs" Subscriptions only): sets SubscriberOptions.NackLazy. The
 //     value must be parseable by `strconv.ParseBool`.
 //   - waittime: sets SubscriberOptions.WaitTime, in time.ParseDuration formats.
+//   - receivermaxbatch: sets the receiver max batch size as an integer.
+//     SQS supports receiving at most 10 messages at a time:
+//     https://godoc.org/github.com/aws/aws-sdk-go/service/sqs#SQS.ReceiveMessage
 //
 // See gocloud.dev/aws/ConfigFromURLParams for other query parameters
 // that affect the default AWS session.
@@ -302,6 +305,17 @@ func (o *URLOpener) OpenSubscriptionURL(ctx context.Context, u *url.URL) (*pubsu
 		}
 		q.Del("waittime")
 	}
+	// SQS supports receiving at most 10 messages at a time:
+	// https://godoc.org/github.com/aws/aws-sdk-go/service/sqs#SQS.ReceiveMessage
+	// if value is higher than 10, it will default to 10
+	if receiverMaxBatchStr := q.Get("receivermaxbatch"); receiverMaxBatchStr != "" {
+		var err error
+		opts.ReceiveBatcherOptions.MaxBatchSize, err = strconv.Atoi(receiverMaxBatchStr)
+		if err != nil {
+			return nil, fmt.Errorf("invalid value %q for receivermaxbatch: %v", receiverMaxBatchStr, err)
+		}
+		q.Del("receivermaxbatch")
+	}
 	qURL := "https://" + path.Join(u.Host, u.Path)
 	if o.UseV2 {
 		cfg, err := gcaws.V2ConfigFromURLParams(ctx, q)

pubsub/awssnssqs/awssnssqs_test.go

@@ -729,6 +729,10 @@ func TestOpenSubscriptionFromURL(t *testing.T) {
 		{"awssqs://sqs.us-east-2.amazonaws.com/99999/my-queue?waittime=foo", true},
 		// Invalid parameter.
 		{"awssqs://sqs.us-east-2.amazonaws.com/99999/my-queue?param=value", true},
+		// OK, setting receivermaxbatch.
+		{"awssqs://sqs.us-east-2.amazonaws.com/99999/my-queue?receivermaxbatch=5", false},
+		// Invalid receivermaxbatch.
+		{"awssqs://sqs.us-east-2.amazonaws.com/99999/my-queue?receivermaxbatch=foo", true},
 	}
 
 	ctx := context.Background()