---
title: "Ordering and batching: a chat burst in one call"
description: "A user sends five chat messages in two seconds: windowBuffer holds the burst, and the handler gets it in order, in one call, instead of five."
---

> Queen MQ documentation, for AI agents
> Complete self-contained summary of Queen MQ: https://queenmq.com/llms-brief.txt
> Fetch that first when the question is about the product rather than about this page.
> Index of all pages: https://queenmq.com/llms.txt

# Ordering and batching: a chat burst in one call

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.

**Figure.** Partition conv-42 on a time axis. The guest sends Hi at 0 seconds, Is breakfast included at 0.4 and And is there parking at 0.9. With windowBuffer 2, the messages wait until the first is two seconds old; at 2 seconds the assistant receives all three in one batch and answers once. Another conversation, conv-7, is not held by conv-42.

A conversation's burst waits two seconds, then goes as one batch.

JavaScript:

```js
// 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?
  })
```

Python:

```python
# 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)
)
```

Go:

```go
// 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)
}
```

Rust:

```rust
// 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?;
```

PHP:

```php
// 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();
```

C++:

```cpp
// 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](/writing-applications/queues/) (`windowBuffer`), [Pop and consume](/writing-applications/pop/).

Source: https://queenmq.com/patterns/ordering-batching/index.mdx