Google Cloud Pub/Sub

Technical Preview

Overview

A client-mode connector that moves broker data in and out of Google Cloud Pub/Sub over the REST v1 API. Use it to stream plant telemetry into Google Cloud (Dataflow, BigQuery subscriptions, Cloud Run, Vertex AI) or to pull commands and results back into MQTT, without running a gRPC client.

Event	Request	Notes
Publish	`POST /v1/projects/{p}/topics/{t}:publish`	`messages[]` with base64 `data`, `attributes`, `orderingKey`; returns `messageIds`
Pull	`POST /v1/projects/{p}/subscriptions/{s}:pull`	`{"maxMessages": N}`; returns `receivedMessages[]` with `ackId`
Acknowledge	`POST /v1/projects/{p}/subscriptions/{s}:acknowledge`	`{"ackIds": [...]}` from Pull
ModifyAckDeadline	`POST /v1/projects/{p}/subscriptions/{s}:modifyAckDeadline`	`ackDeadlineSeconds` 0 = nack (redeliver), 10 to 600 = extend
ListTopics	`GET /v1/projects/{p}/topics?pageSize=N`	Smoke / discovery
CreateTopic	`PUT /v1/projects/{p}/topics/{t}`	`{}` or labels / retention; 409 if it exists
CreateSubscription	`PUT /v1/projects/{p}/subscriptions/{s}`	`topic` full name, `ackDeadlineSeconds`, `enableMessageOrdering`, optional `pushConfig`

Default BASE_ADDRESS is https://pubsub.googleapis.com (no trailing slash). project_id, topic and subscription are short ids taken from the MQTT payload. Topics use data/google-cloud-pub-sub/<object>/<verb>; replies land on <topic>/response. See the base guide: REST API Connectors.

Auth: Authorization: Bearer {value.json.access_token}, an OAuth 2.0 access token minted from a service account with scope https://www.googleapis.com/auth/pubsub (or cloud-platform). Grant roles/pubsub.publisher to publish and roles/pubsub.subscriber to pull and acknowledge; creating topics and subscriptions needs roles/pubsub.editor. Keep the key in Hub secrets (suggested name GCP_PUBSUB_ACCESS_TOKEN) and refresh the token upstream (tokens last about one hour). Never commit keys into the .lot.

Base64 is your job. Pub/Sub requires data to be base64. The connector forwards the payload as-is and does not encode, so encode upstream (for example in a Python script called from an action; base64 is available) or have the device publish base64. Pulled messages come back base64-encoded too. This template has been smoke-tested against the Pub/Sub emulator only, not yet against a live Google Cloud project.

Basic Syntax

DEFINE ROUTE GoogleCloudPubSubRestApi WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH BASE_ADDRESS "https://pubsub.googleapis.com"
        WITH ENABLE_CLIENT "true"
        WITH USE_SSL "true"
        WITH REQUEST_TIMEOUT '120'
        WITH DEFAULT_CONTENT_TYPE "application/json"
        # Smoke: GET /v1/projects/{p}/topics?pageSize=1 → 200. Auth: Bearer {value.json.access_token}. Never commit tokens.
    ADD EVENT Publish
        WITH SOURCE_TOPIC "data/google-cloud-pub-sub/messages/publish"
        WITH DESTINATION_TOPIC "data/google-cloud-pub-sub/messages/publish/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/v1/projects/{value.json.project_id}/topics/{value.json.topic}:publish"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT Pull
        WITH SOURCE_TOPIC "data/google-cloud-pub-sub/messages/pull"
        WITH DESTINATION_TOPIC "data/google-cloud-pub-sub/messages/pull/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/v1/projects/{value.json.project_id}/subscriptions/{value.json.subscription}:pull"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT Acknowledge
        WITH SOURCE_TOPIC "data/google-cloud-pub-sub/messages/ack"
        WITH DESTINATION_TOPIC "data/google-cloud-pub-sub/messages/ack/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/v1/projects/{value.json.project_id}/subscriptions/{value.json.subscription}:acknowledge"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT ModifyAckDeadline
        WITH SOURCE_TOPIC "data/google-cloud-pub-sub/messages/ack-deadline"
        WITH DESTINATION_TOPIC "data/google-cloud-pub-sub/messages/ack-deadline/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/v1/projects/{value.json.project_id}/subscriptions/{value.json.subscription}:modifyAckDeadline"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT ListTopics
        WITH SOURCE_TOPIC "data/google-cloud-pub-sub/topics/list"
        WITH DESTINATION_TOPIC "data/google-cloud-pub-sub/topics/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/v1/projects/{value.json.project_id}/topics?pageSize={value.json.page_size}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT CreateTopic
        WITH SOURCE_TOPIC "data/google-cloud-pub-sub/topics/create"
        WITH DESTINATION_TOPIC "data/google-cloud-pub-sub/topics/create/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/v1/projects/{value.json.project_id}/topics/{value.json.topic}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT CreateSubscription
        WITH SOURCE_TOPIC "data/google-cloud-pub-sub/subscriptions/create"
        WITH DESTINATION_TOPIC "data/google-cloud-pub-sub/subscriptions/create/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/v1/projects/{value.json.project_id}/subscriptions/{value.json.subscription}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"

MQTT examples

Publish (data/google-cloud-pub-sub/messages/publish), expect 200 {"messageIds":["..."]}:

