Queue

A general purpose queue that allows you to enqueue and dequeue Items singularly or in bulk. It is durable, which means that it will persist across restarts or system failures, and transactional, so you can access it safely.

Each Queue will need a unique ID and is created implicitly the first time you use it. Items can be Enqueued as a single Value, bytes, or Any object or can be Enqueued in bulk as a list of Items. They can then be Dequeued singularly or in bulk as well. In practice, you will want to use just one of these formats in a given Queue. Read more about Items to determine the best format for your data.

Imports and set up

To use a Queue, import the library where you would like to use it.

Python

from reboot.std.collections.queue.v1.queue import Queue
from reboot.std.item.v1.item import Item

TypeScript

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

Also make sure to include the Queue library and the dependent SortedMap library 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

async def main():
    application = Application(
        servicers=[MyServicer],
        libraries=[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";

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

Referencing Queues

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

Python

first_queue = Queue.ref("my-first-queue")
second_queue = Queue.ref("my-second-queue")
third_queue = Queue.ref("my-third-queue")

TypeScript

const firstQueue = Queue.ref("my-first-queue");
const secondQueue = Queue.ref("my-second-queue");

Methods

Enqueue

Enqueues 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_queue.enqueue(
    context,
    value=from_dict({"details": "details-go-here"}),
)

await second_queue.enqueue(context, bytes=b"my-bytes")

await third_queue.enqueue(context, any=pack(any))

TypeScript

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

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

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

Bulk Enqueue

Enqueues a list of Items.

Python

from reboot.protobuf import (
    from_bool,
    from_dict,
    from_int,
    from_list,
    from_str,
    pack,
)

await first_queue.enqueue(
    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_queue.enqueue(
    context,
    items=[
        Item(bytes=b"some-bytes"),
        Item(bytes=b"some-more-bytes"),
    ],
)

await third_queue.enqueue(
    context,
    items=[
        Item(any=pack(any)),
        Item(any=pack(any)),
    ],
)

TypeScript

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

await firstQueue.enqueue(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 secondQueue.enqueue(context, {
  items: [
    { bytes: new TextEncoder().encode("some-bytes") },
    { bytes: new TextEncoder().encode("some-more-bytes") },
  ],
});

Dequeue

Dequeues a single Item. If there are no Items in the Queue, this will wait until there is an item to return. For immediate return on an empty queue, see try_dequeue/tryDequeue.

Python

from reboot.protobuf import as_dict

item = await first_queue.dequeue(context)
print(as_dict(item.value)["details"])

item = await second_queue.dequeue(context)
print(item.bytes)

item = await third_queue.dequeue(context)
print(item.any)

TypeScript

const { value } = await firstQueue.dequeue(context);
console.log(value?.toJson()["details"]);

const { bytes } = await secondQueue.dequeue(context);
console.log(bytes);

Bulk Dequeue

By specifying bulk and at_most/atMost, you can also dequeue Items in bulk.

Python

items = await first_queue.dequeue(context, bulk=True, at_most=5)
# items.items is a list of Items

TypeScript

const { items } = await firstQueue.dequeue(context, {
  bulk: true,
  atMost: 5,
});
// `items` is a list of Items

Try Dequeue

Dequeues a single Item. If there are no Items in the Queue, this will return an empty response immediately. Otherwise, it will return item(s) as dequeue does. Like dequeue, passing bulk and at_most/atMost parameters allows dequeueing multiple Items at once.

Python

response = await queue.try_dequeue(context)
if response == DequeueResponse():
    print("The queue is empty.")
else:
    # Something was in the queue; handle it.

TypeScript

const response = await queue.tryDequeue(context, { bulk: true, atMost: 1 });
if (response.items.length === 0) {
  console.log("The queue is empty.");
} else {
  // Something was in the queue; handle it.
}

Empty

Returns a response whose empty field is true if the queue has no items in it, and false otherwise.

Python

response = await queue.empty(context)
if response.empty:
    print("The queue is empty.")

TypeScript

const response = await queue.empty(context);
if (response.empty) {
  console.log("The queue is empty.");
}