Skip to content

Stream processing: windows, reduce and gates

Streams run in your process and read any queue: tumbling, sliding and session windows, reduce, and gates, with the source ack, the state and the output in one commit.

Updated View as Markdown

A stream reads a queue and writes its results to another. The partition is the key, so each customer gets its own windows. Streams are in the JavaScript, Python, Go and Rust clients; PHP and C++ have none, and a stream reads any queue, whoever pushes to it. Like a new group it starts at the tail: push a few more orders and a result arrives a minute later.

import { Stream } from 'queen-mq'

const revenue = await Stream
  .from(queen.queue('orders'))                 // one window per partition: per customer
  .windowTumbling({ seconds: 60 })
  .aggregate({ count: () => 1, sum: (order) => order.total })
  .to(queen.queue('revenue-per-minute'))       // one result per customer per minute
  .run({ queryId: 'revenue-per-minute', url: 'http://localhost:6632' })

A stream reads its queue as a consumer group (streams.<queryId>) and keeps its state, the open windows, in the broker. Each time it closes windows, it writes the new state, pushes the results and acks the messages they came from in one transaction.

One cycle of a stream. The runner pops a batch of the source queue for one partition, under a lease of the stream's consumer group. It runs the window and the aggregate, reading and writing the window state, which lives in the broker. Then it commits one transaction: the ack of the batch, the new window state, and the push of the closed windows to the output queue. A crash before the commit repeats the cycle; after it, nothing is lost and nothing is counted twice.One cycle of a stream. The runner pops a batch of the source queue for one partition, under a lease of the stream's consumer group. It runs the window and the aggregate, reading and writing the window state, which lives in the broker. Then it commits one transaction: the ack of the batch, the new window state, and the push of the closed windows to the output queue. A crash before the commit repeats the cycle; after it, nothing is lost and nothing is counted twice.
One cycle of a stream is one transaction.
Why
counting, summing and grouping events over time is a job most systems hand to a separate stream processor, with its own cluster and its own idea of what was processed. Here it runs in a normal process, its state lives next to the data, and one commit keeps the results and the inputs in step: a crash can neither lose a result nor count an event twice.

Four kinds of window

Window Example What it is, and when to use it
windowTumbling seconds: 60 Fixed windows, one after the other. Revenue per minute.
windowSliding size: 600, slide: 60 Overlapping windows: the last 10 minutes, every minute. Each event counts in several windows, so it keeps more state.
windowSession gap: 1800 One window per burst of activity, closed after a quiet gap. A visit, a conversation.
windowCron every: 'hour' Windows aligned to the clock (second, minute, hour, day, week). Hourly and daily reports.
One partition's events as dots on a time axis: a burst, a quiet stretch, another burst. Tumbling windows of a fixed size sit side by side. Sliding windows overlap, so one event counts in several of them. Session windows follow the bursts: a window ends after a gap of quiet. Cron windows are aligned to the clock, one per hour.One partition's events as dots on a time axis: a burst, a quiet stretch, another burst. Tumbling windows of a fixed size sit side by side. Sliding windows overlap, so one event counts in several of them. Session windows follow the bursts: a window ends after a gap of quiet. Cron windows are aligned to the clock, one per hour.
The four kinds of window on one partition's events.
const loginRate = await Stream
  .from(queen.queue('logins'))                   // one partition per user
  .windowSliding({ size: 600, slide: 60 })     // the last 10 minutes, every minute
  .aggregate({ count: () => 1 })
  .to(queen.queue('login-rate'))
  .run({ queryId: 'login-rate', url: 'http://localhost:6632' })
const visits = await Stream
  .from(queen.queue('clicks'))                   // one partition per visitor
  .windowSession({ gap: 1800 })                  // a visit ends after 30 minutes of quiet
  .aggregate({ count: () => 1 })
  .foreach(async (visit) => console.log('visit ended', visit))   // acked after this returns
  .run({ queryId: 'visits', url: 'http://localhost:6632' })
Why
“in the last ten minutes”, “per visit” and “per hour” are different questions, and each window shape answers one of them without code of your own.

When it happened, or when it arrived

