---
title: "Stream processing: windows, reduce and gates"
description: "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."
---

> 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

# Stream processing: windows, reduce and gates

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.

JavaScript:

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

Python:

```python
from queen import Stream

revenue = await (
    Stream.from_(queen.queue("orders"))           # one window per partition: per customer
    .window_tumbling(seconds=60)
    .aggregate({"count": lambda order: 1, "sum": lambda order: order["total"]})
    .to(queen.queue("revenue-per-minute"))        # one result per customer per minute
    .run(query_id="revenue-per-minute", url="http://localhost:6632")
)
```

Go:

```go
import "github.com/smartpricing/queen/clients/client-go/v2/streams"

revenue, err := streams.
	From(client.Queue("orders").AsStreamSource()). // one window per partition: per customer
	WindowTumbling(60).
	Aggregate(map[string]streams.ExtractorFn{
		"count": func(order interface{}) (float64, error) { return 1, nil },
		"sum": func(order interface{}) (float64, error) {
			return order.(map[string]interface{})["total"].(float64), nil
		},
	}, "count", "sum").
	To(client.Queue("revenue-per-minute")). // one result per customer per minute
	Run(ctx, streams.RunOptions{QueryID: "revenue-per-minute", URL: "http://localhost:6632"})
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
use queen_mq::streams::{RunOptions, Stream};

let revenue = Stream::from(queen.queue("orders")) // one window per partition: per customer
    .window_tumbling(60)
    .aggregate_count("count")
    .aggregate_sum("sum", |order| order.number("total"))
    .to(queen.queue("revenue-per-minute")) // one result per customer per minute
    .run(&queen, RunOptions::new("revenue-per-minute"))
    .await?;
```

PHP: PHP has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

C++: C++ has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

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.

**Figure.** 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. |

**Figure.** 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.

JavaScript:

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

Python:

```python
login_rate = await (
    Stream.from_(queen.queue("logins"))           # one partition per user
    .window_sliding(size=600, slide=60)           # the last 10 minutes, every minute
    .aggregate({"count": lambda login: 1})
    .to(queen.queue("login-rate"))
    .run(query_id="login-rate", url="http://localhost:6632")
)
```

Go:

```go
loginRate, err := streams.
	From(client.Queue("logins").AsStreamSource()). // one partition per user
	WindowSliding(600, 60).                        // the last 10 minutes, every minute
	Aggregate(map[string]streams.ExtractorFn{
		"count": func(login interface{}) (float64, error) { return 1, nil },
	}, "count").
	To(client.Queue("login-rate")).
	Run(ctx, streams.RunOptions{QueryID: "login-rate", URL: "http://localhost:6632"})
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
let login_rate = Stream::from(queen.queue("logins")) // one partition per user
    .window_sliding(600, 60)                // the last 10 minutes, every minute
    .aggregate_count("count")
    .to(queen.queue("login-rate"))
    .run(&queen, RunOptions::new("login-rate"))
    .await?;
```

PHP: PHP has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

C++: C++ has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

JavaScript:

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

Python:

```python
visits = await (
    Stream.from_(queen.queue("clicks"))           # one partition per visitor
    .window_session(gap=1800)                     # a visit ends after 30 minutes of quiet
    .aggregate({"count": lambda click: 1})
    .foreach(lambda visit: print("visit ended", visit))   # acked after this returns
    .run(query_id="visits", url="http://localhost:6632")
)
```

Go:

```go
visits, err := streams.
	From(client.Queue("clicks").AsStreamSource()). // one partition per visitor
	WindowSession(1800).                           // a visit ends after 30 minutes of quiet
	Aggregate(map[string]streams.ExtractorFn{
		"count": func(click interface{}) (float64, error) { return 1, nil },
	}, "count").
	Foreach(func(visit interface{}, _ streams.EmitCtx) error {
		fmt.Println("visit ended", visit)
		return nil // acked after this returns
	}).
	Run(ctx, streams.RunOptions{QueryID: "visits", URL: "http://localhost:6632"})
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
let visits = Stream::from(queen.queue("clicks")) // one partition per visitor
    .window_session(1800)                   // a visit ends after 30 minutes of quiet
    .aggregate_count("count")
    .foreach(|visit, _| async move {
        println!("visit ended {visit}");
        Ok(())
    })                                      // acked after this returns
    .run(&queen, RunOptions::new("visits"))
    .await?;
```

PHP: PHP has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

C++: C++ has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

**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.

JavaScript:

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

Python:

```python
hourly = await (
    Stream.from_(queen.queue("sales"))
    .window_cron(
        every="hour",                  # on the clock: 10:00 to 11:00, 11:00 to 12:00
        event_time=lambda msg: msg["data"]["soldAt"],  # when it happened, not on arrival
        allowed_lateness=120,          # a sale up to 2 minutes late still counts
    )
    .aggregate({
        "count": lambda sale: 1,
        "sum": lambda sale: sale["amount"],
        "max": lambda sale: sale["amount"],
    })
    .to(queen.queue("sales-per-hour"))
    .run(query_id="sales-per-hour", url="http://localhost:6632", subscription_mode="all")
)
```

