Google Cloud Pub/Sub

The Google Cloud client libraries over gRPC with TLS, with no credentials (AnonymousCredentials) and your identifier as the project.

Endpoint and credentials

Topics and subscriptions without setup

A pull names only a subscription, never its topic, so the public service uses a naming convention to create subscriptions on demand:

• Publishing to an unknown topic X creates topic X and a subscription also named X bound to it. Publish to events, pull from events, and nothing published before your first pull is lost.
• Pulling (or streaming-pulling) a subscription named X~name, such as events~audit, creates that subscription on topic X, creating topic X and its subscription X too if needed. It receives what is published from then on, so start it before publishing.
• Pulling any other unknown subscription answers NOT_FOUND.
• CreateTopic and CreateSubscription work as usual, for any names. Creating a topic or subscription that already came into being implicitly answers ALREADY_EXISTS, as for any existing one.

Topics and subscriptions created implicitly are deleted after an hour empty and unused.

Python (google-cloud-pubsub)

pip install google-cloud-pubsub. Publish, then receive with a streaming pull:

from concurrent.futures import TimeoutError

from google.api_core.client_options import ClientOptions
from google.auth.credentials import AnonymousCredentials
from google.cloud import pubsub_v1

PROJECT = "<your-identifier>"
options = ClientOptions(api_endpoint="pubsub.powerqueue.link:443")
publisher = pubsub_v1.PublisherClient(credentials=AnonymousCredentials(), client_options=options)
subscriber = pubsub_v1.SubscriberClient(credentials=AnonymousCredentials(), client_options=options)

topic = publisher.topic_path(PROJECT, "events")
for i in range(3):
    publisher.publish(topic, f"order {i}".encode(), origin="example").result()


def callback(message):
    print(message.data, message.attributes)
    message.ack()


streaming = subscriber.subscribe(subscriber.subscription_path(PROJECT, "events"), callback=callback)
with subscriber:
    try:
        streaming.result(timeout=10)
    except TimeoutError:
        streaming.cancel()
        streaming.result()

Creating entities explicitly works too, with any subscription settings within the limits. With clients made as above:

publisher.create_topic(name=publisher.topic_path(PROJECT, "reports"))
subscriber.create_subscription(
    name=subscriber.subscription_path(PROJECT, "reports-worker"),
    topic=publisher.topic_path(PROJECT, "reports"),
    ack_deadline_seconds=30,
)

Node.js (@google-cloud/pubsub)

npm install @google-cloud/pubsub @grpc/grpc-js. An endpoint outside googleapis.com makes the library assume an emulator, so set emulatorMode: false. TLS channel credentials in sslCreds stop it from looking for Google credentials, and naming the universeDomain stops it from probing for Google Cloud's metadata server:

import { PubSub } from "@google-cloud/pubsub";
import { credentials } from "@grpc/grpc-js";

const pubsub = new PubSub({
  projectId: "<your-identifier>",
  apiEndpoint: "pubsub.powerqueue.link:443",
  emulatorMode: false,
  universeDomain: "googleapis.com",
  sslCreds: credentials.createSsl(),
});

await pubsub.topic("events").publishMessage({ data: Buffer.from("hello") });

const subscription = pubsub.subscription("events");
subscription.on("message", (message) => {
  console.log(message.data.toString());
  message.ack();
});
setTimeout(() => subscription.close(), 10000);

Go (cloud.google.com/go/pubsub/v2)

go get cloud.google.com/go/pubsub/v2. option.WithoutAuthentication sends no credentials and keeps TLS:

package main

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

	"cloud.google.com/go/pubsub/v2"
	"google.golang.org/api/option"
)

func main() {
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, "<your-identifier>",
		option.WithEndpoint("pubsub.powerqueue.link:443"),
		option.WithoutAuthentication(),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	publisher := client.Publisher("events")
	if _, err := publisher.Publish(ctx, &pubsub.Message{Data: []byte("hello")}).Get(ctx); err != nil {
		log.Fatal(err)
	}
	publisher.Stop()

	receiveCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	err = client.Subscriber("events").Receive(receiveCtx, func(_ context.Context, message *pubsub.Message) {
		fmt.Println(string(message.Data))
		message.Ack()
		cancel()
	})
	if err != nil {
		log.Fatal(err)
	}
}

Supported

• Publisher: CreateTopic, UpdateTopic, Publish, GetTopic, ListTopics, ListTopicSubscriptions, ListTopicSnapshots, DeleteTopic, DetachSubscription.
• Subscriber: CreateSubscription, GetSubscription, UpdateSubscription, ListSubscriptions, DeleteSubscription, Pull, StreamingPull, Acknowledge, ModifyAckDeadline, snapshots (CreateSnapshot, GetSnapshot, ListSnapshots, UpdateSnapshot, DeleteSnapshot) and Seek.
• Schemas (Avro and Protocol Buffers, JSON or binary encoding, with revisions), and validation of messages published to a topic with a schema.
• Ordering keys, exactly-once delivery, subscription filters, dead-letter policies and retry policies.
• IAM methods (SetIamPolicy, GetIamPolicy, TestIamPermissions): policies are stored and returned, but grant and deny nothing.

Differences from Google Cloud Pub/Sub

• gRPC only: the REST (HTTP/JSON) API is not served, so use each library's default gRPC transport.
• Push subscriptions are refused with INVALID_ARGUMENT ("Push subscriptions are not available on this public service; use pull."), in CreateSubscription, ModifyPushConfig and UpdateSubscription.
• BigQuery, Cloud Storage and Bigtable subscriptions, topic ingestion sources and message transforms answer UNIMPLEMENTED.
• Message retention defaults to 1 hour, the most allowed, rather than 7 days; longer retention on a topic or subscription is refused.
• Publishing to an unknown topic and pulling an X~name subscription create them instead of answering NOT_FOUND (see above).
• Every resource a request names must be in the request's own project: a subscription on another project's topic, or a dead-letter topic, snapshot or schema in another project, is INVALID_ARGUMENT.

Messages from other interfaces

Bodies sent through the other interfaces arrive as their bytes; a typed Service Bus body arrives as JSON with the attribute x-pq-body-encoding = json. Attributes are strings: numbers arrive as their decimal text, booleans as true or false, AMQP tables and arrays as JSON, and binary values (an SQS Binary attribute, an AMQP byte-array header) base64-encoded, their names listed, comma-separated, in x-pq-base64-attributes. Attribute names starting x-pq- cannot be published.

The full rules, with examples: Using one queue from several interfaces.

Errors

Each client address may have at most 100 calls in progress, over all its connections; each HTTP/2 connection carries at most 100 concurrent streams, and the libraries queue further calls until a stream is free.