Pub/Sub

floci-gcp emulates Google Cloud Pub/Sub over gRPC using the real google.pubsub.v1 protocol. It also exposes the Pub/Sub REST v1 JSON surface used by tools such as Terraform.

Configuration

Variable	Default	Description
FLOCI_GCP_SERVICES_PUBSUB_ENABLED	true	Enable/disable Pub/Sub

Emulator Variable

export PUBSUB_EMULATOR_HOST=localhost:4588

GCP Pub/Sub SDK clients use this variable to route requests to floci-gcp instead of pubsub.googleapis.com.

REST / Terraform Endpoint

Terraform's Google provider can use the REST endpoint with:

provider "google" {
  pubsub_custom_endpoint = "http://localhost:4588/v1/"
}

The REST surface supports topic and subscription create/read/list/update/delete, publish, pull, and acknowledge:

• PUT /v1/projects/{project}/topics/{topic}
• GET /v1/projects/{project}/topics/{topic}
• GET /v1/projects/{project}/topics
• PATCH /v1/projects/{project}/topics/{topic}?updateMask=...
• DELETE /v1/projects/{project}/topics/{topic}
• POST /v1/projects/{project}/topics/{topic}:publish
• PUT /v1/projects/{project}/subscriptions/{subscription}
• GET /v1/projects/{project}/subscriptions/{subscription}
• GET /v1/projects/{project}/subscriptions
• PATCH /v1/projects/{project}/subscriptions/{subscription}?updateMask=...
• DELETE /v1/projects/{project}/subscriptions/{subscription}
• POST /v1/projects/{project}/subscriptions/{subscription}:pull
• POST /v1/projects/{project}/subscriptions/{subscription}:acknowledge

Quick Start

gcloud CLIJavaPython

export PUBSUB_EMULATOR_HOST=localhost:4588
gcloud config set project floci-local

# Create topic and subscription
gcloud pubsub topics create my-topic
gcloud pubsub subscriptions create my-sub --topic=my-topic

# Publish a message
gcloud pubsub topics publish my-topic --message="hello from floci-gcp"

# Pull messages
gcloud pubsub subscriptions pull my-sub --auto-ack --limit=10

ManagedChannel channel = ManagedChannelBuilder
    .forTarget("localhost:4588")
    .usePlaintext()
    .build();

TransportChannelProvider channelProvider =
    FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel));
CredentialsProvider credentialsProvider = NoCredentialsProvider.create();

// Create topic
TopicAdminClient topicAdminClient = TopicAdminClient.create(
    TopicAdminSettings.newBuilder()
        .setTransportChannelProvider(channelProvider)
        .setCredentialsProvider(credentialsProvider)
        .build());

topicAdminClient.createTopic(TopicName.of("floci-local", "my-topic"));

// Create subscription
SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create(
    SubscriptionAdminSettings.newBuilder()
        .setTransportChannelProvider(channelProvider)
        .setCredentialsProvider(credentialsProvider)
        .build());

subscriptionAdminClient.createSubscription(
    SubscriptionName.of("floci-local", "my-sub"),
    TopicName.of("floci-local", "my-topic"),
    PushConfig.getDefaultInstance(),
    10);

// Publish
Publisher publisher = Publisher.newBuilder(TopicName.of("floci-local", "my-topic"))
    .setChannelProvider(channelProvider)
    .setCredentialsProvider(credentialsProvider)
    .build();

PubsubMessage message = PubsubMessage.newBuilder()
    .setData(ByteString.copyFromUtf8("hello from floci-gcp"))
    .build();

publisher.publish(message).get();

// Pull
SubscriberStubSettings subscriberSettings = SubscriberStubSettings.newBuilder()
    .setTransportChannelProvider(channelProvider)
    .setCredentialsProvider(credentialsProvider)
    .build();

try (SubscriberStub subscriber = GrpcSubscriberStub.create(subscriberSettings)) {
    PullRequest pullRequest = PullRequest.newBuilder()
        .setMaxMessages(10)
        .setSubscription(SubscriptionName.of("floci-local", "my-sub").toString())
        .build();

    PullResponse response = subscriber.pullCallable().call(pullRequest);
    response.getReceivedMessagesList().forEach(msg ->
        System.out.println(msg.getMessage().getData().toStringUtf8()));
}

import os
os.environ["PUBSUB_EMULATOR_HOST"] = "localhost:4588"

from google.cloud import pubsub_v1

project_id = "floci-local"

# Create topic
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, "my-topic")
publisher.create_topic(request={"name": topic_path})

# Create subscription
subscriber = pubsub_v1.SubscriberClient()
sub_path = subscriber.subscription_path(project_id, "my-sub")
subscriber.create_subscription(request={
    "name": sub_path,
    "topic": topic_path,
})

