# Fan-Out (/connectors/gcp-pub-sub/how-to/fan-out) **Fan-out** delivers one published message to many independent consumers. A single `Publish` writes the message **once** to the topic's Events Store log `gcp.{topic}`, then the connector fans out one copy to **each** subscription's Queue channel `gcp.sub.{subscription}` — applying that subscription's filter. Every subscription has its own backlog and its own ack state, so a slow or filtered consumer never blocks the others. This is the classic "one event, many consumers" pattern: an order-placed event reaches a billing subscription, a shipping subscription, and an analytics subscription from a single publish. ## Overview [#overview] `CreateTopic` once, then `CreateSubscription` for each consumer that should receive a copy. `Publish` resolves, at publish time, to **every** subscription bound to the topic and enqueues one copy per subscription (skipping ones whose filter does not match, and detached ones). Each subscription is pulled and acknowledged independently. | Step | Action | Behavior | | --------------------- | ------------------------------ | --------------------------------------------------------------------- | | Topic | `CreateTopic("events")` | Events Store log `gcp.events` | | Subscriptions | `CreateSubscription` ×N | Each binds a Queue channel `gcp.sub.{name}` | | Publish | `Publish` | One write to `gcp.events`; one Queue copy fanned out per subscription | | Filtered subscription | `CreateSubscription(filter=…)` | Receives the copy only when its filter matches | | Per-subscription pull | `Pull` / `StreamingPull` | Independent backlog + ack state per subscription | Fan-out semantics: * **One server-assigned message id per publish**, the same across every subscription's copy. * **Zero subscriptions → the publish still succeeds** and the message lands only in the topic log (no error). * **Filters apply at fan-out time** — a non-matching subscription is simply never enqueued (≈ auto-acked), and a publish that matches **no** subscription still succeeds. ## How it works [#how-it-works] A single publish is written once to the topic log `gcp.{topic}`, then the connector fans out one Queue copy to every subscription bound to the topic. Each subscription has its own queue, filter, and ack state; a filtered subscription receives the copy only when its filter matches. *One publish writes once to the topic log `gcp.{topic}` and fans out an independent Queue copy to every subscription on `gcp.sub.{subscription}`, all sharing one message id; a filtered subscription receives its copy only when the message attributes match its filter.* ## Fan one publish to many subscriptions [#fan-one-publish-to-many-subscriptions] Create a topic, attach two subscriptions, publish once, and watch both subscriptions receive the same message id from their own independent queues. Each client sets only `PUBSUB_EMULATOR_HOST` (default `localhost:8085`) and a project id. ```go package main import ( "context" "fmt" "log" "os" "time" "cloud.google.com/go/pubsub" "cloud.google.com/go/pubsub/apiv1/pubsubpb" ) func projectID() string { if v := os.Getenv("PUBSUB_PROJECT_ID"); v != "" { return v } return "my-project" } func main() { ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() // PUBSUB_EMULATOR_HOST routes the official client at the connector (insecure gRPC). client, err := pubsub.NewClient(ctx, projectID()) if err != nil { log.Fatalf("NewClient: %v", err) } defer client.Close() // One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping. topic, err := client.CreateTopic(ctx, "events") if err != nil { log.Fatalf("CreateTopic: %v", err) } defer topic.Stop() for _, name := range []string{"billing", "shipping"} { if _, err := client.CreateSubscription(ctx, name, pubsub.SubscriptionConfig{ Topic: topic, AckDeadline: 10 * time.Second, }); err != nil { log.Fatalf("CreateSubscription %q: %v", name, err) } } // One Publish fans out to both subscriptions with a shared message id. msgID, err := topic.Publish(ctx, &pubsub.Message{Data: []byte("order #1001 placed")}).Get(ctx) if err != nil { log.Fatalf("Publish: %v", err) } fmt.Printf("published: %s\n", msgID) // Each subscription has its own backlog — pull from both. subClient, err := pubsub.NewSubscriberClient(ctx) if err != nil { log.Fatalf("NewSubscriberClient: %v", err) } defer subClient.Close() for _, name := range []string{"billing", "shipping"} { subPath := fmt.Sprintf("projects/%s/subscriptions/%s", projectID(), name) resp, err := subClient.Pull(ctx, &pubsubpb.PullRequest{Subscription: subPath, MaxMessages: 1}) if err != nil { log.Fatalf("Pull %q: %v", name, err) } rm := resp.GetReceivedMessages()[0] fmt.Printf("%s received message id=%s\n", name, rm.GetMessage().GetMessageId()) _ = subClient.Acknowledge(ctx, &pubsubpb.AcknowledgeRequest{ Subscription: subPath, AckIds: []string{rm.GetAckId()}, }) } } ``` ```python import os from google.cloud import pubsub_v1 def project_id() -> str: return os.environ.get("PUBSUB_PROJECT_ID", "my-project") def main() -> None: publisher = pubsub_v1.PublisherClient() subscriber = pubsub_v1.SubscriberClient() proj = project_id() topic_path = publisher.topic_path(proj, "events") # -> gcp.events # One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping. publisher.create_topic(request={"name": topic_path}) sub_paths = {} for name in ("billing", "shipping"): sub_path = subscriber.subscription_path(proj, name) subscriber.create_subscription(request={"name": sub_path, "topic": topic_path}) sub_paths[name] = sub_path # One Publish fans out to both subscriptions with a shared message id. msg_id = publisher.publish(topic_path, b"order #1001 placed").result(timeout=15) print(f"published: {msg_id}") # Each subscription has its own backlog. for name, sub_path in sub_paths.items(): resp = subscriber.pull(request={"subscription": sub_path, "max_messages": 1}, timeout=20) rm = resp.received_messages[0] assert rm.message.message_id == msg_id # same message id across subscriptions print(f"{name} received message id={rm.message.message_id}") subscriber.acknowledge(request={"subscription": sub_path, "ack_ids": [rm.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; import java.util.List; 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"); ManagedChannel channel = ManagedChannelBuilder.forTarget(emulatorHost).usePlaintext().build(); TransportChannelProvider channelProvider = FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel)); NoCredentialsProvider noCreds = NoCredentialsProvider.create(); TopicName topic = TopicName.of(projectId, "events"); // -> gcp.events 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())) { // One topic, two independent subscriptions. topicAdmin.createTopic(topic); for (String name : List.of("billing", "shipping")) { subAdmin.createSubscription(SubscriptionName.of(projectId, name), topic, PushConfig.getDefaultInstance(), 10); } // One Publish fans out to both subscriptions with a shared message id. String msgId = topicAdmin.publish(PublishRequest.newBuilder().setTopic(topic.toString()) .addMessages(PubsubMessage.newBuilder() .setData(ByteString.copyFromUtf8("order #1001 placed")).build()) .build()).getMessageIds(0); System.out.printf("published: %s%n", msgId); // Each subscription has its own backlog. for (String name : List.of("billing", "shipping")) { String subPath = SubscriptionName.of(projectId, name).toString(); PullResponse resp = subStub.pullCallable().call(PullRequest.newBuilder() .setSubscription(subPath).setMaxMessages(1).build()); ReceivedMessage got = resp.getReceivedMessages(0); System.out.printf("%s received message id=%s%n", name, got.getMessage().getMessageId()); subStub.acknowledgeCallable().call(AcknowledgeRequest.newBuilder() .setSubscription(subPath).addAckIds(got.getAckId()).build()); } } finally { channel.shutdown(); } } } ``` ```typescript import { PubSub, v1 } from "@google-cloud/pubsub"; const projectId = process.env["PUBSUB_PROJECT_ID"] ?? "my-project"; 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, "events"); // -> gcp.events // One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping. await publisher.createTopic({ name: topic }); const subNames = ["billing", "shipping"].map((n) => subscriber.subscriptionPath(projectId, n)); for (const sub of subNames) { await subscriber.createSubscription({ name: sub, topic, ackDeadlineSeconds: 10 }); } // One Publish fans out to both subscriptions with a shared message id. const [published] = await publisher.publish({ topic, messages: [{ data: Buffer.from("order #1001 placed") }], }); const msgId = published.messageIds?.[0] ?? ""; console.log(`published: ${msgId}`); // Each subscription has its own backlog. for (const sub of subNames) { const [pull] = await subscriber.pull({ subscription: sub, maxMessages: 1 }); const rm = pull.receivedMessages![0]; console.log(`${sub.split("/").pop()} received message id=${rm.message!.messageId}`); await subscriber.acknowledge({ subscription: sub, ackIds: [rm.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, "events"); // -> gcp.events // The .NET client does NOT auto-detect the emulator — set EmulatorOnly. var publisher = await new PublisherServiceApiClientBuilder { EmulatorDetection = EmulatorDetection.EmulatorOnly, }.BuildAsync(); var subscriber = await new SubscriberServiceApiClientBuilder { EmulatorDetection = EmulatorDetection.EmulatorOnly, }.BuildAsync(); // One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping. await publisher.CreateTopicAsync(topicName); var subNames = new[] { "billing", "shipping" } .Select(n => SubscriptionName.FromProjectSubscription(projectId, n)).ToArray(); foreach (var sub in subNames) { await subscriber.CreateSubscriptionAsync(sub, topicName, pushConfig: null, ackDeadlineSeconds: 10); } // One Publish fans out to both subscriptions with a shared message id. var publishResponse = await publisher.PublishAsync(topicName, new[] { new PubsubMessage { Data = ByteString.CopyFromUtf8("order #1001 placed") }, }); var msgId = publishResponse.MessageIds[0]; Console.WriteLine($"published: {msgId}"); // Each subscription has its own backlog. foreach (var sub in subNames) { var pull = await subscriber.PullAsync(sub, maxMessages: 1); var rm = pull.ReceivedMessages[0]; Console.WriteLine($"{sub.SubscriptionId} received message id={rm.Message.MessageId}"); await subscriber.AcknowledgeAsync(sub, new[] { rm.AckId }); } ``` ```ruby # frozen_string_literal: true require "google/cloud/pubsub" project_id = ENV["PUBSUB_PROJECT_ID"] || "my-project" 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("events") # -> gcp.events # One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping. topic = topic_admin.create_topic(name: topic_path) sub_paths = %w[billing shipping].to_h do |name| sub_path = pubsub.subscription_path(name) sub_admin.create_subscription(name: sub_path, topic: topic_path, ack_deadline_seconds: 10) [name, sub_path] end # One Publish fans out to both subscriptions with a shared message id. publisher = pubsub.publisher(topic.name) msg_id = publisher.publish("order #1001 placed").message_id puts "published: #{msg_id}" # Each subscription has its own backlog. sub_paths.each do |name, sub_path| subscriber = pubsub.subscriber(sub_path) rcv = subscriber.pull(immediate: false, max: 1).first puts "#{name} received message id=#{rcv.message_id}" rcv.acknowledge! end ``` ## Filtered fan-out [#filtered-fan-out] Set a `filter` on a subscription so it receives only the publishes whose **attributes** match. Filters are an attributes-only CEL subset (`attributes:KEY`, `=` / `!=`, `hasPrefix`, `AND` / `OR` / `NOT`), ≤ 256 characters, compiled once at create-time, and applied at fan-out — a non-matching publish is never enqueued for that subscription, and a publish that matches no subscription still succeeds. ```python # A subscription that receives only "order" events. subscriber.create_subscription( request={ "name": subscriber.subscription_path(proj, "orders-only"), "topic": topic_path, "filter": 'attributes.eventType = "order"', } ) # A matching publish is delivered; a non-matching one is suppressed for this subscription. publisher.publish(topic_path, b"an order event", eventType="order") publisher.publish(topic_path, b"a metric event", eventType="metric") # suppressed for orders-only ``` **The filter is immutable and attributes-only.** It is compiled at `CreateSubscription` and cannot be changed afterward (`UpdateSubscription` rejects a `filter` change); a malformed filter is rejected at create-time with `INVALID_ARGUMENT`. Put the values you filter on into message **attributes**, not the body. See [Message filtering](/connectors/gcp-pub-sub/how-to/filtering). ## Independent backlogs and ack state [#independent-backlogs-and-ack-state] Each subscription is a separate Queue channel `gcp.sub.{subscription}` with its own backlog, ack-deadline leases, dead-letter policy, and retention. A consumer that lags, nacks, or dead-letters on one subscription has **no effect** on any other subscription bound to the same topic — the topic log `gcp.{topic}` is the single shared, replayable source, and each subscription replays from it independently via [Seek](/connectors/gcp-pub-sub/how-to/seek-and-snapshots). ## Related [#related]