Queue

The queue delivery method lets you consume a topic as an unordered set of messages.

Messages are delivered in parallel across all consumers of a consumer group, and any given message is delivered to only one consumer.

Publishing messages

In the msgs component, publishing is done to topics that are shared across all delivery methods, with the different delivery methods only applying at consumption time. Please refer to the topic publishing API for information on how to publish messages to a topic.

Receiving messages

Receiving messages is done using the diom.msgs.queue.receive API as follows:


import { Diom } from "diom";
import { Temporal } from "temporal-polyfill-lite";
 
const client = new Diom("AUTH_TOKEN");
const out = await client.msgs.queue.receive("topic-123", "group1", {
    batchSize: 100,
    batchWait: Temporal.Duration.from({ seconds: 10 }),
    leaseDuration: Temporal.Duration.from({ seconds: 30 }),
});


from datetime import timedelta
from diom import Diom
from diom.models import MsgQueueReceiveIn
 
client = Diom("AUTH_TOKEN")
out = client.msgs.queue.receive("topic-123", "group1", MsgQueueReceiveIn(
    batch_size=100,
    batch_wait_ms=timedelta(seconds=10),
    lease_duration_ms=timedelta(seconds=30),
))


use diom::{DiomClient, models::MsgQueueReceiveIn};
use std::time::Duration;
 
let client = DiomClient::new("AUTH_TOKEN".to_string(), None);
let out = client.msgs().queue().receive(
    "topic-123".to_string(),
    "group1".to_string(),
    MsgQueueReceiveIn::new()
        .with_batch_size(100)
        .with_batch_wait(Duration::from_secs(10))
        .with_lease_duration(Duration::from_secs(30)),
)
.await?;


import diom "diom.com/go/diom"
 
client, _ := diom.New("AUTH_TOKEN", nil)
batchSize := uint16(100)
batchWait := diom.DurationMs(10_000)
leaseDuration := diom.DurationMs(30_000)
out, err := client.Msgs().Queue().Receive(ctx, "topic-123", "group1", diom.MsgQueueReceiveIn{
    BatchSize:     &batchSize,
    BatchWait:     &batchWait,
    LeaseDuration: &leaseDuration,
})


import com.svix.diom.Diom;
import com.svix.diom.models.MsgQueueReceiveIn;
import com.svix.diom.models.MsgQueueReceiveOut;
import java.time.Duration;
 
Diom diom = new Diom("AUTH_TOKEN");
MsgQueueReceiveOut out = diom.msgs().queue().receive("topic-123", "group1",
    new MsgQueueReceiveIn()
        .batchSize((short) 100)
        .batchWait(Duration.ofSeconds(10))
        .leaseDuration(Duration.ofSeconds(30)));


export DIOM_AUTH_TOKEN='AUTH_TOKEN'
diom msgs queue receive 'topic-123' 'group1' '{
  "batch_size": 100,
  "batch_wait_ms": 10000,
  "lease_duration_ms": 30000
}'

Which returns:


{
    msgs: [
        {
            msg_id: "msg1",
            topic: "topic-123",
            value: Uint8Array(...),
            timestamp: Temporal.Instant(...),
        },
        {
            msg_id: "msg2",
            topic: "topic-123",
            value: Uint8Array(...),
            headers: {},
            timestamp: Temporal.Instant(...),
        },
        {
            msg_id: "msg3",
            topic: "topic-123",
            value: Uint8Array(...),
            headers: { h1: "v1" },
            timestamp: Temporal.Instant(...),
        },
    ]
}

Processing messages

When calling receive, the consumers own a message for the lease duration within which they need to finish processing the message and either acknowledge its processing (ack), or report an error (nack). If the lease expires before either is done, the message is treated as if there was an explicit nack.

Acknowledging messages (`ack`)

Acknowledging a message is how you mark a message processing as successful which means it won’t be returned again using the receive API.


