Ordered Delivery

package main

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

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

func projectID() string {
	if v := os.Getenv("PUBSUB_PROJECT_ID"); v != "" {
		return v
	}
	return "my-project"
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	client, err := pubsub.NewClient(ctx, projectID())
	if err != nil {
		log.Fatalf("NewClient: %v", err)
	}
	defer client.Close()

	// CreateTopic, then enable ordering on the publisher handle (it serializes
	// publishes per key). The subscription must enable ordering too — both sides.
	topic, err := client.CreateTopic(ctx, "ordered")
	if err != nil {
		log.Fatalf("CreateTopic: %v", err)
	}
	topic.EnableMessageOrdering = true
	defer topic.Stop()

	if _, err := client.CreateSubscription(ctx, "sub-ordered", pubsub.SubscriptionConfig{
		Topic:                 topic,
		AckDeadline:           10 * time.Second,
		EnableMessageOrdering: true,
	}); err != nil {
		log.Fatalf("CreateSubscription: %v", err)
	}

	// Two keys interleaved + one keyless. Await each result so the per-key publish
	// order is preserved across keys.
	plan := []struct{ key, body string }{
		{"cust-7", "A1"}, {"cust-9", "B1"}, {"cust-7", "A2"},
		{"", "keyless"}, {"cust-9", "B2"}, {"cust-7", "A3"}, {"cust-9", "B3"},
	}
	for _, p := range plan {
		m := &pubsub.Message{Data: []byte(p.body)}
		if p.key != "" {
			m.OrderingKey = p.key
		}
		if _, err := topic.Publish(ctx, m).Get(ctx); err != nil {
			log.Fatalf("Publish %q: %v", p.body, err)
		}
	}

	// Pull one at a time, ack before the next pull → connector releases the next
	// per-key message in order.
	subClient, err := pubsub.NewSubscriberClient(ctx)
	if err != nil {
		log.Fatalf("NewSubscriberClient: %v", err)
	}
	defer subClient.Close()
	subPath := fmt.Sprintf("projects/%s/subscriptions/%s", projectID(), "sub-ordered")

	for i := 0; i < len(plan); i++ {
		resp, err := subClient.Pull(ctx, &pubsubpb.PullRequest{Subscription: subPath, MaxMessages: 1})
		if err != nil {
			log.Fatalf("Pull: %v", err)
		}
		if len(resp.GetReceivedMessages()) == 0 {
			i--
			continue
		}
		rm := resp.GetReceivedMessages()[0]
		key := rm.GetMessage().GetOrderingKey() // surfaced from _pubsub_ordering_key.
		fmt.Printf("received body=%q ordering_key=%q\n", string(rm.GetMessage().GetData()), key)
		_ = subClient.Acknowledge(ctx, &pubsubpb.AcknowledgeRequest{
			Subscription: subPath, AckIds: []string{rm.GetAckId()},
		})
	}
	// Per-key order is preserved: cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3.
}

import os

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


def project_id() -> str:
    return os.environ.get("PUBSUB_PROJECT_ID", "my-project")


