Implement support for Publish Many for GCPS Publisher

Author: AJCStriker

src/queue/drivers/gcps/GCPSClient.ts

@@ -61,18 +61,19 @@ export class GCPSClient {
         })
     }
 
-    public async publish(topic: string, message: GCPSMessage): Promise<void> {
+    public async publish(
+        topic: string,
+        messages: Array<GCPSMessage>
+    ): Promise<void> {
         await Axios.post(
             `https://pubsub.googleapis.com/v1/${topic}:publish`,
             {
-                messages: [
-                    {
-                        attributes: message.attributes,
-                        data: Buffer.from(
-                            JSON.stringify(message.data)
-                        ).toString("base64"),
-                    },
-                ],
+                messages: messages.map((m) => ({
+                    attributes: m.attributes,
+                    data: Buffer.from(JSON.stringify(m.data)).toString(
+                        "base64"
+                    ),
+                })),
             },
             {
                 headers: {

src/queue/drivers/gcps/GCPSConsumer.ts

@@ -3,6 +3,7 @@ import { QueueHandler } from "../../abstract/QueueHandler"
 import { SerializedTask } from "../../abstract/SerializedTask"
 import { TransientError } from "../../../errors/TransientError"
 import { Container } from "inversify"
+import { isEmpty } from "lodash"
 import { Logger } from "../../../logging"
 import { Process } from "../../../runtime"
 import { ConstructorOf } from "../../../utils/types"
@@ -45,7 +46,7 @@ export class GCPSConsumer implements Process {
             this.subscriptionName
         )
 
-        if (subscription.pushConfig !== {}) {
+        if (!isEmpty(subscription.pushConfig)) {
             throw new Error(
                 "The Strontium GCPS Consumer does not support Push based GCPS subscriptions. " +
                     "Please change the subscription inside Google Cloud Platform to operate on a Pull Based model if you wish " +
@@ -92,16 +93,17 @@ export class GCPSConsumer implements Process {
             )
 
             await Promise.all(
-                messages.map(async (m) =>
-                    this.executeTask(
-                        m.ackId,
-                        {
-                            eventName:
-                                m.message.attributes.STRONTIUM_EVENT_NAME,
-                            message: m.message.data,
-                        },
-                        container
-                    )
+                messages.map(
+                    async (m) =>
+                        await this.executeTask(
+                            m.ackId,
+                            {
+                                eventName:
+                                    m.message.attributes.STRONTIUM_EVENT_NAME,
+                                message: m.message.data,
+                            },
+                            container
+                        )
                 )
             )
         }

src/queue/drivers/gcps/GCPSPublisher.ts

@@ -24,14 +24,21 @@ export class GCPSPublisher extends QueuePublisher implements Process {
     public async publish<Q extends {}>(
         queueName: string,
         eventName: string,
-        message: Q
+        messages: Q | Array<Q>
     ): Promise<void> {
-        return this.client.publish(queueName, {
-            attributes: {
-                STRONTIUM_EVENT_NAME: eventName,
-            },
-            data: message,
-        })
+        if (!Array.isArray(messages)) {
+            messages = [messages]
+        }
+
+        return this.client.publish(
+            queueName,
+            messages.map((m) => ({
+                attributes: {
+                    STRONTIUM_EVENT_NAME: eventName,
+                },
+                data: m,
+            }))
+        )
     }
 
     public async shutdown(container: Container): Promise<void> {

tests/queue/drivers/GCPSClient.spec.ts

@@ -10,7 +10,7 @@ describe("GCPSClient", () => {
 
     before(function() {
         // If the test suite is not running in CI then skip this suite - it's slow and requires credentials
-        if (process.env.CI !== "true") {
+        if (process.env.CI !== "true" || isNaN(Number(process.env.TRAVIS_PULL_REQUEST)) ) {
             this.skip()
         }
     })
@@ -36,10 +36,12 @@ describe("GCPSClient", () => {
     it("Should publish a message and correctly reconstruct it", async () => {
         await client.publish(
             "projects/strontium-tests/topics/integrationTestTopic",
-            {
-                data: "MY-INTEGRATION-TEST",
-                attributes: {},
-            }
+            [
+                {
+                    data: "MY-INTEGRATION-TEST",
+                    attributes: {},
+                },
+            ]
         )
 
         let messages = await client.pullTasks(
@@ -59,10 +61,12 @@ describe("GCPSClient", () => {
     it("Acking a message should remove it from the queue", async () => {
         await client.publish(
             "projects/strontium-tests/topics/integrationTestTopic",
-            {
-                data: "MY-INTEGRATION-TEST",
-                attributes: {},
-            }
+            [
+                {
+                    data: "MY-INTEGRATION-TEST",
+                    attributes: {},
+                },
+            ]
         )
 
         let messages = await client.pullTasks(
@@ -92,10 +96,12 @@ describe("GCPSClient", () => {
     it("Nacking a message should readd it to the queue", async () => {
         await client.publish(
             "projects/strontium-tests/topics/integrationTestTopic",
-            {
-                data: "NACKED-MESSAGE",
-                attributes: {},
-            }
+            [
+                {
+                    data: "NACKED-MESSAGE",
+                    attributes: {},
+                },
+            ]
         )
 
         let messages = await client.pullTasks(