A chat: every conversation must be handled in order, and a slow one must not delay the others. And a guest rarely sends one message. They send three in a row: “Hi!”, “Is breakfast included?”, “And is there parking?”. An assistant that answers each message at once answers three times.
Every conversation is a partition of its own, created by its first message, so its messages stay in order and it waits on nobody else. windowBuffer adds the batching: new messages of a conversation wait until the first of them is two seconds old, and then go to the handler together.
// A burst waits 2 seconds, then goes to the handler as one batch.
await queen.queue('chat').config({ windowBuffer: 2 }).create()
// The guest sends three messages in a row.
for (const text of ['Hi!', 'Is breakfast included?', 'And is there parking?']) {
await queen.queue('chat').partition('conv-42').push([{ data: { text } }])
}
// The assistant answers the whole burst once.
await queen.queue('chat')
.group('assistant')
.subscriptionMode('all')
.batch(10) // the handler gets every message of the burst
.consume(async (messages) => {
question = messages.map((m) => m.data.text).join(' ')
console.log(question) // Hi! Is breakfast included? And is there parking?
})# A burst waits 2 seconds, then goes to the handler as one batch.
await queen.queue("chat").config({"windowBuffer": 2}).create()
# The guest sends three messages in a row.
for text in ["Hi!", "Is breakfast included?", "And is there parking?"]:
await queen.queue("chat").partition("conv-42").push([{"data": {"text": text}}])
# The assistant answers the whole burst once.
async def answer(messages):
question = " ".join(m["data"]["text"] for m in messages)
print(question) # Hi! Is breakfast included? And is there parking?
await (
queen.queue("chat")
.group("assistant")
.subscription_mode("all")
.batch(10) # the handler gets every message of the burst
.consume(answer)
)// A burst waits 2 seconds, then goes to the handler as one batch.
_, err = client.Queue("chat").Config(queen.QueueConfig{WindowBuffer: 2}).
Create().Execute(ctx)
if err != nil {
log.Fatal(err)
}
// The guest sends three messages in a row.
for _, text := range []string{"Hi!", "Is breakfast included?", "And is there parking?"} {
_, err := client.Queue("chat").Partition("conv-42").
Push(map[string]interface{}{"text": text}).
Execute(ctx)
if err != nil {
log.Fatal(err)
}
}
// The assistant answers the whole burst once.
err = client.Queue("chat").
Group("assistant").
SubscriptionMode(queen.SubscriptionModeAll).
Batch(10). // the handler gets every message of the burst
ConsumeBatch(ctx, func(ctx context.Context, messages []*queen.Message) error {
texts := make([]string, len(messages))
for i, m := range messages {
texts[i] = m.Data["text"].(string)
}
question := strings.Join(texts, " ")
fmt.Println(question) // Hi! Is breakfast included? And is there parking?
return nil
}).
Execute(ctx)
if err != nil {
log.Fatal(err)
}// A burst waits 2 seconds, then goes to the handler as one batch.
queen
.queue("chat")
.configure(QueueOptions { window_buffer: Some(2), ..Default::default() })
.await?;
// The guest sends three messages in a row.
for text in ["Hi!", "Is breakfast included?", "And is there parking?"] {
queen.queue("chat").partition("conv-42").push(json!({ "text": text })).await?;
}
// The assistant answers the whole burst once.
queen
.queue("chat")
.group("assistant")
.subscription_mode(SubscriptionMode::All)
.batch(10) // the handler gets every message of the burst
.consume_batch(|messages: Vec<Message>| async move {
let question = messages
.iter()
.filter_map(|m| m.data["text"].as_str())
.collect::<Vec<_>>()
.join(" ");
println!("{question}"); // Hi! Is breakfast included? And is there parking?
Ok::<_, queen_mq::Error>(())
})
.await?;// A burst waits 2 seconds, then goes to the handler as one batch.
$queen->queue('chat')->config(['windowBuffer' => 2])->create()->execute();
// The guest sends three messages in a row.
foreach (['Hi!', 'Is breakfast included?', 'And is there parking?'] as $text) {
$queen->queue('chat')->partition('conv-42')->push([['data' => ['text' => $text]]])
->execute();
}
// The assistant answers the whole burst once.
$queen->queue('chat')
->group('assistant')
->subscriptionMode('all')
->batch(10) // the handler gets every message of the burst
->consume(function (array $messages) {
$question = implode(' ', array_map(fn (array $m) => $m['data']['text'], $messages));
echo $question, "\n"; // Hi! Is breakfast included? And is there parking?
})
->execute();// A burst waits 2 seconds, then goes to the handler as one batch.
queen::QueueConfig chat;
chat.window_buffer = 2;
client.queue("chat").config(chat).create();
// The guest sends three messages in a row.
for (const char* text : {"Hi!", "Is breakfast included?", "And is there parking?"}) {
client.queue("chat").partition("conv-42").push({{{"data", {{"text", text}}}}});
}
// The assistant answers the whole burst once.
client.queue("chat")
.group("assistant")
.subscription_mode("all")
.batch(10) // the handler gets every message of the burst
.consume([&](const json& messages) {
question.clear();
for (const json& m : messages) {
if (!question.empty()) question += " ";
question += m["data"]["text"].get<std::string>();
}
// Hi! Is breakfast included? And is there parking?
std::cout << question << "\n";
});The handler ran once, two seconds after the first message, with the three texts.
Why
a conversation comes in bursts, not one message at a time. Waiting a moment gives the assistant the whole question. Each conversation waits only for its own burst, never for the others.
One thing to know: the window ends two seconds after the oldest waiting message, even if the guest keeps typing, so a long burst is answered in parts, every two seconds.
Built from: Queues (windowBuffer), Pop and consume.