# Cross-Protocol Interop (/connectors/gcp-pub-sub/concepts/cross-protocol-interop)



Because every Pub/Sub **topic** is a normal KubeMQ **Events Store** log (`gcp.{topic}`), a Google Pub/Sub application and a native KubeMQ gRPC/REST client can work the **same** stream. A message published by `google-cloud-pubsub` to topic `orders` is consumable by a native KubeMQ Events Store subscriber on channel `gcp.orders` — carrying the connector's reserved `_pubsub_*` tags across the wire. This lets you **bridge a legacy native consumer during a migration**, or run Pub/Sub producers alongside native KubeMQ consumers without changing either side's protocol.

## Overview [#overview]

A `Publish` writes the message **once** to the topic's Events Store log `gcp.{topic}` — the authoritative, replayable source — before fanning out per-subscription queue copies. The native side reads that topic log directly: there is no subscription on the native path, just an Events Store subscribe on `gcp.{topic}`. The Pub/Sub side speaks the v1 gRPC wire protocol to the connector (port 8085); the native side speaks gRPC/REST to the KubeMQ broker directly (default `localhost:50000`).

| Direction        | Producer                            | Consumer                                      | What carries over                                |
| ---------------- | ----------------------------------- | --------------------------------------------- | ------------------------------------------------ |
| Pub/Sub → native | Pub/Sub `Publish` on topic `orders` | native `SubscribeToEventsStore("gcp.orders")` | Body + your attributes + the three reserved tags |

The three reserved tags are **visible to native consumers** (and stripped from `attributes` for Pub/Sub clients):

* `_pubsub_message_id` — the server-assigned message id (matches the `Publish` return value)
* `_pubsub_publish_time` — the publish timestamp
* `_pubsub_ordering_key` — the ordering key, if any

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

The Pub/Sub publish writes once to the Events Store log `gcp.{topic}` through the message broker. A native KubeMQ Events Store subscriber attached to the same `gcp.{topic}` channel reads exactly that message, including the reserved `_pubsub_*` tags the connector stamps on the wire.

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

PUB -- &#x22;Publish&#x22; --> CONN
CONN -- &#x22;SendEventsStore&#x22; --> LOG
LOG --> BROKER
NATIVE -- &#x22;SubscribeToEventsStore (gRPC)&#x22; --> BROKER
BROKER -. &#x22;deliver body + _pubsub_* tags&#x22; .-> NATIVE

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

*A Pub/Sub publish writes once to the Events Store log `gcp.{topic}`; a native KubeMQ Events Store subscriber on the same channel reads that message with the reserved `_pubsub_*` tags carried across the wire.*

## The Pub/Sub publish side [#the-pubsub-publish-side]

