The booking page needs a price from the pricing service, now. If no answer comes in half a second, it shows “ask us”. A durable queue is too heavy for that, and a direct HTTP call ties the two services together.
Ephemeral queues live in memory, outside the log. The caller makes a private inbox, sends it as replyTo, and waits for the answer with a timeout. Any instance of the pricing service can answer.
// The pricing service answers quote requests, in memory.
async function pricing() {
const { messages } = await queen.ephemeral.pop('quotes', { group: 'pricing', wait: true })
for (const m of messages) {
await queen.ephemeral.push(m.payload.replyTo, [{ room: m.payload.room, price: 129 }])
}
await queen.ephemeral.ack('quotes', messages, { group: 'pricing' })
}
// The booking page asks, and waits at most half a second for the answer.
async function quote(room) {
const inbox = `inbox-${crypto.randomUUID()}`
await queen.ephemeral.push('quotes', [{ room, replyTo: inbox }])
const reply = await queen.ephemeral.pop(inbox,
{ wait: true, timeout: 500, commitOnDelivery: true })
return reply.messages[0]?.payload ?? { room, price: null } // too slow: show "ask us"
}
const [deluxe] = await Promise.all([quote('deluxe'), pricing()])
console.log(deluxe) // { room: 'deluxe', price: 129 }# The pricing service answers quote requests, in memory.
async def pricing():
messages = (await queen.ephemeral.pop("quotes", group="pricing", wait=True))["messages"]
for m in messages:
reply = {"room": m["payload"]["room"], "price": 129}
await queen.ephemeral.push(m["payload"]["replyTo"], [reply])
await queen.ephemeral.ack("quotes", messages, group="pricing")
# The booking page asks, and waits at most half a second for the answer.
async def quote(room):
inbox = f"inbox-{uuid.uuid4()}"
await queen.ephemeral.push("quotes", [{"room": room, "replyTo": inbox}])
reply = await queen.ephemeral.pop(inbox, wait=True, timeout=500,
commit_on_delivery=True)
if reply["messages"]:
return reply["messages"][0]["payload"]
return {"room": room, "price": None} # too slow: show "ask us"
deluxe, _ = await asyncio.gather(quote("deluxe"), pricing())
print(deluxe) # {'room': 'deluxe', 'price': 129}// The pricing service answers quote requests, in memory.
pricing := func() error {
batch, err := client.Ephemeral().Pop(ctx, "quotes",
queen.EphemeralPopOptions{Group: "pricing", Wait: true})
if err != nil {
return err
}
for _, m := range batch.Messages {
var req struct{ Room, ReplyTo string }
if err := json.Unmarshal(m.Payload, &req); err != nil {
return err
}
_, err := client.Ephemeral().Push(ctx, req.ReplyTo,
[]map[string]interface{}{{"room": req.Room, "price": 129}})
if err != nil {
return err
}
}
_, err = client.Ephemeral().Ack(ctx, "quotes", batch.Messages,
queen.EphemeralAckOptions{Group: "pricing"})
return err
}
// The booking page asks, and waits at most half a second for the answer.
quote := func(room string) (map[string]interface{}, error) {
inbox := "inbox-" + queen.GenerateUUID()
_, err := client.Ephemeral().Push(ctx, "quotes",
[]map[string]interface{}{{"room": room, "replyTo": inbox}})
if err != nil {
return nil, err
}
reply, err := client.Ephemeral().Pop(ctx, inbox,
queen.EphemeralPopOptions{Wait: true, TimeoutMillis: 500, CommitOnDelivery: true})
if err != nil {
return nil, err
}
answer := map[string]interface{}{"room": room, "price": nil} // too slow: show "ask us"
if len(reply.Messages) > 0 {
err = json.Unmarshal(reply.Messages[0].Payload, &answer)
}
return answer, err
}
answered := make(chan error, 1)
go func() { answered <- pricing() }()
deluxe, err := quote("deluxe")
if err := errors.Join(err, <-answered); err != nil {
log.Fatal(err)
}
fmt.Println(deluxe) // map[price:129 room:deluxe]// The pricing service answers quote requests, in memory.
async fn pricing(queen: &Queen) -> queen_mq::Result<()> {
let eph = queen.ephemeral();
let batch = eph.pop("quotes").group("pricing").wait(true).send().await?;
for m in &batch.messages {
let reply_to = m.payload["replyTo"].as_str().unwrap_or_default();
let answer = json!({ "room": m.payload["room"], "price": 129 });
eph.push(reply_to, answer).send().await?;
}
eph.ack(&batch).await?;
Ok(())
}
// The booking page asks, and waits at most half a second for the answer.
async fn quote(queen: &Queen, room: &str) -> queen_mq::Result<Value> {
let eph = queen.ephemeral();
let inbox = format!("inbox-{}", queen_mq::uuid::uuidv7());
eph.push("quotes", json!({ "room": room, "replyTo": inbox })).send().await?;
let reply = eph
.pop(&inbox)
.wait(true)
.poll_timeout(Duration::from_millis(500))
.commit_on_delivery(true)
.send()
.await?;
Ok(match reply.messages.first() {
Some(m) => m.payload.clone(),
None => json!({ "room": room, "price": null }), // too slow: show "ask us"
})
}
let (deluxe, _) = tokio::try_join!(quote(&queen, "deluxe"), pricing(&queen))?;
println!("{deluxe}"); // {"price":129,"room":"deluxe"}// The pricing service answers quote requests, in memory.
$pricing = function () use ($queen): void {
$messages = $queen->ephemeral()
->pop('quotes', ['group' => 'pricing', 'wait' => true])['messages'];
foreach ($messages as $m) {
$queen->ephemeral()->push($m['payload']['replyTo'],
[['room' => $m['payload']['room'], 'price' => 129]]);
}
$queen->ephemeral()->ack('quotes', $messages, ['group' => 'pricing']);
};
// The booking page asks, and waits at most half a second for the answer.
$quote = function (string $room) use ($queen): array {
$inbox = 'inbox-' . bin2hex(random_bytes(8));
$queen->ephemeral()->push('quotes', [['room' => $room, 'replyTo' => $inbox]]);
$reply = $queen->ephemeral()->pop($inbox,
['wait' => true, 'timeout' => 500, 'commitOnDelivery' => true]);
// too slow: show "ask us"
return $reply['messages'][0]['payload'] ?? ['room' => $room, 'price' => null];
};
if (($pid = pcntl_fork()) === 0) { // PHP has no threads: pricing runs
$pricing(); // in a process of its own
exit(0);
}
$deluxe = $quote('deluxe');
pcntl_waitpid($pid, $status);
print_r($deluxe); // [room => deluxe, price => 129]// The pricing service answers quote requests, in memory.
auto pricing = [&] {
queen::EphemeralPopOptions take;
take.group = "pricing";
take.wait = true;
json batch = client.ephemeral().pop("quotes", take);
for (const json& m : batch["messages"]) {
json answer = {{"room", m["payload"]["room"]}, {"price", 129}};
client.ephemeral().push(m["payload"]["replyTo"], json::array({answer}));
}
queen::EphemeralAckOptions done_with;
done_with.group = "pricing";
client.ephemeral().ack("quotes", batch["messages"], done_with);
};
// The booking page asks, and waits at most half a second for the answer.
auto quote = [&](const std::string& room) -> json {
auto inbox = "inbox-" + queen::util::generate_uuid_v7();
json request = {{"room", room}, {"replyTo", inbox}};
client.ephemeral().push("quotes", json::array({request}));
queen::EphemeralPopOptions reply_wait;
reply_wait.wait = true;
reply_wait.timeout_millis = 500;
reply_wait.commit_on_delivery = true;
json reply = client.ephemeral().pop(inbox, reply_wait);
if (reply["messages"].empty()) {
return {{"room", room}, {"price", nullptr}}; // too slow: show "ask us"
}
return reply["messages"][0]["payload"];
};
auto answering = std::async(std::launch::async, pricing);
json deluxe = quote("deluxe");
answering.get();
std::cout << deluxe << "\n"; // {"price":129,"room":"deluxe"}The answer came back in time: {room: 'deluxe', price: 129}. With no pricing service running, the caller gave up after half a second and got price: null.
Why
an answer that comes too late is worth nothing, so it does not need a disk. In memory a round trip costs very little, and the service scales like any consumer group.
One thing to know: a node that crashes loses the ephemeral messages it held. There are no transactions, dead letters or replays on these queues.
Built from: Ephemeral queues.