# Getting Started (/connectors/gcp-pub-sub/tutorials/getting-started)



Get a message flowing through the KubeMQ Google Cloud Pub/Sub connector in minutes. You point
a standard Pub/Sub SDK at the connector's gRPC endpoint, create a topic and a subscription,
publish a message, and pull it back — all over the genuine Pub/Sub v1 wire protocol, with no
emulator to install and no KubeMQ SDK. The only change versus a real-GCP app is one environment
variable: `PUBSUB_EMULATOR_HOST`.

## Prerequisites [#prerequisites]

* A running **kubemq-server** with the Pub/Sub connector **enabled** and reachable on **gRPC port 8085**.
  The connector is **opt-in (disabled by default)** — see the enable step below.
* One of the first-party Google Cloud Pub/Sub clients below for your language. There is no
  KubeMQ SDK; you use the official Google client with only the emulator host set.
* **No credentials.** When `PUBSUB_EMULATOR_HOST` is set, the SDK clears its Google
  credentials, skips Google auth, and dials insecure gRPC — exactly as against Google's local
  emulator.

## Enable the connector [#enable-the-connector]

The GCP Pub/Sub connector is **disabled by default** — a stock kubemq-server does **not**
bind gRPC port 8085 until you turn it on. Enable it with its enable variable:

<RunKubeMQ ports="[8085, 50000]" env="{ CONNECTORS_GCP_ENABLE: 'true' }" />

<Callout type="warn">
  **The enable variable is `CONNECTORS_GCP_ENABLE`.** A stock server does not serve Pub/Sub
  until you set this to `true`. For Kubernetes, set `spec.gcp.enabled: true` in the
  `KubemqCluster` CR.
</Callout>

## Connect the SDK [#connect-the-sdk]

Every official Pub/Sub client library and `gcloud` honour the standard `PUBSUB_EMULATOR_HOST`
environment variable. Export it (and any project id) before running your app:

```bash
export PUBSUB_EMULATOR_HOST=localhost:8085   # connector default gRPC port; SDK uses the insecure path
export PUBSUB_PROJECT_ID=my-project          # any id; the project segment is parsed but ignored
# Some clients and gcloud also read this alias:
# export GOOGLE_CLOUD_PROJECT=my-project
```

<Callout type="info">
  The &#x2A;*project id is parsed but ignored.** The connector is single-tenant (like the emulator),
  so resource ids are global across projects — topic `orders` is always the Events Store log
  `gcp.orders` regardless of the project segment. Any project id works.
</Callout>

## How it works [#how-it-works]

`CreateTopic("orders")` maps the topic to the KubeMQ Events Store log `gcp.orders`;
`CreateSubscription("sub-orders")` maps the subscription to the Queue channel
`gcp.sub.sub-orders`. `Publish` writes once to the topic log through the message broker, then
fans out one Queue copy per subscription; `Pull` returns the message plus an opaque `ack_id`,
and `Acknowledge` removes it from the subscription queue.

<Mermaid
  chart="`
