# Architecture (/connectors/gcp-pub-sub/concepts/architecture)



The KubeMQ **Google Cloud Pub/Sub connector** is an embedded, wire-protocol bridge inside
kubemq-server that speaks the genuine Pub/Sub v1 **gRPC** services on a dedicated gRPC listener
(default TCP **8085**, the Pub/Sub emulator convention). The connector is **opt-in (disabled by
default)** — enable it with `CONNECTORS_GCP_ENABLE=true` (Docker) or `spec.gcp.enabled: true`
(Kubernetes). It runs in **emulator mode**: no authentication, no TLS, insecure gRPC — exactly
like Google's local emulator. Any standard Pub/Sub client connects to it by setting one
environment variable, `PUBSUB_EMULATOR_HOST` — no code changes, no library swap, no emulator
to install.

Two KubeMQ primitives back the model:

* **A topic** maps onto a native KubeMQ **Events Store** log `gcp.{topic}` — the authoritative,
  cross-protocol, replayable source of truth.
* **A subscription** maps onto a native KubeMQ **Queue** channel `gcp.sub.{subscription}` — one
  filtered copy per subscription.

A publish is written **once** to the topic log, then fanned out to one Queue copy per
subscription. The connector touches the **Events Store** and **Queue** primitives only — it does
**not** map onto KubeMQ's commands / queries / events RPC patterns, and there is **no
request/reply** anywhere.

## The gRPC emulator listener [#the-grpc-emulator-listener]

A single gRPC server on `Connectors.Gcp.Port` (default **8085**) implements the real Pub/Sub v1
wire protocol — **38 RPCs** across four services. Each request passes a fixed three-stage
interceptor chain — **Recovery → Logger → Traffic-gate** — and there is **no auth interceptor**
(emulator mode):

| Service                          | RPCs | What it covers                                                                                               |
| -------------------------------- | ---- | ------------------------------------------------------------------------------------------------------------ |
| `google.pubsub.v1.Publisher`     | 9    | Topics and publish (`CreateTopic`, batch `Publish`, `UpdateTopic`, `DeleteTopic`, listing).                  |
| `google.pubsub.v1.Subscriber`    | 16   | Subscriptions, `Pull` / `StreamingPull`, `Acknowledge`, `ModifyAckDeadline`, push config, snapshots, `Seek`. |
| `google.pubsub.v1.SchemaService` | 10   | Avro and Protobuf schema definitions, revisions, and validation.                                             |
| `google.iam.v1.IAMPolicy`        | 3    | Permissive stubs — no enforcement (emulator parity).                                                         |

The **Traffic-gate** interceptor short-circuits requests with transient `UNAVAILABLE` while the
broker is not ready; on a not-ready → ready transition the connector **drops all in-memory
leases** (their downstream transactions are dead) and the poller rebuilds. SDKs see the
transient `UNAVAILABLE` and retry. A nil auth seam is reserved so a future release can add token
validation without touching handlers. See
[Capabilities](/connectors/gcp-pub-sub/reference/capabilities) for the full RPC matrix.

## How topics & subscriptions map to KubeMQ [#how-topics--subscriptions-map-to-kubemq]

A request arrives at the gRPC listener; the handler parses the resource, dispatches to the
matching service, and lands on a KubeMQ primitive. A `Publish` writes once to the topic's
Events Store log and fans out one Queue copy per bound subscription, applying each
subscription's filter.

<Mermaid
  chart="`