await client.msgs.queue.ack("topic-123", "group1", { msgIds: ["msg2"] });


from diom.models import MsgQueueAckIn
 
client.msgs.queue.ack("topic-123", "group1", MsgQueueAckIn(msg_ids=["msg2"]))


use diom::models::MsgQueueAckIn;
 
client.msgs().queue().ack(
    "topic-123".to_string(),
    "group1".to_string(),
    MsgQueueAckIn::new(vec!["msg2".to_string()]),
)
.await?;


client.Msgs().Queue().Ack(ctx, "topic-123", "group1", diom.MsgQueueAckIn{
    MsgIds: []string{"msg2"},
})


import com.svix.diom.models.MsgQueueAckIn;
import java.util.List;
 
diom.msgs().queue().ack("topic-123", "group1",
    new MsgQueueAckIn().msgIds(List.of("msg2")));


diom msgs queue ack 'topic-123' 'group1' '{"msg_ids": ["msg2"]}'

Rejecting messages (`nack`)

Rejecting a message is how you report an error in the processing of the message. Rejecting a message immediately releases the lease and the message will become available for consumption again based on the configured retry schedule.


await client.msgs.queue.nack("topic-123", "group1", { msgIds: ["msg2"] });


from diom.models import MsgQueueNackIn
 
client.msgs.queue.nack("topic-123", "group1", MsgQueueNackIn(msg_ids=["msg2"]))


use diom::models::MsgQueueNackIn;
 
client.msgs().queue().nack(
    "topic-123".to_string(),
    "group1".to_string(),
    MsgQueueNackIn::new(vec!["msg2".to_string()]),
)
.await?;


client.Msgs().Queue().Nack(ctx, "topic-123", "group1", diom.MsgQueueNackIn{
    MsgIds: []string{"msg2"},
})


import com.svix.diom.models.MsgQueueNackIn;
import java.util.List;
 
diom.msgs().queue().nack("topic-123", "group1",
    new MsgQueueNackIn().msgIds(List.of("msg2")));


diom msgs queue nack 'topic-123' 'group1' '{"msg_ids": ["msg2"]}'

Extending message lease

If you need more time when processing a message, you can extend the message lease like so:


await client.msgs.queue.extendLease("topic-123", "group1", {
    msgIds: ["msg3"],
    // New lease duration starting from this moment
    leaseDuration: Temporal.Duration.from({ seconds: 10 }),
});


from datetime import timedelta
from diom.models import MsgQueueExtendLeaseIn
 
client.msgs.queue.extend_lease("topic-123", "group1", MsgQueueExtendLeaseIn(
    msg_ids=["msg3"],
    # New lease duration starting from this moment
    lease_duration_ms=timedelta(seconds=10),
))


use diom::models::MsgQueueExtendLeaseIn;
use std::time::Duration;
 
client.msgs().queue().extend_lease(
    "topic-123".to_string(),
    "group1".to_string(),
    MsgQueueExtendLeaseIn::new(vec!["msg3".to_string()])
        .with_lease_duration(Duration::from_secs(10)),
)
.await?;


leaseDuration := diom.DurationMs(10_000)
client.Msgs().Queue().ExtendLease(ctx, "topic-123", "group1", diom.MsgQueueExtendLeaseIn{
    MsgIds:        []string{"msg3"},
    LeaseDuration: &leaseDuration,
})


import com.svix.diom.models.MsgQueueExtendLeaseIn;
import java.time.Duration;
import java.util.List;
 
diom.msgs().queue().extendLease("topic-123", "group1", new MsgQueueExtendLeaseIn()
    .msgIds(List.of("msg3"))
    .leaseDuration(Duration.ofSeconds(10)));


diom msgs queue extend-lease 'topic-123' 'group1' '{
  "msg_ids": ["msg3"],
  "lease_duration_ms": 10000
}'

Dead letter queue (DLQ)

A dead letter queue (DLQ) is a holding queue where messages that fail to be processed successfully (after retries) are redirected for later inspection and debugging.