By default a window holds the events that arrived during it. Give eventTime a function that reads the time from the message, and windows hold the events that happened during it instead. It gets the whole message, so the event’s own fields are under data; the aggregate’s functions get the event itself. Go and Rust want the time as epoch milliseconds.

const hourly = await Stream
  .from(queen.queue('sales'))
  .windowCron({
    every: 'hour',                     // on the clock: 10:00 to 11:00, 11:00 to 12:00
    eventTime: (msg) => msg.data.soldAt,  // when the sale happened, not when it arrived
    allowedLateness: 120,              // a sale up to 2 minutes late still counts
  })
  .aggregate({ count: () => 1, sum: (sale) => sale.amount, max: (sale) => sale.amount })
  .to(queen.queue('sales-per-hour'))
  .run({ queryId: 'sales-per-hour', url: 'http://localhost:6632', subscriptionMode: 'all' })

allowedLateness says how far behind the newest event an event may be and still count. A later one is dropped, or with onLate: 'include' it opens a fresh window, so the window is sent a second time with only the late events. gracePeriod keeps a window open a little after its end, and idleFlushMs (5 seconds) closes the windows of partitions that went quiet. subscriptionMode: 'all' makes a new stream read what the queue already holds.

Why
events arrive late and out of order: a phone that was offline, a batch upload, a retry. An hourly report should count a sale in the hour it happened, not in the hour its message got through.

Shaping the data

  • map(fn) changes each message, filter(fn) drops the ones you do not want, and flatMap(fn) turns one message into several values.
  • aggregate({ ... }) folds a window into named numbers. The name decides the fold: count, sum, min, max and avg; any other name adds up. reduce(fn, initial) is the general form, for anything else.
  • keyBy(fn) groups by something other than the partition (per page instead of per user). Several workers can then write the same key, which slows them down; when that matters, push to a queue partitioned by the new key first, and stream from it.
Why
most stream jobs are the same few steps: drop noise, reshape, group, fold. Having them as building blocks keeps a stream to a few readable lines.

Two ways to end a stream

.to(queue) pushes each result to another queue, in the same commit as the state and the acks, into the partition of the same name unless you pass { partition }. .foreach(fn) calls your function for each result, and acks only after it returns: a crash in the middle calls it again.

Why
a result that stays inside Queen can be exactly once. A call to the outside world (an email, an API) cannot, so foreach makes it at least once and says so. For an external effect that must happen once, send the result to a queue and let a worker do it with an idempotency key.

Rate limits with a gate

gate(fn) decides, message by message, whether a message may go on now, with state kept per partition. Two ready-made gates are included: tokenBucketGate and slidingWindowGate. Rust has no ready-made gates yet, so its code writes the same token bucket as a function, with the same state.

import { tokenBucketGate } from 'queen-mq'

const limiter = await Stream
  .from(queen.queue('outgoing-sms'))             // one partition per customer
  .gate(tokenBucketGate({ capacity: 10, refillPerSec: 1 }))   // 10 at once, then 1 a second
  .foreach(async (sms) => sendSms(sms))
  .run({ queryId: 'sms-limiter', url: 'http://localhost:6632' })

A message the gate refuses stops its partition there: it and the ones after it are delivered again, in order, when the lease runs out.

Why
an SMS provider, a partner API or a rate-limited email service accepts only so many calls a second. A gate per customer keeps you inside those limits without losing order and without a queue of your own for the waiting messages.

Running a stream

run() takes the stream’s queryId, its identity: its state is stored under it, so a process restarted with the same id carries on with the same windows, and several processes with the same id share the partitions between them. batchSize (200) and maxPartitions (4) size each cycle, and stop() on the handle run() returns stops it cleanly.

Why
a stream is a long-running worker like any other: it has to survive restarts, scale out, and stop during a deploy without losing its place.

Push, consume, ack, replay, fetch, transactions, KV, timers, locks, streams and ephemeral queues all combine. Build on them and Queen carries the logic you would otherwise write around a broker. Patterns shows them combined, one real case each.

See also

Edit this page
Navigation

Type to search…

↑↓ navigate↵ selectEsc close