graph TB
CLIENT[&#x22;Pub/Sub SDK client<br/>PUBSUB_EMULATOR_HOST=host:8085&#x22;]
LISTEN[&#x22;gRPC listener :8085<br/>(insecure, emulator mode)&#x22;]
CHAIN[&#x22;Interceptors<br/>Recovery → Logger → Traffic-gate&#x22;]
PUB[&#x22;Publisher / Subscriber /<br/>SchemaService / IAMPolicy (38 RPCs)&#x22;]
LOG[(&#x22;Events Store log<br/>gcp.{topic}&#x22;)]
Q1{{&#x22;Queue channel<br/>gcp.sub.sub-a&#x22;}}
Q2{{&#x22;Queue channel<br/>gcp.sub.sub-b&#x22;}}
REG[&#x22;Registry record<br/>(snapshots / schemas)&#x22;]
BROKER[&#x22;Message Broker&#x22;]

CLIENT --> LISTEN
LISTEN --> CHAIN
CHAIN --> PUB
PUB -- &#x22;Publish (write-once)&#x22; --> LOG
PUB -- &#x22;snapshot / schema&#x22; --> REG
LOG -- &#x22;fan-out + filter&#x22; --> Q1
LOG -- &#x22;fan-out + filter&#x22; --> Q2
LOG --> BROKER
Q1 --> BROKER
Q2 --> BROKER

class CLIENT client
class LISTEN,CHAIN,PUB connector
class LOG,Q1,Q2,BROKER broker
class REG connector
`"
/>

*A publish writes once to the Events Store log `gcp.{topic}` and fans out one filtered Queue copy per subscription on `gcp.sub.{subscription}`, all backed by the message broker. Snapshots and schemas are registry records with no channel.*

The channel mapping is the single most important mental model:

| Concept                      | Behavior                                                                                                                                                                                                      |
| ---------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Listener**                 | A dedicated gRPC server on `Connectors.Gcp.Port` (default **8085**). Insecure gRPC, no auth, no TLS — the emulator contract.                                                                                  |
| **Topic → Events Store log** | Topic `projects/{p}/topics/{t}` ↔ native KubeMQ **Events Store** log `gcp.{t}`. A publish writes **once** here (`Array.SendEventsStore`) — the authoritative, replayable, cross-protocol source.              |
| **Subscription → Queue**     | Subscription `projects/{p}/subscriptions/{s}` ↔ native KubeMQ **Queue** channel `gcp.sub.{s}`. One Queue copy is fanned out per subscription (`Array.SendQueueMessage`), applying that subscription's filter. |
| **Ordering-key channel**     | A keyed message fans onto its own channel `gcp.sub.{sub}.k.{enc(key)}` (sha256 fallback `gcp.sub.{sub}.h.{hash}`) for at-most-one-in-flight per key.                                                          |
| **Snapshots / schemas**      | A per-node replicated registry record — no native channel.                                                                                                                                                    |
| **Project segment**          | `{p}` is parsed and validated but **ignored** — the connector is single-tenant, so resource ids are global across projects.                                                                                   |
| **Reserved namespace**       | Topic ids may **not** start with `sub.` — it collides with the `gcp.sub.*` subscription-queue namespace.                                                                                                      |

See [Channel mapping](/connectors/gcp-pub-sub/reference/channel-mapping) for the full
grammar and [Configuration](/connectors/gcp-pub-sub/concepts/configuration) for the connector knobs.

## Publish once, then fan out [#publish-once-then-fan-out]

A `Publish` is the heart of the model and follows a strict order:

1. **Validate the whole batch before enqueuing anything** (atomicity): batch size 1..1000; per
   message — total ≤ 10 MiB, ≤ 100 attributes, attribute key ≤ 256 B (no `goog` prefix),
   attribute value ≤ 1024 B, ordering key ≤ 1024 B, and `data` **or** `attributes` non-empty.
   Any failure rejects the entire batch with `INVALID_ARGUMENT` and nothing is published.
2. **Schema enforcement** — if the topic references a schema, every message is validated against
   it; the whole batch is rejected on the first non-conforming message.
3. **Topic-log write** — each message is written once to the Events Store log `gcp.{topic}`
   (`Array.SendEventsStore`), assigning a server message id and publish time. This single record
   is the authoritative, cross-protocol, replayable copy and the source for `Seek`.
4. **Fan-out** — one Queue copy per subscription (`Array.SendQueueMessage`), **applying each
   subscription's filter**; a filtered-out message is never enqueued. Detached subscriptions are
   skipped.

Delivery is then driven from each subscription's queue: every delivered message gets an opaque
`ack_id` under an ack-deadline lease; a 250 ms sweeper expires overdue leases, applies retry
backoff, and redelivers — or dead-letters once the receive count exceeds the policy. See
[Publishing](/connectors/gcp-pub-sub/how-to/publishing) and
[Subscribing](/connectors/gcp-pub-sub/how-to/subscribing).

## Reserved tags & project segment [#reserved-tags--project-segment]

A `PubsubMessage` becomes a KubeMQ message whose body is `data` and whose tags are the message
`attributes` plus **three reserved tags** carried across the wire:

| Reserved tag           | Carries                         |
| ---------------------- | ------------------------------- |
| `_pubsub_message_id`   | The server-assigned message id. |
| `_pubsub_publish_time` | The publish timestamp.          |
| `_pubsub_ordering_key` | The ordering key (when set).    |

Native KubeMQ consumers **see** these tags; they are &#x2A;*stripped from `attributes`** when a
message is delivered back to a Pub/Sub client. The `{p}` project segment is parsed and validated
but ignored (single-tenant), and the `sub.` topic-id prefix is reserved.

## Cross-protocol interop [#cross-protocol-interop]

Because every topic is a normal KubeMQ Events Store log, a Pub/Sub `Publish` to topic `orders`
(written to `gcp.orders`) is consumable by a **native** KubeMQ Events Store subscriber on the
same channel — and the native side sees the three reserved `_pubsub_*` tags that are stripped
for Pub/Sub clients. This lets you migrate one side at a time, or run Pub/Sub-SDK producers
alongside native KubeMQ consumers.

<Mermaid
  chart="`
graph LR
GCP[&#x22;Pub/Sub SDK client<br/>topic: orders&#x22;]
CONN[&#x22;Pub/Sub connector&#x22;]
LOG[(&#x22;Events Store log<br/>gcp.orders&#x22;)]
NATIVE[&#x22;Native KubeMQ client<br/>SubscribeToEventsStore(gcp.orders)&#x22;]
BROKER[&#x22;Message Broker&#x22;]

GCP -- &#x22;Publish&#x22; --> CONN
CONN -- &#x22;SendEventsStore&#x22; --> LOG
LOG --> BROKER
BROKER -. &#x22;deliver (with _pubsub_* tags)&#x22; .-> NATIVE

class GCP,NATIVE client
class CONN connector
class LOG,BROKER broker
`"
/>

*A Pub/Sub publish lands on the Events Store log `gcp.orders`; a native KubeMQ Events Store subscriber reads the same record, including the reserved `_pubsub_*` tags.*

<Callout type="info">
  Topic, subscription, snapshot, and schema **records** are synchronized across cluster nodes with
  a **last-writer-wins** rule (a per-node replicated registry). Message **data** itself rides the
  existing Events Store / Queues replication. Exactly-once tokens and `StreamingPull` leases are
  **node-local** — pin an exactly-once subscription's `StreamingPull` to one node, or accept
  at-least-once across nodes. See
  [Reliability](/connectors/gcp-pub-sub/how-to/reliability).
</Callout>

## Related [#related]

<Cards>
  <Card title="Channel mapping" href="/connectors/gcp-pub-sub/reference/channel-mapping" description="The gcp.{topic} and gcp.sub.{subscription} grammar, the reserved tags, and the ignored project segment." />

  <Card title="Publish & Subscribe" href="/connectors/gcp-pub-sub/how-to/publish-subscribe" description="The core topic-to-subscription round-trip with complete multi-language code." />

  <Card title="Cross-Protocol Interop" href="/connectors/gcp-pub-sub/concepts/cross-protocol-interop" description="Publish with a Pub/Sub SDK and consume natively on the same Events Store log gcp.{topic}." />

  <Card title="Capabilities" href="/connectors/gcp-pub-sub/reference/capabilities" description="The 38 v1 RPCs, the delivery features, and the operations the connector ignores or rejects." />
</Cards>