# Publish
future = publisher.publish(topic_path, b"hello from floci-gcp")
future.result()

# Pull
response = subscriber.pull(request={"subscription": sub_path, "max_messages": 10})
for msg in response.received_messages:
    print(msg.message.data.decode())

Subscription Filters

A subscription can declare a filter so it only receives messages whose attributes match it. Non-matching messages are never delivered to that subscription — as in GCP, which acknowledges them automatically on your behalf. Subscriptions without a filter receive every message published to the topic.

subscriptionAdminClient.createSubscription(Subscription.newBuilder()
    .setName(SubscriptionName.of("floci-local", "invoices").toString())
    .setTopic(TopicName.of("floci-local", "events").toString())
    .setAckDeadlineSeconds(10)
    .setFilter("attributes.event_type = \"ocr-invoice\"")
    .build());

subscriber.create_subscription(request={
    "name": sub_path,
    "topic": topic_path,
    "filter": 'attributes.event_type = "ocr-invoice"',
})

Supported syntax

Form	Example	Matches
Attribute exists	attributes:name	messages that carry a name attribute
Quoted key	attributes:"iana.org/language_tag"	keys with characters other than hyphens, underscores or alphanumerics
Equality	attributes.name = "com"	name is exactly com
Inequality	attributes.name != "com"	name differs from com, including when the attribute is absent
Prefix	hasPrefix(attributes.name, "co")	name starts with co
Conjunction	attributes:a AND attributes.b = "1"	both operands match
Disjunction	attributes.a = "1" OR attributes.a = "2"	either operand matches
Negation	NOT attributes:a / -attributes:a	operand does not match

Rules that mirror GCP:

• Keys and values are case-sensitive; AND, OR and NOT must be uppercase.
• NOT has the highest precedence; - is a unary alias for it.
• AND and OR cannot be combined without parentheses. a AND b OR c is a syntax error; write a AND (b OR c).
• hasPrefix is the only function — there is no regular-expression support.
• String literals may contain unicode, hexadecimal and octal escape sequences, for example attributes:"みんな". Escapes outside a string literal are invalid.
• A filter must be at most 256 bytes.

An unparseable filter, or one over the byte limit, is rejected with INVALID_ARGUMENT at creation time, rather than being accepted and silently ignored.

The filter is immutable

As in GCP, the filter is a property of the subscription that cannot change after creation. A subscriptions.patch that names filter in its update mask is rejected with INVALID_ARGUMENT, whatever value it carries — GCP rejects on the presence of the field in the mask, not on whether the value differs, so restating the current filter fails too. A patch that does not name filter in its mask succeeds and leaves the filter untouched, even when the request body carries one — the update mask governs, so clients that echo a whole subscription back keep working.

GCP requires an update mask on subscriptions.patch; floci-gcp also accepts a patch without one and treats it as replacing every field. On that path a body carrying the subscription's current filter is accepted and the filter is left alone, a body carrying a different one is rejected, and a body omitting it leaves the filter in place rather than clearing it.

To change a filter, follow the same path as in GCP: snapshot the subscription, create a new one with the desired filter, Seek to the snapshot, move subscribers over, then delete the old subscription.

Push Subscriptions

floci-gcp supports push subscriptions — it delivers messages to an HTTP endpoint you configure:

subscriptionAdminClient.createSubscription(
    SubscriptionName.of("floci-local", "my-sub"),
    TopicName.of("floci-local", "my-topic"),
    PushConfig.newBuilder()
        .setPushEndpoint("http://my-app:8080/pubsub/push")
        .build(),
    0);

Messages are delivered as HTTP POST requests to the configured endpoint.

Snapshots

Create and restore snapshots to replay messages:

// Create snapshot
snapshotAdminClient.createSnapshot(
    SnapshotName.of("floci-local", "my-snapshot"),
    SubscriptionName.of("floci-local", "my-sub"));

// Seek to snapshot (replay messages from snapshot point)
subscriptionAdminClient.seek(SeekRequest.newBuilder()
    .setSubscription(SubscriptionName.of("floci-local", "my-sub").toString())
    .setSnapshot(SnapshotName.of("floci-local", "my-snapshot").toString())
    .build());

Supported Operations

Publisher:

• CreateTopic
• UpdateTopic
• DeleteTopic
• GetTopic
• ListTopics
• ListTopicSubscriptions
• Publish

Subscriber:

• CreateSubscription
• UpdateSubscription
• DeleteSubscription
• GetSubscription
• ListSubscriptions
• Pull
• StreamingPull
• Acknowledge
• ModifyAckDeadline
• ModifyPushConfig
• CreateSnapshot
• GetSnapshot
• ListSnapshots
• UpdateSnapshot
• DeleteSnapshot
• Seek