Getting Started

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

from google.cloud import pubsub_v1

# Both clients honour PUBSUB_EMULATOR_HOST automatically.
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()

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

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

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

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

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)

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

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

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

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

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

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

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

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

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

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

subscriber = pubsub.subscriber(sub_path)
rcv = subscriber.pull(immediate: false, max: 1).first
puts "received: #{rcv.data.inspect}"

rcv.acknowledge!