{
  "project_id": "my-plant-project",
  "topic": "press-telemetry",
  "access_token": "ya29.<service-account-access-token>",
  "body": {
    "messages": [
      {
        "data": "eyJEZXZpY2VJZCI6InByZXNzLTAxIiwidGVtcCI6MzcuMn0=",
        "attributes": { "line": "A", "source": "coreflux" },
        "orderingKey": "press-01"
      }
    ]
  }
}

data above is base64 of {"DeviceId":"press-01","temp":37.2}. Up to 1000 messages and 10 MB per request. Pull (data/google-cloud-pub-sub/messages/pull), expect 200 with receivedMessages (or {} when nothing is waiting):

{
  "project_id": "my-plant-project",
  "subscription": "press-telemetry-coreflux",
  "access_token": "ya29.<service-account-access-token>",
  "body": { "maxMessages": 10 }
}

Response shape:

{
  "receivedMessages": [
    {
      "ackId": "RkhRNxkIaFEOT14jPzUgKEUSA...",
      "message": {
        "data": "eyJEZXZpY2VJZCI6InByZXNzLTAxIiwidGVtcCI6MzcuMn0=",
        "attributes": { "line": "A", "source": "coreflux" },
        "messageId": "1234567890",
        "publishTime": "2026-10-05T00:40:00.123Z",
        "orderingKey": "press-01"
      }
    }
  ]
}

Acknowledge (data/google-cloud-pub-sub/messages/ack), expect 200 {}:

{
  "project_id": "my-plant-project",
  "subscription": "press-telemetry-coreflux",
  "access_token": "ya29.<service-account-access-token>",
  "body": { "ackIds": ["RkhRNxkIaFEOT14jPzUgKEUSA..."] }
}

ModifyAckDeadline (data/google-cloud-pub-sub/messages/ack-deadline), expect 200 {}. Use 0 to nack so the message is redelivered right away:

{
  "project_id": "my-plant-project",
  "subscription": "press-telemetry-coreflux",
  "access_token": "ya29.<service-account-access-token>",
  "body": { "ackIds": ["RkhRNxkIaFEOT14jPzUgKEUSA..."], "ackDeadlineSeconds": 0 }
}

CreateTopic (data/google-cloud-pub-sub/topics/create), expect 200 with the topic resource:

{
  "project_id": "my-plant-project",
  "topic": "press-telemetry",
  "access_token": "ya29.<service-account-access-token>",
  "body": { "labels": { "site": "porto" } }
}

CreateSubscription (data/google-cloud-pub-sub/subscriptions/create), expect 200 with the subscription resource:

{
  "project_id": "my-plant-project",
  "subscription": "press-telemetry-coreflux",
  "access_token": "ya29.<service-account-access-token>",
  "body": {
    "topic": "projects/my-plant-project/topics/press-telemetry",
    "ackDeadlineSeconds": 30,
    "enableMessageOrdering": true
  }
}

ListTopics (data/google-cloud-pub-sub/topics/list): {"project_id": "my-plant-project", "page_size": "50", "access_token": "ya29.<service-account-access-token>"}. body becomes the HTTP body, so control fields (project_id, topic, subscription, access_token, page_size) are never sent to Google.

Getting an access token

For a quick test, with the service account active in the Google Cloud CLI:

gcloud auth print-access-token --impersonate-service-account=coreflux-pubsub@my-plant-project.iam.gserviceaccount.com

In production, mint the token from the service account key (JWT bearer grant to https://oauth2.googleapis.com/token) in a helper that reads the key from Hub secrets, and refresh it before the one-hour expiry.

Push subscriptions (companion, not packaged)

Pub/Sub can also push messages to an HTTPS endpoint (pushConfig.pushEndpoint on CreateSubscription). Each delivery is a POST with a JSON envelope

{"message": {"data", "attributes", "messageId", "publishTime", "orderingKey"}, "subscription": "projects/.../subscriptions/..."}

, and a 200, 201, 202, 204 or 102 reply acknowledges it. Receiving pushes needs a separate ENABLE_SERVER REST API connector on a publicly reachable HTTPS address (optionally verifying the OIDC token Pub/Sub adds to Authorization). That companion connector is not part of this template.

Notes

Strict ordering by orderingKey needs enableMessageOrdering on the subscription and publishing through one regional endpoint (for example https://europe-west1-pubsub.googleapis.com).
Pull can hold the request open for a bounded time when the subscription is empty, so REQUEST_TIMEOUT is 120. returnImmediately is deprecated.
The emulator (gcloud beta emulators pubsub start) speaks the same REST API on http://localhost:8085, ignores auth, and is handy for local tests. Set USE_SSL "false" and point BASE_ADDRESS at it.
Not covered: deletes, seek and snapshots, schemas, IAM policy calls, StreamingPull (gRPC only).
Suggested branches: feat/rest-api-route-google-cloud-pub-sub, feat/docs-rest-api-google-cloud-pub-sub (Hugo pushes).

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

Overview

Basic Syntax

MQTT examples

Getting an access token

Push subscriptions (companion, not packaged)

Notes

​Overview

​Basic Syntax

​MQTT examples

​Getting an access token

​Push subscriptions (companion, not packaged)

​Notes

​Related

Overview

Basic Syntax

MQTT examples

Getting an access token

Push subscriptions (companion, not packaged)

Notes

Related