PubSub

A pub/sub implementation that allows you to publish to Topics and subscribe Queues to those Topics.

Each Topic is associated with a unique ID and is created implicitly the first time it is published to or subscribed to.

Messages are marshalled as Items to be published to Topics. They can be published as a single Value, bytes, or Any object or can be published in bulk as a list of Items. In practice, you will want to use just one of these formats for your messages in a given Topic. Read more about Items to determine the best format for your data.

Subscription is Queue based. That is, when you subscribe to a Topic, you reference a Queue by its ID. Any messages published to that Topic will be received on the specified Queue and can be dequeued to be read.

Imports and servicers

To use the pub/sub library, import the library where you would like to use it. To read messages from a subscribed Queue, you will also need to import the Queue library.

Python

from reboot.std.collections.queue.v1.queue import Queue
from reboot.std.pubsub.v1.pubsub import Topic

TypeScript

import { Queue } from "@reboot-dev/reboot-std/collections/queue/v1";
import { Topic } from "@reboot-dev/reboot-std/pubsub/v1";

Also make sure to include the pub/sub library and the dependent Queue and SortedMap libraries when starting up your Application. (Note: this import is different from above.)

Python

from reboot.std.collections.queue.v1.queue import queue_library
from reboot.std.collections.v1.sorted_map import sorted_map_library
from reboot.std.pubsub.v1.pubsub import pubsub_library

async def main():
    application = Application(
        servicers=[MyServicer],
        libraries=[
            pubsub_library(),
            queue_library(),
            sorted_map_library(),
        ],
    )
    await application.run()

TypeScript

import { queueLibrary } from "@reboot-dev/reboot-std/collections/queue/v1";
import { sortedMapLibrary } from "@reboot-dev/reboot-std/collections/v1/sorted_map.js";
import { pubsubLibrary } from "@reboot-dev/reboot-std/pubsub/v1";

new Application({
  servicers: [MyServicer],
  libraries: [pubsubLibrary(), queueLibrary(), sortedMapLibrary()],
  initialize,
}).run();

Referencing Topics

This creates references to Topics with IDs such as "my-first-topic" and "my-second-topic". You can then call methods on these references.

Python

first_topic = Topic.ref("my-first-topic")
second_topic = Topic.ref("my-second-topic")
third_topic = Topic.ref("my-third-topic")

TypeScript

const firstTopic = Topic.ref("my-first-topic");
const secondTopic = Topic.ref("my-second-topic");

Methods

Publish

Publish a single Item directly from a Value, bytes, or Any.

Note: The ability to add an Item with an Any format is only supported in Python, not TypeScript.

Python

from reboot.protobuf import from_dict, pack

await first_topic.publish(
    context,
    value=from_dict({"details": "details-go-here"}),
)

await second_topic.publish(context, bytes=b"my-bytes")

await third_topic.publish(context, any=pack(any))

TypeScript

import { Value } from "@bufbuild/protobuf";

await firstTopic.publish(context, {
  value: Value.fromJson({ details: "details-go-here" }),
});

await secondTopic.publish(context, {
  bytes: new TextEncoder().encode("my-bytes"),
});

Subscribe

Subscribe a single Queue as referenced by its ID to a Topic. Any subsequent messages published to the Topic will be received by dequeueing on the Queue.

Note: Any messages published to the Topic before the Queue is subscribed are not guaranteed to be received by the Queue.

Python

await first_topic.subscribe(context, queue_id="receiving-queue")

response = await Queue.ref("receiving-queue").dequeue(context)

TypeScript

await firstTopic.subscribe(context, {
  queueId: "receiving-queue",
});

const response = await Queue.ref("receiving-queue").dequeue(context);

Bulk Publish

Publish a list of messages, formatted as Items.

This example demonstrates how to store different types of data in Value, but in practice, you will want to use a uniform structure for all Values published to a Topic.

Note: The ability to add an Item with an Any format is only supported in Python, not TypeScript.

Python

from reboot.protobuf import (
    from_bool,
    from_dict,
    from_int,
    from_list,
    from_str,
    pack,
)
from reboot.std.item.v1.item import Item

await first_topic.publish(
    context,
    items=[
        Item(value=from_bool(True)),
        Item(value=from_int(3)),
        Item(value=from_str("apple")),
        Item(value=from_list(["a", "b", "c"])),
        Item(value=from_dict({"details": "details-go-here"})),
    ],
)

await second_topic.publish(
    context,
    items=[
        Item(bytes=b"some-bytes"),
        Item(bytes=b"some-more-bytes"),
    ],
)

await third_topic.publish(
    context,
    items=[
        Item(any=pack(any)),
        Item(any=pack(any)),
    ],
)

TypeScript

import { Value } from "@bufbuild/protobuf";

await firstTopic.publish(context, {
  items: [
    { value: Value.fromJson(null) },
    { value: Value.fromJson(true) },
    { value: Value.fromJson(3) },
    { value: Value.fromJson("apple") },
    { value: Value.fromJson(["a", "b", "c"]) },
    { value: Value.fromJson({ details: "details-go-here" }) },
  ],
});

await secondTopic.publish(context, {
  items: [
    { bytes: new TextEncoder().encode("some-bytes") },
    { bytes: new TextEncoder().encode("some-more-bytes") },
  ],
});