Go:

```go
hourly, err := streams.
	From(client.Queue("sales").AsStreamSource()).
	WindowCron("hour", // on the clock: 10:00 to 11:00, 11:00 to 12:00
		streams.WithEventTime(func(msg interface{}) (int64, error) {
			// when the sale happened, not when it arrived
			// (eventTime gets the whole message: the sale is its data)
			sale := msg.(map[string]interface{})["data"].(map[string]interface{})
			at, err := time.Parse(time.RFC3339, sale["soldAt"].(string))
			return at.UnixMilli(), err
		}),
		streams.WithAllowedLateness(120), // a sale up to 2 minutes late still counts
	).
	Aggregate(map[string]streams.ExtractorFn{
		"count": func(sale interface{}) (float64, error) { return 1, nil },
		"sum": func(sale interface{}) (float64, error) {
			return sale.(map[string]interface{})["amount"].(float64), nil
		},
		"max": func(sale interface{}) (float64, error) {
			return sale.(map[string]interface{})["amount"].(float64), nil
		},
	}, "count", "sum", "max").
	To(client.Queue("sales-per-hour")).
	Run(ctx, streams.RunOptions{QueryID: "sales-per-hour", URL: "http://localhost:6632",
		SubscriptionMode: queen.SubscriptionModeAll})
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
use queen_mq::streams::ops::parse_iso_ms;
use queen_mq::streams::Every;

let hourly = Stream::from(queen.queue("sales"))
    .window_cron(Every::Hour)               // on the clock: 10:00 to 11:00, 11:00 to 12:00
    // when the sale happened, not when it arrived
    .event_time(|sale| parse_iso_ms(sale.data["soldAt"].as_str()?))
    .allowed_lateness(120)                  // a sale up to 2 minutes late still counts
    .aggregate_count("count")
    .aggregate_sum("sum", |sale| sale.number("amount"))
    .aggregate_max("max", |sale| sale.number("amount"))
    .to(queen.queue("sales-per-hour"))
    .run(&queen, RunOptions::new("sales-per-hour").subscription_mode(SubscriptionMode::All))
    .await?;
```

PHP: PHP has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

C++: C++ has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

`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.

JavaScript:

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

Python:

```python
from queen import token_bucket_gate

limiter = await (
    Stream.from_(queen.queue("outgoing-sms"))     # one partition per customer
    .gate(token_bucket_gate(capacity=10, refill_per_sec=1))  # 10 at once, then 1 a second
    .foreach(send_sms)
    .run(query_id="sms-limiter", url="http://localhost:6632")
)
```

Go:

```go
import "github.com/smartpricing/queen/clients/client-go/v2/streams/helpers"

limiter, err := streams.
	From(client.Queue("outgoing-sms").AsStreamSource()). // one partition per customer
	Gate(helpers.TokenBucketGate(helpers.TokenBucketGateOptions{
		Capacity: 10, RefillPerSec: 1, // 10 at once, then 1 a second
	})).
	Foreach(func(sms interface{}, _ streams.EmitCtx) error { return sendSms(sms) }).
	Run(ctx, streams.RunOptions{QueryID: "sms-limiter", URL: "http://localhost:6632"})
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
use queen_mq::streams::{GateCtx, Record};

// The Rust client has no tokenBucketGate helper; this is the same bucket. Its
// state is kept per partition in the broker, and saved only when a message passes.
fn token_bucket_gate(
    capacity: f64,
    refill_per_sec: f64,
) -> impl Fn(&Record, &mut GateCtx<'_>) -> bool + Send + Sync + 'static {
    move |_msg, ctx| {
        let now = ctx.stream_time_ms as f64;
        let elapsed = ((now - ctx.num("lastRefillAt", now)) / 1000.0).max(0.0);
        let tokens = (ctx.num("tokens", capacity) + elapsed * refill_per_sec).min(capacity);
        ctx.set_num("lastRefillAt", now);
        if tokens < 1.0 {
            return false; // refused: it comes back, in order, when its lease ends
        }
        ctx.set_num("tokens", tokens - 1.0);
        true
    }
}

let limiter = Stream::from(queen.queue("outgoing-sms")) // one partition per customer
    .gate(token_bucket_gate(10.0, 1.0))     // 10 at once, then 1 a second
    .foreach(|sms, _| async move { send_sms(sms).await })
    .run(&queen, RunOptions::new("sms-limiter"))
    .await?;
```

PHP: PHP has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

C++: C++ has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

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](/patterns/) shows them combined, one real case each.

## See also

- [Windowed aggregation](/patterns/windowed-aggregation/): a session window per trace.
- [Rate limiting](/patterns/rate-limiting/): a token bucket gate per customer.

Source: https://queenmq.com/writing-applications/streams/index.mdx