The publish half is ordinary Pub/Sub code — `CreateTopic` then `Publish` against topic `orders` (which maps to `gcp.orders`). Each client sets only `PUBSUB_EMULATOR_HOST` (default `localhost:8085`) and a project id. To make the `_pubsub_ordering_key` tag observable on the native side, the publisher enables ordering and supplies an `ordering_key`.

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"fmt"
    	"log"
    	"os"
    	"time"

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

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    	defer cancel()
    	projectID := os.Getenv("PUBSUB_PROJECT_ID")
    	if projectID == "" {
    		projectID = "my-project"
    	}

    	// 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()

    	// Topic "orders" maps to the Events Store log "gcp.orders".
    	topic, err := client.CreateTopic(ctx, "orders")
    	if err != nil {
    		log.Fatalf("CreateTopic: %v", err)
    	}
    	defer topic.Stop()
    	topic.EnableMessageOrdering = true // makes _pubsub_ordering_key observable natively.

    	// Publish one message; a native consumer on gcp.orders reads it.
    	id, err := topic.Publish(ctx, &pubsub.Message{
    		Data:        []byte("from-gcp-pubsub"),
    		OrderingKey: "shipments",
    		Attributes:  map[string]string{"region": "emea"}, // rides along as a plain tag.
    	}).Get(ctx)
    	if err != nil {
    		log.Fatalf("Publish: %v", err)
    	}
    	fmt.Printf("published: %s (native channel: gcp.orders)\n", id)
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os

    from google.cloud import pubsub_v1
    from google.cloud.pubsub_v1.types import PublisherOptions


    def main() -> None:
        project_id = os.environ.get("PUBSUB_PROJECT_ID", "my-project")
        # Ordering must be enabled to publish with an ordering key.
        publisher = pubsub_v1.PublisherClient(
            publisher_options=PublisherOptions(enable_message_ordering=True)
        )
        topic_path = publisher.topic_path(project_id, "orders")  # -> gcp.orders
        publisher.create_topic(request={"name": topic_path})

        # Publish one message; a native consumer on gcp.orders reads it.
        future = publisher.publish(
            topic_path,
            b"from-gcp-pubsub",
            ordering_key="shipments",
            region="emea",  # an ordinary attribute — rides along as a plain tag.
        )
        print(f"published: {future.result(timeout=15)} (native channel: gcp.orders)")


    if __name__ == "__main__":
        main()
    ```
  </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 com.google.cloud.pubsub.v1.TopicAdminClient;
    import com.google.cloud.pubsub.v1.TopicAdminSettings;
    import com.google.protobuf.ByteString;
    import com.google.pubsub.v1.PublishRequest;
    import com.google.pubsub.v1.PubsubMessage;
    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");

            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

            try (TopicAdminClient topicAdmin = TopicAdminClient.create(TopicAdminSettings.newBuilder()
                    .setTransportChannelProvider(channelProvider).setCredentialsProvider(noCreds).build())) {

                topicAdmin.createTopic(topic);

                // Publish one message; a native consumer on gcp.orders reads it.
                String id = topicAdmin.publish(PublishRequest.newBuilder()
                        .setTopic(topic.toString())
                        .addMessages(PubsubMessage.newBuilder()
                                .setData(ByteString.copyFromUtf8("from-gcp-pubsub"))
                                .setOrderingKey("shipments")
                                .putAttributes("region", "emea") // rides along as a plain tag.
                                .build())
                        .build()).getMessageIds(0);
                System.out.printf("published: %s (native channel: gcp.orders)%n", id);
            } finally {
                channel.shutdown();
            }
        }
    }
    ```
  </Tab>

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

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

    async function main(): Promise<void> {
      // The high-level client auto-detects PUBSUB_EMULATOR_HOST (insecure gRPC).
      const pubsub = new PubSub({ projectId });

      // Topic "orders" maps to the Events Store log "gcp.orders".
      const [topic] = await pubsub.createTopic("orders");
      topic.setPublishOptions({ messageOrdering: true }); // makes _pubsub_ordering_key observable.

      // Publish one message; a native consumer on gcp.orders reads it.
      const messageId = await topic.publishMessage({
        data: Buffer.from("from-gcp-pubsub"),
        orderingKey: "shipments",
        attributes: { region: "emea" }, // rides along as a plain tag.
      });
      console.log(`published: ${messageId} (native channel: gcp.orders)`);
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```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

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

    await publisher.CreateTopicAsync(topicName);

    // Publish one message; a native consumer on gcp.orders reads it.
    var publishResponse = await publisher.PublishAsync(topicName, new[]
    {
        new PubsubMessage
        {
            Data = ByteString.CopyFromUtf8("from-gcp-pubsub"),
            OrderingKey = "shipments",
            Attributes = { ["region"] = "emea" }, // rides along as a plain tag.
        },
    });
    Console.WriteLine($"published: {publishResponse.MessageIds[0]} (native channel: gcp.orders)");
    ```
  </Tab>

  <Tab value="Ruby">
    ```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
    topic_path  = pubsub.topic_path("orders") # -> gcp.orders

    topic = topic_admin.create_topic(name: topic_path)

    # ordered: true makes _pubsub_ordering_key observable on the native side.
    publisher = pubsub.publisher(topic.name, async: { ordered: true })
    msg = publisher.publish("from-gcp-pubsub", ordering_key: "shipments", region: "emea")
    publisher.async_publisher.stop!
    puts "published: #{msg.message_id} (native channel: gcp.orders)"
    ```
  </Tab>
</Tabs>

## The native consumer [#the-native-consumer]

The other side is an ordinary KubeMQ **Events Store** subscriber talking gRPC to the broker (default `localhost:50000`) on the `gcp.orders&#x60; channel — no Pub/Sub SDK involved. Subscribe with the &#x2A;*"new only"** start position and confirm the stream is open **before** publishing, so the published message is in-window (Events Store subscribers attach to a stream, not a fixed offset). The message the Pub/Sub side published carries its body, your attributes, and the three reserved `_pubsub_*` tags.

<Tabs groupId="language" items="['Go','Python','JavaScript']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"fmt"
    	"log"
    	"os"
    	"time"

    	kubemq "github.com/kubemq-io/kubemq-go/v2"
    )

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    	defer cancel()
    	grpcAddress := os.Getenv("KUBEMQ_GRPC_ADDRESS")
    	if grpcAddress == "" {
    		grpcAddress = "localhost:50000"
    	}

    	// Native KubeMQ gRPC client on the shared Events Store channel gcp.orders.
    	native, err := kubemq.NewClient(ctx,
    		kubemq.WithAddress("localhost", 50000),
    		kubemq.WithClientId("gcp-interop-native-go"),
    	)
    	if err != nil {
    		log.Fatalf("connect native gRPC: %v", err)
    	}
    	defer native.Close()

    	received := make(chan *kubemq.EventStoreReceive, 1)

    	// Subscribe FIRST with start policy "new only"; the stream is open on return.
    	sub, err := native.SubscribeToEventsStore(ctx, "gcp.orders", "", kubemq.StartFromNewEvents(),
    		kubemq.WithOnEventStoreReceive(func(ev *kubemq.EventStoreReceive) { received <- ev }),
    		kubemq.WithOnError(func(e error) { log.Printf("subscribe error: %v", e) }),
    	)
    	if err != nil {
    		log.Fatalf("SubscribeToEventsStore: %v", err)
    	}
    	defer sub.Cancel()
    	fmt.Println("native SubscribeToEventsStore(gcp.orders, startAt=new) -> stream open")

    	// (Run the Pub/Sub publish side now; it lands on gcp.orders.)
    	select {
    	case ev := <-received:
    		fmt.Printf("native received %q\n", string(ev.Body))
    		fmt.Printf("  _pubsub_message_id=%s\n", ev.Tags["_pubsub_message_id"])
    		fmt.Printf("  _pubsub_ordering_key=%s\n", ev.Tags["_pubsub_ordering_key"])
    		fmt.Printf("  region (attribute)=%s\n", ev.Tags["region"])
    	case <-time.After(15 * time.Second):
    		log.Fatal("timed out waiting for the native event")
    	}
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os
    import queue

    from kubemq import EventsStoreSubscription, EventStoreReceived, PubSubClient
    from kubemq.pubsub.events_store_subscription import EventStoreStartPosition

    GRPC_ADDRESS = os.environ.get("KUBEMQ_GRPC_ADDRESS", "localhost:50000")
    CHANNEL = "gcp.orders"


    def main() -> None:
        received: queue.Queue[EventStoreReceived] = queue.Queue(maxsize=1)
        native = PubSubClient(address=GRPC_ADDRESS, client_id="gcp-interop-native-python")

        # Subscribe FIRST with start policy "new only".
        native.subscribe_to_events_store(
            EventsStoreSubscription(
                channel=CHANNEL,
                events_store_type=EventStoreStartPosition.StartFromNew,
                on_receive_event_callback=received.put,
                on_error_callback=lambda err: print(f"subscribe error: {err}"),
            )
        )
        print(f"native SubscribeToEventsStore({CHANNEL}, startAt=new) -> stream open")

        # (Run the Pub/Sub publish side now; it lands on gcp.orders.)
        event = received.get(timeout=15)
        print(f"native received {event.body.decode()!r}")
        print(f"  _pubsub_message_id={event.tags.get('_pubsub_message_id')}")
        print(f"  _pubsub_ordering_key={event.tags.get('_pubsub_ordering_key')}")
        print(f"  region (attribute)={event.tags.get('region')}")
        native.close()


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import { KubeMQClient, EventStoreStartPosition } from "kubemq-js";
    import type { EventStoreReceived, KubeMQError } from "kubemq-js";

    const grpcAddress = process.env["KUBEMQ_GRPC_ADDRESS"] ?? "localhost:50000";
    const CHANNEL = "gcp.orders";

    async function main(): Promise<void> {
      // KubeMQClient.create() connects before returning a ready client.
      const kube = await KubeMQClient.create({ address: grpcAddress, clientId: "gcp-interop-native-js" });

      const received = new Promise<EventStoreReceived>((resolve, reject) => {
        // Subscribe FIRST with start position "new only".
        kube.subscribeToEventsStore({
          channel: CHANNEL,
          startFrom: EventStoreStartPosition.StartFromNew,
          onEvent: (event: EventStoreReceived) => resolve(event),
          onError: (err: KubeMQError) => reject(err),
        });
      });
      console.log(`native SubscribeToEventsStore(${CHANNEL}, startAt=new) -> stream open`);

      // (Run the Pub/Sub publish side now; it lands on gcp.orders.)
      const event = await received;
      console.log(`native received ${JSON.stringify(Buffer.from(event.body).toString("utf8"))}`);
      console.log(`  _pubsub_message_id=${event.tags["_pubsub_message_id"]}`);
      console.log(`  _pubsub_ordering_key=${event.tags["_pubsub_ordering_key"]}`);
      console.log(`  region (attribute)=${event.tags["region"]}`);
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>
</Tabs>

<Callout type="info">
  The native consumer is the **only** place a KubeMQ SDK appears in the Pub/Sub connector examples — the Pub/Sub publish half above is idiomatic Google client code in every language. The native half is shown in Go, Python, and JavaScript; where a language's native Events Store client is less mature, it can fall back to a `kubemq-go` sidecar or a REST Events Store call.
</Callout>

## Subscribe before publish [#subscribe-before-publish]

<Callout type="warn">
  **Establish the native subscribe stream before publishing.** An Events Store subscriber with the "new only" start position receives only events published **after** the stream is open. A naive "publish then subscribe" races — the publish can land before the subscriber attaches and be missed. Open the native subscription first, confirm the stream is up, then run the Pub/Sub publish so the message is in-window.
</Callout>

## Topic log, not subscription queue [#topic-log-not-subscription-queue]

The native path reads the **topic log** `gcp.{topic}` directly — the authoritative, replayable source written once per publish. It does not read a subscription's Queue channel `gcp.sub.{subscription}`; those are the per-subscription fan-out copies consumed by Pub/Sub `Pull`. For point-to-point native consumption of a subscription's backlog instead, subscribe to its `gcp.sub.{subscription}` Queue channel. See the [channel mapping reference](/connectors/gcp-pub-sub/reference/channel-mapping) for the full grammar.

## Related [#related]

<Cards>
  <Card title="Architecture" href="/connectors/gcp-pub-sub/concepts/architecture" description="The publish-once-then-fan-out model and how topics and subscriptions map to KubeMQ primitives." />

  <Card title="Channel mapping" href="/connectors/gcp-pub-sub/reference/channel-mapping" description="The gcp.{topic} log, gcp.sub.{s} queue grammar, and the reserved _pubsub_* tags." />

  <Card title="Migrating from Google Cloud Pub/Sub" href="/connectors/gcp-pub-sub/reference/migration-from-gcp" description="Point an existing Pub/Sub app at KubeMQ and bridge native consumers during a migration." />
</Cards>