> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreflux.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Google Cloud Pub/Sub

> Client-mode REST API connector for Google Cloud Pub/Sub: publish base64 messages with attributes and ordering keys, pull, acknowledge, nack or extend, and create topics and subscriptions over HTTPS with a service account OAuth token

Technical Preview

## Overview

A client-mode connector that moves broker data in and out of [Google Cloud Pub/Sub](https://cloud.google.com/pubsub/docs/overview) 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](/latest/lot-language/routes/rest-api-routes).

> [!NOTE]
>
> **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`.

> [!NOTE]
>
> **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

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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":["..."]}`:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "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):

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "project_id": "my-plant-project",
  "subscription": "press-telemetry-coreflux",
  "access_token": "ya29.<service-account-access-token>",
  "body": { "maxMessages": 10 }
}
```

Response shape:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "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** `{}`:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "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:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "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:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "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:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "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:

```bash theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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).

## Related

* [REST API Connectors](/latest/lot-language/routes/rest-api-routes)
* [Pub/Sub REST reference](https://cloud.google.com/pubsub/docs/reference/rest)
* [topics.publish](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/publish)
* [subscriptions.pull](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.subscriptions/pull)
* [Push subscriptions](https://cloud.google.com/pubsub/docs/push)
* [Access control](https://cloud.google.com/pubsub/docs/access-control)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.