Publish & Subscribe

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)
	}
}

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

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();
        }
    }
}

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<void> {
  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);
});

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 });

# 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!