# Publish & Subscribe (/connectors/gcp-pub-sub/how-to/publish-subscribe) **Publish & subscribe** is the core Pub/Sub round-trip: create a **topic**, attach a **subscription**, publish a message, then pull it back and acknowledge it. The connector maps this directly onto KubeMQ primitives — topic `orders` becomes the Events Store log `gcp.orders`, and subscription `sub-orders` becomes the Queue channel `gcp.sub.sub-orders`. Your Pub/Sub SDK code does not change; you only set `PUBSUB_EMULATOR_HOST` to point at the connector. ## Overview [#overview] `CreateTopic` registers the topic in the connector's registry and binds it to an Events Store log. `CreateSubscription` binds a Queue channel to that topic. `Publish` writes the message **once** to the topic log (the authoritative, replayable source) and fans out one Queue copy per subscription. `Pull` (or `StreamingPull`) delivers the message under an **ack-deadline lease** with an opaque `ack_id`; `Acknowledge(ack_id)` acks it off the subscription. If you never acknowledge, the lease expires and the message is redelivered. | Pub/Sub operation | KubeMQ mapping | Notes | | ---------------------------------- | ---------------------------------------------------------- | --------------------------------------------------- | | `CreateTopic("orders")` | Register Events Store log `gcp.orders` | Topic ids may not start with `sub.` | | `CreateSubscription("sub-orders")` | Bind Queue channel `gcp.sub.sub-orders` | Ack deadline 10..600 s (default 10) | | `Publish` | `SendEventsStore(gcp.orders)` + per-sub `SendQueueMessage` | Returns a server-assigned message id + publish time | | `Pull` / `StreamingPull` | Credit-driven `Get` from the Queue | Each message leased with an opaque `ack_id` | | `Acknowledge(ackId)` | `AckRange` — message removed | Acks the broker sequence under the lease | | ack-deadline expiry (no ack) | `NAckRange` — redelivered | A 250 ms sweeper applies retry backoff | A publish writes once to the topic log and is fanned out per subscription, so the message body and attributes round-trip losslessly. The three reserved tags — `_pubsub_message_id`, `_pubsub_publish_time`, `_pubsub_ordering_key` — are carried across the wire and stripped from `attributes` when delivered back to a Pub/Sub client. ## How it works [#how-it-works] A publisher publishes to a topic; the connector writes the message once to the Events Store log `gcp.{topic}` through the message broker, then fans out one Queue copy per subscription on `gcp.sub.{subscription}`. A subscriber pulls the message under an ack-deadline lease and acknowledges it by `ack_id`. *A publish writes once to the Events Store log `gcp.{topic}` and fans out one Queue copy per subscription on `gcp.sub.{subscription}`; a subscriber pulls each message under an ack-deadline lease and acks it by `ack_id`.* ## The full round-trip [#the-full-round-trip] The lifecycle is `CreateTopic` → `CreateSubscription` → `Publish` → `Pull` → `Acknowledge`. Each client sets only `PUBSUB_EMULATOR_HOST` (default `localhost:8085`) and a project id (parsed but ignored). Most clients auto-detect the emulator from the env var — **C#** needs `EmulatorDetection.EmulatorOnly`, **Ruby** needs an explicit `emulator_host:`, and **Java** points a plaintext `ManagedChannel` at the host with `NoCredentialsProvider`. ```go package main import ( "context" "fmt" "log" "os" "sync" "time" "cloud.google.com/go/pubsub" ) func projectID() string { if v := os.Getenv("PUBSUB_PROJECT_ID"); v != "" { return v } return "my-project" // any id; the project segment is parsed but ignored. } func main() { ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() // PUBSUB_EMULATOR_HOST (default localhost:8085) routes the official client at // the connector over insecure gRPC with credentials cleared — no code change. client, err := pubsub.NewClient(ctx, projectID()) if err != nil { log.Fatalf("NewClient: %v", err) } defer client.Close() // 1. CreateTopic "orders" → Events Store log "gcp.orders". topic, err := client.CreateTopic(ctx, "orders") if err != nil { log.Fatalf("CreateTopic: %v", err) } defer topic.Stop() // 2. CreateSubscription "sub-orders" → Queue channel "gcp.sub.sub-orders". sub, err := client.CreateSubscription(ctx, "sub-orders", pubsub.SubscriptionConfig{ Topic: topic, AckDeadline: 10 * time.Second, // connector default; valid range 10..600s. }) if err != nil { log.Fatalf("CreateSubscription: %v", err) } // 3. Publish one message; the future resolves to the server-assigned id. id, err := topic.Publish(ctx, &pubsub.Message{ Data: []byte("order #4242 — 3x widget"), Attributes: map[string]string{"priority": "express"}, }).Get(ctx) if err != nil { log.Fatalf("Publish: %v", err) } fmt.Printf("published: %s\n", id) // 4. Pull exactly one message via Receive (StreamingPull), then stop the loop. recvCtx, recvCancel := context.WithTimeout(ctx, 15*time.Second) defer recvCancel() var once sync.Once err = sub.Receive(recvCtx, func(_ context.Context, m *pubsub.Message) { fmt.Printf("received: %q attr[priority]=%q\n", string(m.Data), m.Attributes["priority"]) m.Ack() // 5. Acknowledge by ack_id under its lease. once.Do(recvCancel) }) if err != nil && recvCtx.Err() == nil { log.Fatalf("Receive: %v", err) } } ``` ```python import os from google.cloud import pubsub_v1 def project_id() -> str: # Any id works — the project segment is parsed but ignored. return os.environ.get("PUBSUB_PROJECT_ID", "my-project") def main() -> None: # Both clients honour PUBSUB_EMULATOR_HOST (default localhost:8085): they clear # credentials, skip Google auth, and dial insecure gRPC. publisher = pubsub_v1.PublisherClient() subscriber = pubsub_v1.SubscriberClient() proj = project_id() topic_path = publisher.topic_path(proj, "orders") # -> gcp.orders sub_path = subscriber.subscription_path(proj, "sub-orders") # -> gcp.sub.sub-orders # 1. CreateTopic + 2. CreateSubscription. publisher.create_topic(request={"name": topic_path}) subscriber.create_subscription(request={"name": sub_path, "topic": topic_path}) # 3. Publish one message with a user attribute. future = publisher.publish(topic_path, b"order #4242 — 3x widget", priority="express") print(f"published: {future.result(timeout=15)}") # 4. Pull and read the message back. resp = subscriber.pull(request={"subscription": sub_path, "max_messages": 1}, timeout=20) msg = resp.received_messages[0] print(f"received: {msg.message.data.decode()!r} attrs={dict(msg.message.attributes)}") # 5. Acknowledge by ack_id; the message leaves the subscription queue. subscriber.acknowledge(request={"subscription": sub_path, "ack_ids": [msg.ack_id]}) subscriber.close() if __name__ == "__main__": main() ``` ```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 com.google.cloud.pubsub.v1.SubscriptionAdminClient; import com.google.cloud.pubsub.v1.SubscriptionAdminSettings; import com.google.cloud.pubsub.v1.TopicAdminClient; import com.google.cloud.pubsub.v1.TopicAdminSettings; import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub; import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings; import com.google.protobuf.ByteString; import com.google.pubsub.v1.AcknowledgeRequest; import com.google.pubsub.v1.PublishRequest; import com.google.pubsub.v1.PubsubMessage; import com.google.pubsub.v1.PullRequest; import com.google.pubsub.v1.PullResponse; import com.google.pubsub.v1.PushConfig; import com.google.pubsub.v1.ReceivedMessage; import com.google.pubsub.v1.SubscriptionName; import com.google.pubsub.v1.TopicName; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; public final class Main { public static void main(String[] args) throws Exception { String emulatorHost = System.getenv().getOrDefault("PUBSUB_EMULATOR_HOST", "localhost:8085"); String projectId = System.getenv().getOrDefault("PUBSUB_PROJECT_ID", "my-project"); // Java points a plaintext gRPC channel at the host explicitly with NoCredentialsProvider. ManagedChannel channel = ManagedChannelBuilder.forTarget(emulatorHost).usePlaintext().build(); TransportChannelProvider channelProvider = FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel)); NoCredentialsProvider noCreds = NoCredentialsProvider.create(); TopicName topic = TopicName.of(projectId, "orders"); // -> gcp.orders SubscriptionName sub = SubscriptionName.of(projectId, "sub-orders"); // -> gcp.sub.sub-orders try (TopicAdminClient topicAdmin = TopicAdminClient.create(TopicAdminSettings.newBuilder() .setTransportChannelProvider(channelProvider).setCredentialsProvider(noCreds).build()); SubscriptionAdminClient subAdmin = SubscriptionAdminClient.create( SubscriptionAdminSettings.newBuilder() .setTransportChannelProvider(channelProvider) .setCredentialsProvider(noCreds).build()); GrpcSubscriberStub subStub = GrpcSubscriberStub.create(SubscriberStubSettings.newBuilder() .setTransportChannelProvider(channelProvider) .setCredentialsProvider(noCreds).build())) { // 1. CreateTopic + 2. CreateSubscription (ack deadline 10s). topicAdmin.createTopic(topic); subAdmin.createSubscription(sub, topic, PushConfig.getDefaultInstance(), 10); // 3. Publish one message with a user attribute. topicAdmin.publish(PublishRequest.newBuilder() .setTopic(topic.toString()) .addMessages(PubsubMessage.newBuilder() .setData(ByteString.copyFromUtf8("order #4242 — 3x widget")) .putAttributes("priority", "express").build()) .build()); // 4. Pull exactly one message. 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()); // 5. Acknowledge by ack_id. subStub.acknowledgeCallable().call(AcknowledgeRequest.newBuilder() .setSubscription(sub.toString()).addAckIds(got.getAckId()).build()); } finally { channel.shutdown(); } } } ``` ```typescript import { PubSub, v1 } from "@google-cloud/pubsub"; const projectId = process.env["PUBSUB_PROJECT_ID"] ?? "my-project"; // The high-level PubSub client reads PUBSUB_EMULATOR_HOST (default localhost:8085) // and resolves the insecure emulator transport; 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); async function main(): Promise { const topic = publisher.projectTopicsPath(projectId, "orders"); // -> gcp.orders const sub = subscriber.subscriptionPath(projectId, "sub-orders"); // -> gcp.sub.sub-orders // 1. CreateTopic + 2. CreateSubscription (ack deadline 10s). await publisher.createTopic({ name: topic }); await subscriber.createSubscription({ name: sub, topic, ackDeadlineSeconds: 10 }); // 3. Publish one message with a user attribute. const [published] = await publisher.publish({ topic, messages: [{ data: Buffer.from("order #4242 — 3x widget"), attributes: { priority: "express" } }], }); console.log(`published: ${published.messageIds?.[0]}`); // 4. Pull exactly one message. 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")}`); // 5. Acknowledge by ack_id. await subscriber.acknowledge({ subscription: sub, ackIds: [received.ackId!] }); } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```csharp using Google.Api.Gax; using Google.Cloud.PubSub.V1; using Google.Protobuf; var projectId = Environment.GetEnvironmentVariable("PUBSUB_PROJECT_ID") ?? "my-project"; var topicName = TopicName.FromProjectTopic(projectId, "orders"); // -> gcp.orders var subName = SubscriptionName.FromProjectSubscription(projectId, "sub-orders"); // -> gcp.sub.sub-orders // The .NET client does NOT auto-detect the emulator — set EmulatorOnly so each // client reads PUBSUB_EMULATOR_HOST (default localhost:8085) and dials insecurely. var publisher = await new PublisherServiceApiClientBuilder { EmulatorDetection = EmulatorDetection.EmulatorOnly, }.BuildAsync(); var subscriber = await new SubscriberServiceApiClientBuilder { EmulatorDetection = EmulatorDetection.EmulatorOnly, }.BuildAsync(); // 1. CreateTopic + 2. CreateSubscription (ack deadline 10s). await publisher.CreateTopicAsync(topicName); await subscriber.CreateSubscriptionAsync(subName, topicName, pushConfig: null, ackDeadlineSeconds: 10); // 3. Publish one message with a user attribute. var publishResponse = await publisher.PublishAsync(topicName, new[] { new PubsubMessage { Data = ByteString.CopyFromUtf8("order #4242 — 3x widget"), Attributes = { ["priority"] = "express" }, }, }); Console.WriteLine($"published: {publishResponse.MessageIds[0]}"); // 4. Pull exactly one message. var pull = await subscriber.PullAsync(subName, maxMessages: 1); var received = pull.ReceivedMessages[0]; Console.WriteLine($"received: {received.Message.Data.ToStringUtf8()}"); // 5. Acknowledge by ack_id. await subscriber.AcknowledgeAsync(subName, new[] { received.AckId }); ``` ```ruby # frozen_string_literal: true require "google/cloud/pubsub" project_id = ENV["PUBSUB_PROJECT_ID"] || "my-project" # Ruby needs the emulator host passed explicitly (it does not always read the env var). emulator_host = ENV["PUBSUB_EMULATOR_HOST"] || "localhost:8085" pubsub = Google::Cloud::PubSub.new(project_id: project_id, emulator_host: emulator_host) topic_admin = pubsub.topic_admin sub_admin = pubsub.subscription_admin topic_path = pubsub.topic_path("orders") # -> gcp.orders sub_path = pubsub.subscription_path("sub-orders") # -> gcp.sub.sub-orders # 1. CreateTopic + 2. CreateSubscription (ack deadline 10s). topic = topic_admin.create_topic(name: topic_path) sub_admin.create_subscription(name: sub_path, topic: topic_path, ack_deadline_seconds: 10) # 3. Publish one message with a user attribute. publisher = pubsub.publisher(topic.name) msg = publisher.publish("order #4242 — 3x widget", priority: "express") puts "published: #{msg.message_id}" # 4. Pull exactly one message. subscriber = pubsub.subscriber(sub_path) rcv = subscriber.pull(immediate: false, max: 1).first puts "received: #{rcv.data.inspect} attrs=#{rcv.attributes.to_h.inspect}" # 5. Acknowledge by ack_id; the message leaves the subscription queue. rcv.acknowledge! ``` The pulled message carries only **your own attributes** — the three reserved tags (`_pubsub_message_id`, `_pubsub_publish_time`, `_pubsub_ordering_key`) are stamped on the wire for native consumers and stripped from `attributes` before delivery to a Pub/Sub client. The message id and publish time are surfaced through the SDK's own fields, not the attribute map. ## Lease, ack, and redelivery [#lease-ack-and-redelivery] Every delivered message gets an opaque `ack_id` and is held under an **ack-deadline lease** (default 10 s, range 10..600 s). Acknowledging by `ack_id` acks the broker sequence and removes the message. If the deadline passes without an ack, a 250 ms sweeper applies the retry backoff and **redelivers** — `ModifyAckDeadline(0)` is an explicit nack that redelivers immediately, while `ModifyAckDeadline(n)` extends the lease. See [Subscribing](/connectors/gcp-pub-sub/how-to/subscribing) for `Pull` vs `StreamingPull`, flow control, and exactly-once delivery. **Exactly-once and leases are node-local.** An `ack_id` 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. Single-node deployments are unaffected. ## Related [#related]