The availability of a hotel changes many times a minute. A screen that shows it needs only the current value, and handling every old update on the way there is wasted work.
A consumer group with .conflation() gets only the newest message of each partition, and its cursor moves past the older ones.
// The availability of a hotel changes often; the screen needs only the last value.
for (const rooms of [12, 11, 9]) {
await queen.queue('availability').partition('hotel-7')
.push([{ data: { hotel: 7, rooms } }])
}
const [latest] = await queen.queue('availability')
.group('screen')
.subscriptionMode('all')
.conflation() // the newest message of each hotel, the rest skipped
.wait(false)
.pop()
console.log(latest.data) // { hotel: 7, rooms: 9 }
await queen.ack(latest, 'completed', { group: 'screen' })# The availability of a hotel changes often; the screen needs only the last value.
for rooms in [12, 11, 9]:
await queen.queue("availability").partition("hotel-7").push(
[{"data": {"hotel": 7, "rooms": rooms}}]
)
[latest] = await (
queen.queue("availability")
.group("screen")
.subscription_mode("all")
.conflation() # the newest message of each hotel, the rest skipped
.wait(False)
.pop()
)
print(latest["data"]) # {'hotel': 7, 'rooms': 9}
await queen.ack(latest, "completed", {"group": "screen"})// The availability of a hotel changes often; the screen needs only the last value.
for _, rooms := range []int{12, 11, 9} {
_, err := client.Queue("availability").Partition("hotel-7").
Push(map[string]interface{}{"hotel": 7, "rooms": rooms}).
Execute(ctx)
if err != nil {
log.Fatal(err)
}
}
newest, err := client.Queue("availability").
Group("screen").
SubscriptionMode(queen.SubscriptionModeAll).
Conflation(true). // the newest message of each hotel, the rest skipped
Wait(false).
Pop(ctx)
if err != nil {
log.Fatal(err)
}
latest := newest[0]
fmt.Println(latest.Data) // map[hotel:7 rooms:9]
_, err = client.Ack(ctx, latest, true, queen.AckOptions{ConsumerGroup: "screen"})
if err != nil {
log.Fatal(err)
}// The availability of a hotel changes often; the screen needs only the last value.
for rooms in [12, 11, 9] {
queen
.queue("availability")
.partition("hotel-7")
.push(json!({ "hotel": 7, "rooms": rooms }))
.await?;
}
let newest = queen
.queue("availability")
.group("screen")
.subscription_mode(SubscriptionMode::All)
.conflation(true) // the newest message of each hotel, the rest skipped
.wait(false)
.pop()
.await?;
let latest = &newest[0];
println!("{}", latest.data); // {"hotel":7,"rooms":9}
queen.ack(latest).await?;// The availability of a hotel changes often; the screen needs only the last value.
foreach ([12, 11, 9] as $rooms) {
$queen->queue('availability')->partition('hotel-7')
->push([['data' => ['hotel' => 7, 'rooms' => $rooms]]])
->execute();
}
[$latest] = $queen->queue('availability')
->group('screen')
->subscriptionMode('all')
->conflation() // the newest message of each hotel, the rest skipped
->wait(false)
->pop();
print_r($latest['data']); // [hotel => 7, rooms => 9]
$queen->ack($latest, 'completed', ['group' => 'screen']);// The availability of a hotel changes often; the screen needs only the last value.
for (int rooms : {12, 11, 9}) {
client.queue("availability").partition("hotel-7")
.push({{{"data", {{"hotel", 7}, {"rooms", rooms}}}}});
}
json latest = client.queue("availability")
.group("screen")
.subscription_mode("all")
.conflation() // the newest message of each hotel, the rest skipped
.wait(false)
.pop()
.at(0);
std::cout << latest["data"] << "\n"; // {"hotel":7,"rooms":9}
client.ack(latest, true, {{"group", "screen"}});Three updates were pushed (12, 11 and 9 rooms). The screen got 9, and the older two were never delivered to it.
Why
for a value that changes, only the last one matters. Conflation lets a slow consumer catch up in one step instead of replaying every update. Other groups of the same queue still get every message.
One thing to know: use it only where an old message has no meaning of its own. For events that each count, such as payments, every message must be handled.
Built from: Pop and consume.