event.json

{
  "attributes": {"hello": "world"},
  "data": {"my": "event"}
}

pubsub.js

const { randomInt } = require("crypto")
const { readFileSync, writeFileSync } = require("fs")
const { URL } = require("url")

const USAGE = `
node pubsub print event.json

  Prints the JSON that would be posted to the server in a push based pubsub subscription

node pubsub write event.json body.json

  Write the JSON that would be posted to the server into body.json, you can then post it using CURL

node pubsub post event.json http://example.com

  Post an event to the provided URL in the same way as pubsub would
`
const [COMMAND, EVENT_FILE, OUTPUT] = process.argv.slice(2)

if (process.argv.length < 4) {
  fail(USAGE)
}

switch (COMMAND) {
  case `print`:
    if (process.argv.length !== 4) {
      fail(USAGE)
    }
    break
  case `write`:
  case `post`:
    if (process.argv.length !== 5) {
      fail(USAGE)
    }
    break
  default:
    fail(USAGE)
}

const eventObj = JSON.parse(readFileSync(EVENT_FILE, `utf8`))

if (typeof eventObj !== "object" || eventObj === null) {
  fail(`Expected ${EVENT_FILE} to contain a JSON encoded object`)
}
for (const key of Object.keys(eventObj)) {
  if (key !== `attributes` && key !== `data` && key !== `subscription`) {
    fail(
      `Expected ${EVENT_FILE} to be an object with keys: "attributes", "data" and optionally "subscription" but found "${key}"`
    )
  }
}
if (!(`attributes` in eventObj)) {
  fail(`The key "attributes" was not found in ${EVENT_FILE}`)
}
if (!(`data` in eventObj)) {
  fail(`The key "data" was not found in ${EVENT_FILE}`)
}
const {
  attributes,
  data,
  subscription = "projects/myproject/subscriptions/mysubscription",
} = eventObj

if (typeof subscription !== "string") {
  fail(`Expected subscription to be a string`)
}

if (typeof attributes !== "object" || attributes === null) {
  fail(`Expected attributes to be an object with strings as values`)
}
for (const [k, v] of Object.entries(attributes)) {
  if (typeof k !== "string") {
    fail(`Expected attribute key to be a string`)
  }
  if (typeof v !== "string") {
    fail(`Expected attribute value for "${k}" to be a string`)
  }
}

// example from https://cloud.google.com/pubsub/docs/push

// {
//   "message": {
//       "attributes": {
//           "key": "value"
//       },
//       "data": "SGVsbG8gQ2xvdWQgUHViL1N1YiEgSGVyZSBpcyBteSBtZXNzYWdlIQ==",
//       "messageId": "2070443601311540",
//       "message_id": "2070443601311540",
//       "publishTime": "2021-02-26T19:13:55.749Z",
//       "publish_time": "2021-02-26T19:13:55.749Z",
//   },
//  "subscription": "projects/myproject/subscriptions/mysubscription"
// }
const messageId = randomInt(10000000000000, 99999999999999).toString(10)
const publishTime = new Date().toISOString()

const body = {
  message: {
    attributes,
    data: Buffer.from(JSON.stringify(data), "utf8").toString("base64"),
    messageId,
    message_id: messageId,
    publishTime,
    publish_time: publishTime,
  },
  subscription,
}

if (COMMAND === `print`) {
  console.info(JSON.stringify(body, null, `  `))
} else if (COMMAND === `write`) {
  writeFileSync(OUTPUT, JSON.stringify(body, null, `  `))
} else if (COMMAND === `post`) {
  const url = new URL(OUTPUT)
  if (url.protocol !== "http:" && url.protocol !== "https:") {
    fail(`URL must be http or https: ${OUTPUT}`)
  }
  const { request } = require(url.protocol === "http:" ? "http" : "https")
  const req = request(
    url,
    {
      method: `POST`,
      headers: {
        "Content-Type": `application/json`,
      },
    },
    res => {
      console.info(`Status Code: ${res.statusCode}`)
      if (res.statusCode >= 400) {
        process.exitCode = 1
      }
      res.pipe(process.stdout)
    }
  )

  req.on("error", e => {
    fail(`Request failed: ${e.message}`)
  })

  // Write data to request body
  req.write(JSON.stringify(body))
  req.end()
}

function fail(str) {
  console.error(str)
  process.exit(1)
}