def main() -> None:
    proj = project_id()
    # The publisher MUST enable message ordering (it serializes publishes per key).
    publisher = pubsub_v1.PublisherClient(
        publisher_options=PublisherOptions(enable_message_ordering=True)
    )
    subscriber = pubsub_v1.SubscriberClient()
    topic_path = publisher.topic_path(proj, "ordered")          # -> gcp.ordered
    sub_path = subscriber.subscription_path(proj, "sub-ordered")  # -> gcp.sub.sub-ordered

    publisher.create_topic(request={"name": topic_path})
    # The subscription must also enable ordering.
    subscriber.create_subscription(
        request={"name": sub_path, "topic": topic_path, "enable_message_ordering": True}
    )

    # Two keys interleaved + one keyless; await each publish to preserve order.
    plan = [("cust-7", "A1"), ("cust-9", "B1"), ("cust-7", "A2"),
            ("", "keyless"), ("cust-9", "B2"), ("cust-7", "A3"), ("cust-9", "B3")]
    for key, body in plan:
        publisher.publish(topic_path, body.encode(), ordering_key=key).result(timeout=15)

    # Pull one at a time, ack before the next pull → next per-key message released.
    for _ in range(len(plan)):
        resp = subscriber.pull(request={"subscription": sub_path, "max_messages": 1}, timeout=20)
        if not resp.received_messages:
            continue
        rm = resp.received_messages[0]
        print(f"received body={rm.message.data.decode()!r} ordering_key={rm.message.ordering_key!r}")
        subscriber.acknowledge(request={"subscription": sub_path, "ack_ids": [rm.ack_id]})
    subscriber.close()
    # cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.


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.Subscription;
import com.google.pubsub.v1.SubscriptionName;
import com.google.pubsub.v1.TopicName;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import java.util.List;

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, "ordered");                 // -> gcp.ordered
        SubscriptionName sub = SubscriptionName.of(projectId, "sub-ordered"); // -> gcp.sub.sub-ordered

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

            topicAdmin.createTopic(topic);
            // The subscription enables ordering; the publisher orders per key below.
            subAdmin.createSubscription(Subscription.newBuilder()
                    .setName(sub.toString()).setTopic(topic.toString())
                    .setAckDeadlineSeconds(10).setEnableMessageOrdering(true).build());

            // Two keys interleaved + one keyless; publish in order with an ordering key.
            String[][] plan = {
                {"cust-7", "A1"}, {"cust-9", "B1"}, {"cust-7", "A2"},
                {"", "keyless"}, {"cust-9", "B2"}, {"cust-7", "A3"}, {"cust-9", "B3"},
            };
            for (String[] p : plan) {
                PubsubMessage.Builder m = PubsubMessage.newBuilder()
                        .setData(ByteString.copyFromUtf8(p[1]));
                if (!p[0].isEmpty()) {
                    m.setOrderingKey(p[0]);
                }
                topicAdmin.publish(PublishRequest.newBuilder()
                        .setTopic(topic.toString()).addMessages(m.build()).build());
            }

            // Pull one at a time, ack before the next pull → next per-key message released.
            for (int i = 0; i < plan.length; i++) {
                PullResponse resp = subStub.pullCallable().call(PullRequest.newBuilder()
                        .setSubscription(sub.toString()).setMaxMessages(1).build());
                if (resp.getReceivedMessagesCount() == 0) {
                    i--;
                    continue;
                }
                ReceivedMessage got = resp.getReceivedMessages(0);
                System.out.printf("received body=%s ordering_key=%s%n",
                        got.getMessage().getData().toStringUtf8(), got.getMessage().getOrderingKey());
                subStub.acknowledgeCallable().call(AcknowledgeRequest.newBuilder()
                        .setSubscription(sub.toString()).addAckIds(got.getAckId()).build());
            }
            // cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.
            List.of(); // no-op to keep imports tidy
        } finally {
            channel.shutdown();
        }
    }
}

import { PubSub, v1 } from "@google-cloud/pubsub";

const projectId = process.env["PUBSUB_PROJECT_ID"] ?? "my-project";
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, "ordered");   // -> gcp.ordered
  const sub = subscriber.subscriptionPath(projectId, "sub-ordered"); // -> gcp.sub.sub-ordered

  await publisher.createTopic({ name: topic });
  // The subscription enables ordering; each publish below carries an orderingKey.
  await subscriber.createSubscription({ name: sub, topic, ackDeadlineSeconds: 10, enableMessageOrdering: true });

  // Two keys interleaved + one keyless; publish each in turn to preserve order.
  const plan: Array<[string, string]> = [
    ["cust-7", "A1"], ["cust-9", "B1"], ["cust-7", "A2"],
    ["", "keyless"], ["cust-9", "B2"], ["cust-7", "A3"], ["cust-9", "B3"],
  ];
  for (const [orderingKey, body] of plan) {
    await publisher.publish({ topic, messages: [{ data: Buffer.from(body), orderingKey }] });
  }

  // Pull one at a time, ack before the next pull → next per-key message released.
  for (let i = 0; i < plan.length; i++) {
    const [pull] = await subscriber.pull({ subscription: sub, maxMessages: 1 });
    if (!pull.receivedMessages || pull.receivedMessages.length === 0) {
      i--;
      continue;
    }
    const rm = pull.receivedMessages[0];
    const body = Buffer.from(rm.message!.data as Uint8Array).toString("utf8");
    console.log(`received body=${body} ordering_key=${rm.message!.orderingKey}`);
    await subscriber.acknowledge({ subscription: sub, ackIds: [rm.ackId!] });
  }
  // cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.
}

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, "ordered");                  // -> gcp.ordered
var subName = SubscriptionName.FromProjectSubscription(projectId, "sub-ordered");  // -> gcp.sub.sub-ordered