graph LR
APP[&#x22;Pub/Sub SDK<br/>PUBSUB_EMULATOR_HOST=:8085&#x22;]
CONN[&#x22;Pub/Sub connector&#x22;]
LOG[(&#x22;Events Store log<br/>gcp.orders&#x22;)]
Q{{&#x22;Queue channel<br/>gcp.sub.sub-orders&#x22;}}
BROKER[&#x22;Message Broker&#x22;]

APP -- &#x22;CreateTopic / Publish&#x22; --> CONN
CONN -- &#x22;SendEventsStore&#x22; --> LOG
LOG -- &#x22;fan-out&#x22; --> Q
LOG --> BROKER
Q --> BROKER
BROKER -. &#x22;Pull + ack_id&#x22; .-> CONN
CONN -. &#x22;message&#x22; .-> APP

class APP client
class CONN connector
class LOG,Q,BROKER broker
`"
/>

*The topic `orders` maps to the Events Store log `gcp.orders`; a publish fans out one copy to the subscription queue `gcp.sub.sub-orders`, and the pull returns it with a node-local `ack_id`.*

## Steps [#steps]

<Steps>
  <Step>
    ### Point the SDK at the connector [#point-the-sdk-at-the-connector]

    Build a standard Google Cloud Pub/Sub client. Most clients auto-detect the emulator from
    `PUBSUB_EMULATOR_HOST` (default `localhost:8085`); three need a one-line opt-in — &#x2A;*C#** sets
    `EmulatorDetection.EmulatorOnly`, **Ruby** passes `emulator_host:`, and **Java** points a
    plaintext `ManagedChannel` at the host with `NoCredentialsProvider`.

    <Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby']">
      <Tab value="Go">
        ```go
        import (
        	"context"
        	"os"

        	"cloud.google.com/go/pubsub"
        )

        // pubsub.NewClient auto-reads PUBSUB_EMULATOR_HOST and dials insecurely — no flag.
        client, err := pubsub.NewClient(ctx, os.Getenv("PUBSUB_PROJECT_ID"))
        ```
      </Tab>

      <Tab value="Python">
        ```python
        from google.cloud import pubsub_v1

        # Both clients honour PUBSUB_EMULATOR_HOST automatically.
        publisher = pubsub_v1.PublisherClient()
        subscriber = pubsub_v1.SubscriberClient()
        ```
      </Tab>

      <Tab value="Java">
        ```java
        import com.google.api.gax.core.NoCredentialsProvider;
        import com.google.api.gax.grpc.GrpcTransportChannel;
        import com.google.api.gax.rpc.FixedTransportChannelProvider;
        import com.google.api.gax.rpc.TransportChannelProvider;
        import io.grpc.ManagedChannel;
        import io.grpc.ManagedChannelBuilder;

        String emulatorHost = System.getenv().getOrDefault("PUBSUB_EMULATOR_HOST", "localhost:8085");

        // Java needs the emulator host wired explicitly: a plaintext channel + no credentials.
        ManagedChannel channel = ManagedChannelBuilder.forTarget(emulatorHost).usePlaintext().build();
        TransportChannelProvider channelProvider =
                FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel));
        NoCredentialsProvider noCreds = NoCredentialsProvider.create();
        ```
      </Tab>

      <Tab value="JavaScript">
        ```typescript
        import { PubSub, v1 } from "@google-cloud/pubsub";

        const projectId = process.env["PUBSUB_PROJECT_ID"] ?? "my-project";

        // The high-level client reads PUBSUB_EMULATOR_HOST; reuse its options for the v1 clients.
        const baseOptions = new PubSub({ projectId }).options;
        const options = { ...baseOptions, port: baseOptions.port === undefined ? undefined : Number(baseOptions.port) };
        const publisher = new v1.PublisherClient(options);
        const subscriber = new v1.SubscriberClient(options);
        ```
      </Tab>

      <Tab value="C#">
        ```csharp
        using Google.Api.Gax;
        using Google.Cloud.PubSub.V1;

        // The .NET client does NOT auto-detect the emulator — set EmulatorOnly explicitly.
        var publisher = await new PublisherServiceApiClientBuilder
        {
            EmulatorDetection = EmulatorDetection.EmulatorOnly,
        }.BuildAsync();
        var subscriber = await new SubscriberServiceApiClientBuilder
        {
            EmulatorDetection = EmulatorDetection.EmulatorOnly,
        }.BuildAsync();
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby
        require "google/cloud/pubsub"

        # Ruby needs emulator_host passed explicitly (it does not always read the env var).
        pubsub = Google::Cloud::PubSub.new(
          project_id:    ENV["PUBSUB_PROJECT_ID"] || "my-project",
          emulator_host: ENV["PUBSUB_EMULATOR_HOST"] || "localhost:8085"
        )
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Create a topic and publish [#create-a-topic-and-publish]

    `CreateTopic("orders")` registers the topic and maps it to the Events Store log `gcp.orders`;
    `CreateSubscription("sub-orders")` maps to the Queue channel `gcp.sub.sub-orders`. `Publish`
    writes the message once to the topic log and returns a server-assigned message id.

    <Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby']">
      <Tab value="Go">
        ```go
        topic, _ := client.CreateTopic(ctx, "orders")                 // -> gcp.orders
        sub, _ := client.CreateSubscription(ctx, "sub-orders", pubsub.SubscriptionConfig{
        	Topic:       topic,
        	AckDeadline: 10 * time.Second, // connector default; valid range 10..600s.
        }) // -> gcp.sub.sub-orders

        id, _ := topic.Publish(ctx, &pubsub.Message{Data: []byte("hello kubemq")}).Get(ctx)
        fmt.Println("published:", id)
        ```
      </Tab>

      <Tab value="Python">
        ```python
        topic_path = publisher.topic_path(proj, "orders")              # -> gcp.orders
        sub_path = subscriber.subscription_path(proj, "sub-orders")    # -> gcp.sub.sub-orders

        publisher.create_topic(request={"name": topic_path})
        subscriber.create_subscription(request={"name": sub_path, "topic": topic_path})

        future = publisher.publish(topic_path, b"hello kubemq")
        print("published:", future.result(timeout=15))
        ```
      </Tab>

      <Tab value="Java">
        ```java
        TopicName topic = TopicName.of(projectId, "orders");                 // -> gcp.orders
        SubscriptionName sub = SubscriptionName.of(projectId, "sub-orders"); // -> gcp.sub.sub-orders

        topicAdmin.createTopic(topic);
        subAdmin.createSubscription(sub, topic, PushConfig.getDefaultInstance(), 10);

        topicAdmin.publish(PublishRequest.newBuilder()
                .setTopic(topic.toString())
                .addMessages(PubsubMessage.newBuilder()
                        .setData(ByteString.copyFromUtf8("hello kubemq")).build())
                .build());
        ```
      </Tab>

      <Tab value="JavaScript">
        ```typescript
        const topic = publisher.projectTopicsPath(projectId, "orders");    // -> gcp.orders
        const sub = subscriber.subscriptionPath(projectId, "sub-orders");  // -> gcp.sub.sub-orders

        await publisher.createTopic({ name: topic });
        await subscriber.createSubscription({ name: sub, topic, ackDeadlineSeconds: 10 });

        const [published] = await publisher.publish({
          topic,
          messages: [{ data: Buffer.from("hello kubemq") }],
        });
        console.log("published:", published.messageIds?.[0]);
        ```
      </Tab>

      <Tab value="C#">
        ```csharp
        var topicName = TopicName.FromProjectTopic(projectId, "orders");                 // -> gcp.orders
        var subName = SubscriptionName.FromProjectSubscription(projectId, "sub-orders"); // -> gcp.sub.sub-orders

        await publisher.CreateTopicAsync(topicName);
        await subscriber.CreateSubscriptionAsync(subName, topicName, pushConfig: null, ackDeadlineSeconds: 10);

        var published = await publisher.PublishAsync(topicName, new[]
        {
            new PubsubMessage { Data = ByteString.CopyFromUtf8("hello kubemq") },
        });
        Console.WriteLine($"published: {published.MessageIds[0]}");
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby
        topic_path = pubsub.topic_path("orders")            # -> gcp.orders
        sub_path   = pubsub.subscription_path("sub-orders") # -> gcp.sub.sub-orders

        topic = pubsub.topic_admin.create_topic(name: topic_path)
        pubsub.subscription_admin.create_subscription(name: sub_path, topic: topic_path, ack_deadline_seconds: 10)

        msg = pubsub.publisher(topic.name).publish("hello kubemq")
        puts "published: #{msg.message_id}"
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Pull and acknowledge [#pull-and-acknowledge]

    `Pull` returns the message plus an opaque `ack_id`, holding it under an ack-deadline lease.
    `Acknowledge(ack_id)` removes it from the subscription queue. A successful run prints the body
    you published.

    <Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby']">
      <Tab value="Go">
        ```go
        recvCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
        defer cancel()
        sub.Receive(recvCtx, func(_ context.Context, m *pubsub.Message) {
        	fmt.Printf("received: %q\n", string(m.Data))
        	m.Ack() // Acknowledge by ack_id under its lease.
        	cancel()
        })
        ```
      </Tab>

      <Tab value="Python">
        ```python
        resp = subscriber.pull(request={"subscription": sub_path, "max_messages": 1}, timeout=20)
        msg = resp.received_messages[0]
        print("received:", msg.message.data.decode())

        subscriber.acknowledge(request={"subscription": sub_path, "ack_ids": [msg.ack_id]})
        ```
      </Tab>

      <Tab value="Java">
        ```java
        PullResponse resp = subStub.pullCallable().call(PullRequest.newBuilder()
                .setSubscription(sub.toString()).setMaxMessages(1).build());
        ReceivedMessage got = resp.getReceivedMessages(0);
        System.out.printf("received: %s%n", got.getMessage().getData().toStringUtf8());

        subStub.acknowledgeCallable().call(AcknowledgeRequest.newBuilder()
                .setSubscription(sub.toString()).addAckIds(got.getAckId()).build());
        ```
      </Tab>

      <Tab value="JavaScript">
        ```typescript
        const [pull] = await subscriber.pull({ subscription: sub, maxMessages: 1 });
        const received = pull.receivedMessages![0];
        console.log("received:", Buffer.from(received.message!.data as Uint8Array).toString("utf8"));

        await subscriber.acknowledge({ subscription: sub, ackIds: [received.ackId!] });
        ```
      </Tab>

      <Tab value="C#">
        ```csharp
        var pull = await subscriber.PullAsync(subName, maxMessages: 1);
        var received = pull.ReceivedMessages[0];
        Console.WriteLine($"received: {received.Message.Data.ToStringUtf8()}");

        await subscriber.AcknowledgeAsync(subName, new[] { received.AckId });
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby
        subscriber = pubsub.subscriber(sub_path)
        rcv = subscriber.pull(immediate: false, max: 1).first
        puts "received: #{rcv.data.inspect}"

        rcv.acknowledge!
        ```
      </Tab>
    </Tabs>

    <Callout type="info">
      The `ack_id` is **node-local** under `StreamingPull` — a lease minted on one node is invalid on
      another. In a clustered deployment, pin a subscription's `StreamingPull` traffic to one node
      (session-affinity load balancer), or accept at-least-once delivery across nodes. See
      [Subscribing](/connectors/gcp-pub-sub/how-to/subscribing).
    </Callout>
  </Step>
</Steps>

## Next steps [#next-steps]

<Cards>
  <Card title="Configuration" href="/connectors/gcp-pub-sub/concepts/configuration" description="The opt-in CONNECTORS_GCP_ENABLE variable, gRPC port 8085, and the thirteen CONNECTORS_GCP_* settings." />

  <Card title="Architecture" href="/connectors/gcp-pub-sub/concepts/architecture" description="The gRPC emulator listener, the 38-RPC surface, and how topics and subscriptions map to KubeMQ primitives." />

  <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="Connectivity & emulator mode" href="/connectors/gcp-pub-sub/how-to/connectivity-and-emulator-mode" description="The PUBSUB_EMULATOR_HOST drop-in, per-language emulator opt-in, and the insecure-gRPC posture." />
</Cards>