Message retry schedule

You can configure a retry schedule for a specific topic + consumer group combination, letting you manage how many times a message to retry, and whether to add a delay between retries.

Once a retry schedule is exhausted messages are sent to the DLQ if one is configured, otherwise they are just marked as processed and skipped. If no retry schedule is configured, messages are sent directly to the DLQ on failure.

Redriving the DLQ

You can consume the DLQ like any other topic, or alternatively you can automatically redrive all the messages back to the original queue:


await client.msgs.queue.redriveDlq("topic-123", "group1", {});


from diom.models import MsgQueueRedriveDlqIn
 
client.msgs.queue.redrive_dlq("topic-123", "group1", MsgQueueRedriveDlqIn())


use diom::models::MsgQueueRedriveDlqIn;
 
client.msgs().queue().redrive_dlq(
    "topic-123".to_string(),
    "group1".to_string(),
    MsgQueueRedriveDlqIn::new(),
)
.await?;


client.Msgs().Queue().RedriveDlq(ctx, "topic-123", "group1", diom.MsgQueueRedriveDlqIn{})


import com.svix.diom.models.MsgQueueRedriveDlqIn;
 
diom.msgs().queue().redriveDlq("topic-123", "group1", new MsgQueueRedriveDlqIn());


diom msgs queue redrive-dlq 'topic-123' 'group1'

Configuration

You can configure the behavior of consumer groups for a specific topic using the diom.msgs.queue.configure API as follows:


await client.msgs.queue.configure("topic-123", "group1", {
    // Retry delays in milliseconds: immediately, after 5s, after 10s, after 10s
    retrySchedule: [0, 5_000, 10_000, 10_000],
    dlqTopic: "dlq-for-topic-123",
});


from diom.models import MsgQueueConfigureIn
 
client.msgs.queue.configure("topic-123", "group1", MsgQueueConfigureIn(
    # Retry delays in milliseconds: immediately, after 5s, after 10s, after 10s
    retry_schedule=[0, 5_000, 10_000, 10_000],
    dlq_topic="dlq-for-topic-123",
))


use diom::models::MsgQueueConfigureIn;
 
client.msgs().queue().configure(
    "topic-123".to_string(),
    "group1".to_string(),
    MsgQueueConfigureIn::new()
        // Retry delays in milliseconds: immediately, after 5s, after 10s, after 10s
        .with_retry_schedule(vec![0, 5_000, 10_000, 10_000])
        .with_dlq_topic("dlq-for-topic-123".to_string()),
)
.await?;


dlq := "dlq-for-topic-123"
client.Msgs().Queue().Configure(ctx, "topic-123", "group1", diom.MsgQueueConfigureIn{
    // Retry delays in milliseconds: immediately, after 5s, after 10s, after 10s
    RetrySchedule: []uint64{0, 5_000, 10_000, 10_000},
    DlqTopic:      &dlq,
})


import com.svix.diom.models.MsgQueueConfigureIn;
import java.util.List;
 
diom.msgs().queue().configure("topic-123", "group1", new MsgQueueConfigureIn()
    // Retry delays in milliseconds: immediately, after 5s, after 10s, after 10s
    .retrySchedule(List.of(0L, 5_000L, 10_000L, 10_000L))
    .dlqTopic("dlq-for-topic-123"));


diom msgs queue configure 'topic-123' 'group1' '{
  "retry_schedule": [0, 5000, 10000, 10000],
  "dlq_topic": "dlq-for-topic-123"
}'

Advanced functionality

Time travel (`reset`)

Thanks to how topics and consumer groups work, Diom supports resetting a consumer group so that it behaves as if it started consuming a topic at a previous point in time.


diom.msgs.queue.reset(
    topic="topic-123",
    consumer_group="group1",
    // Can only set either timestamp or position, not both.
    "timestamp": Date(...),
    "position": "earliest" | "latest",
}