var publisher = await new PublisherServiceApiClientBuilder
{
    EmulatorDetection = EmulatorDetection.EmulatorOnly,
}.BuildAsync();
var subscriber = await new SubscriberServiceApiClientBuilder
{
    EmulatorDetection = EmulatorDetection.EmulatorOnly,
}.BuildAsync();

await publisher.CreateTopicAsync(topicName);
// The subscription enables ordering; each publish below carries an OrderingKey.
await subscriber.CreateSubscriptionAsync(new Subscription
{
    SubscriptionName = subName,
    TopicAsTopicName = topicName,
    AckDeadlineSeconds = 10,
    EnableMessageOrdering = true,
});

// Two keys interleaved + one keyless; publish each in turn to preserve order.
var plan = new[]
{
    ("cust-7", "A1"), ("cust-9", "B1"), ("cust-7", "A2"),
    ("", "keyless"), ("cust-9", "B2"), ("cust-7", "A3"), ("cust-9", "B3"),
};
foreach (var (orderingKey, body) in plan)
{
    var msg = new PubsubMessage { Data = ByteString.CopyFromUtf8(body) };
    if (orderingKey.Length > 0) msg.OrderingKey = orderingKey;
    await publisher.PublishAsync(topicName, new[] { msg });
}

// Pull one at a time, ack before the next pull → next per-key message released.
for (var i = 0; i < plan.Length; i++)
{
    var pull = await subscriber.PullAsync(subName, maxMessages: 1);
    if (pull.ReceivedMessages.Count == 0) { i--; continue; }
    var rm = pull.ReceivedMessages[0];
    Console.WriteLine($"received body={rm.Message.Data.ToStringUtf8()} ordering_key={rm.Message.OrderingKey}");
    await subscriber.AcknowledgeAsync(subName, new[] { rm.AckId });
}
// cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.

# 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
sub_admin   = pubsub.subscription_admin
topic_path  = pubsub.topic_path("ordered")          # -> gcp.ordered
sub_path    = pubsub.subscription_path("sub-ordered") # -> gcp.sub.sub-ordered

topic = topic_admin.create_topic(name: topic_path)
# The subscription enables ordering; each publish below carries an ordering_key.
sub_admin.create_subscription(name: sub_path, topic: topic_path,
                              ack_deadline_seconds: 10, enable_message_ordering: true)

# Two keys interleaved + one keyless. message_ordering: true serializes per key.
publisher = pubsub.publisher(topic.name, async: { ordered: true })
plan = [["cust-7", "A1"], ["cust-9", "B1"], ["cust-7", "A2"],
        ["", "keyless"], ["cust-9", "B2"], ["cust-7", "A3"], ["cust-9", "B3"]]
plan.each do |key, body|
  publisher.publish(body, ordering_key: key)
end
publisher.async_publisher.stop! # flush ordered publishes in order.

# Pull one at a time, ack before the next pull → next per-key message released.
subscriber = pubsub.subscriber(sub_path)
plan.length.times do
  received = subscriber.pull(immediate: false, max: 1)
  next if received.empty?

  rcv = received.first
  puts "received body=#{rcv.data} ordering_key=#{rcv.ordering_key.inspect}"
  rcv.acknowledge!
end
# cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.