Help us shape the future of Go! Take the 2026 Go Developer Survey

eventorder

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 28, 2026 License: MIT Imports: 6 Imported by: 0

README

event-order logo

event-order

Event-time ordering, safe commit offsets, duplicate windows and aligned watermarks for Go streams.
Zero dependencies, bounded memory, one import: no cluster to run.

The gaps | Try it | At a glance | Why not heap or Flink | Proof | Roadmap

CI Coverage OpenSSF Best Practices Go Reference Go version

Events arrive shuffled and are released in event-time order

Status: v0.x, the API may change between minor versions. Last updated: 2026-10-28. Public, no release tagged yet.

Contents

About

Brokers, retrying producers, sensors and mobile clients deliver events late, shuffled and sometimes twice. Code that trusts arrival order computes the wrong thing: a reading applied before an older one, a total counted twice, an offset committed past events that were never processed.

event-order is the in-process answer. Add events as they arrive and get back the ones that are now safe to process, sorted by event time. Around that core it solves three problems that sorting alone leaves open, each tested against a reference model that shares no code with it.

The three gaps

Sorting events is the easy part. These are the parts that are easy to get wrong, and what this library does about each.

A1. Offsets you can commit while events are held

Safe commit offset 3 against the naive commit of 8

A consumer that holds events to reorder them cannot commit the offset of the last message it fetched, because a crash then loses everything still held.

// Before: commit what was fetched. A crash loses every held event.
b.Add(msg.Time, msg)
commit(msg.Partition, msg.Offset+1)

// After: commit only up to the lowest event not yet processed.
for _, r := range s.Ingest(msg.Partition, msg.Offset, msg.Time, msg.ID, msg.Reading) {
    process(r.Value)
    s.Done(msg.Partition, r.Offset)
}
if off, ok := s.Commit(msg.Partition); ok {
    commit(msg.Partition, off)
}

The Kafka example simulates a crash at a random point in a shuffled three-partition stream and restarts from the committed offsets: across 200 random crashes no reading is lost, and the consumer that commits the offset of every fetched message loses readings in all 200.

A2. Duplicates from at-least-once sources

A repeated id inside the window is dropped

Brokers redeliver. Dedup drops repeated ids inside a bounded window of event time, so memory stays near the window size instead of growing with the stream.

// Before: a redelivered message is processed twice.
// After: the repeat never reaches the buffer.
d := eventorder.NewDedup[string](time.Minute, 100_000)
if !d.Seen(msg.Time, msg.ID) {
    b.Add(msg.Time, msg.Reading)
}

A3. One watermark over several sources

Aligned watermark at the slowest active source

Release by the fastest partition's clock and the slower partitions' events become late. Aligned holds the watermark at the minimum over the sources, leaves out a source that went quiet, and Lag names the one holding it back.

// Before: each source advances the shared buffer, so a slow source's events are dropped as late.
// After: the buffer advances only to the minimum of the source watermarks.
al := eventorder.NewAligned[string](eventorder.WithSourceIdleTimeout(30 * time.Second))
al.Register("p0")
al.Register("p1")
if wm, advanced, ok := al.Observe(src, sourceWatermark); ok && advanced {
    for _, ev := range merged.AdvanceTo(wm) {
        process(ev.Value)
    }
}

Stream joins A1 and A2 with per-key ordering in one type; Aligned feeds the downstream buffer that orders the merged stream. A property test runs three partitions through both into one Buffer and checks global order with no late events.

Try it

b := eventorder.New[Reading](
    eventorder.WithMaxDelay(5*time.Second), // how far out of order events may arrive
    eventorder.WithMaxBuffered(10_000),     // memory bound
)
for msg := range incoming {
    for _, ev := range b.Add(msg.Time, msg.Reading) {
        process(ev.Value) // arrives sorted by event time
    }
}
for _, ev := range b.Flush() { // end of input
    process(ev.Value)
}

Install with go get github.com/A-RYAN-1/event-order (Go 1.23 or newer). The import path ends in event-order; the package is eventorder. To see it work first, clone the repository and run go run ./examples/sensors. Real output, 2026-10-27:

Terminal output of the sensors example

Six readings arrive out of order and the retry is too late. Change WithMaxDelay(3*time.Second) to 10*time.Second and the late one is in order.

At a glance

You have Use
One stream Buffer[T]: Add returns what is ready, AdvanceTo moves the watermark from a timer, Flush empties it
Many streams (devices, users, partitions) Keyed[K, T]: one buffer per key, Flush(key) and AdvanceAll forget finished keys
A channel Pipe: channel in, ordered channel out, with a context and WithIdleTimeout for quiet sources
Several goroutines Safe, SafeKeyed: the same with a mutex
A Kafka or MQTT consumer Stream: duplicates, per-key ordering and safe commit offsets in one type
One watermark over several sources Aligned: minimum of the source watermarks, idle sources excluded, Lag names the holder
Several ordered sources Merge over iter.Seq; Reorder wraps one iter.Seq
A broker that needs commits Tracker, KeyedTracker: the offset that is safe to commit while events are held, Revoke for rebalances
An at-least-once source Dedup: drops repeated ids inside a bounded window of event time
A log or CSV in a shell the event-order command, streaming (see Command line)

Features

  • Event-time ordering with a bounded delay. Buffer releases events once the watermark passes them, sorted.
    • The watermark is the largest event time seen minus the delay, inclusive of events at exactly that time.
    • Equal times keep arrival order; an event at or before the watermark when it arrives is late.
    • WithMaxBuffered caps memory: the oldest events are released early rather than the buffer growing.
    • WithLatePolicy either drops and counts late events (Drop) or emits them flagged Late (Emit); OnLate sees every one.
    • WithDelayStrategy(PercentileDelay(0.99, 1024)) lets the delay follow the disorder actually seen.
  • Per-key buffers (Keyed) that forget idle keys, and goroutine-safe wrappers (Safe, SafeKeyed, Pipe).
  • Safe commit offsets (Tracker, KeyedTracker) with Revoke for rebalances.
  • Bounded duplicate window (Dedup).
  • Aligned watermark (Aligned) over several sources, with idle handling and a lag report.
  • Consumer glue (Stream): Ingest, Done, Commit, Skip, Revoke, Flush, AdvanceAll, Watermark, Stats.
  • A command line that sorts a JSON-lines log or CSV as a stream.
  • Examples as their own modules for Kafka, MQTT and Watermill, so the library stays dependency-free.

The package guide on pkg.go.dev (doc.go) says how to choose a type and a delay.

Why it exists

Ordering by event time is a solved idea in stream processors, but a Go service that reads a Kafka topic or an MQTT feed has no small piece to do it with, and no piece at all for the commit-offset and duplicate problems that appear the moment events are held. Building it taught that the ordering is the smallest part: the work is in the boundary cases and in proving them against a model.

How it works

How the pieces fit

The same flow as a diagram GitHub can inspect:

flowchart LR
    S1[Source 1] --> D[Dedup]
    S2[Source 2] --> D
    D --> K[Keyed Buffer per key]
    K --> P[Process in event order]
    P --> T[KeyedTracker Done]
    T --> C[Commit lowest offset not done]
    K -- per-key watermark --> A[Aligned minimum]
    A -- AdvanceTo --> M[Merged Buffer]

Design decisions, each with its price:

  • Hold in memory, track offsets separately. Held events are lost on a crash, so Tracker keeps the commit position behind them and the broker replays. The price is replay after a crash, not loss.
  • Bounded everything. The buffer has a cap, Dedup a window and a capacity. The price is that an extreme burst releases early and a very late duplicate is not caught.
  • Event time trusts the producer clock. The library does not repair skew; use a sequence number where you control the producer.
  • Examples in separate modules. A broker client never enters go.mod. The price is a second go test to run.

Kafka, MQTT and Watermill

Each is its own Go module under examples/, and each takes a small interface or a channel so its tests run without a broker. CI also runs the Kafka and MQTT tests against real brokers (Apache Kafka 3.9 and Eclipse Mosquitto 2 in Docker).

  • examples/kafka (segmentio/kafka-go): per-partition ordering, Dedup, and commits only up to the lowest event not yet processed, with an idle drain so the tail of a quiet topic is not stranded.
  • examples/mqtt (eclipse/paho.mqtt.golang): a subscriber that orders readings with Pipe and releases the tail when the broker goes quiet.
  • examples/watermill (ThreeDotsLabs/watermill): a router handler that holds messages and publishes them in event-time order, plus an idle drain for the tail of a quiet topic. Limit: Watermill acknowledges a message when the handler returns, so a message being held is lost on a crash; where that matters use Stream with the broker client, as in the Kafka example.

Rebalances: kafka-go can deliver messages fetched before a rebalance after it (its issue 1398 describes it, still open when this was written). Tracker cannot fix that, but it keeps your commit position correct while you handle it: flush the partition with Keyed.Flush, mark the flushed events Done, then KeyedTracker.Revoke.

container/heap. You can write the ordering part yourself in about forty lines, and a hand-rolled container/heap loop is not slower in a way you will notice. An earlier hand-rolled loop measured within noise of this library (163 to 429 ns against 192 ns, 2026-10-24, not kept in the repo, so no current figure is claimed). Buffer.Add costs 105 to 131 ns per event with 0 allocations (see Proof). Speed is not the reason to use this library.

The reason is the decisions around the heap that are easy to get subtly wrong: the exact lateness boundary; ties kept in arrival order; the memory cap never exceeded; no event lost or invented; per-key buffers that forget idle keys; a quiet source that does not hold its tail forever; a commit offset that never passes a held event; a duplicate window that stays bounded. Each is tested here against a reference implementation. If you only need to sort a slice you already have, use slices.SortFunc.

Flink, Beam, Kafka Streams. They solve watermarks well, with state, windows and exactly-once sinks, and they are the right tool when you already run a cluster or a JVM stack. A Go service that only needs "these events, in the order they happened, with a safe commit" should not need one. This library is the single watermark idea in-process, with no checkpointing: it does not give exactly-once, and it does not replace a stream engine.

Engineering quality and proof

Buffer.Add 131 ns against 1493 ns for sorting on every add

Measured on Windows, amd64, Go 1.27.0, i5-13450HX: coverage and cost rows on 2026-10-27, the other rows on 2026-10-24. CI runs the same tests with -race on Linux, plus Windows and macOS.

Check Result
Statement coverage 97.4 percent on 2026-10-27 (go test -cover .); CI fails under 95 percent
Reference-model property tests 3,000 random operation sequences for Buffer, 1,000 for Keyed, 2,000 each for Tracker and Dedup, output identical to a sort-every-step reference, call for call
Crash safety 300 random runs of Buffer plus Tracker: the commit offset never passes an unprocessed event
Mutation check removing the boundary nanosecond, the arrival tie-break, the head advance of Tracker, the expiry bound or the cap of Dedup each fails tests
Fuzz FuzzKeyed 857,827 executions, FuzzMerge 45,326, FuzzTracker 3,852,990, FuzzDedup 3,684,370, no failure
Memory about 100 bytes per held event; Len never above the cap over 200,000 adds; Dedup holds about window-many ids
Cost per call median of 10 runs, 0 allocations each: Buffer.Add 105 ns (jitter 16) and 131 ns (jitter 256); Aligned.Observe 109 ns; Stream.Ingest 230 ns; Tracker Track plus Done plus Commit 50 ns; Dedup.Seen 237 ns. Run-to-run spread is 1 to 5 percent for the first four rows and 31 to 75 percent for Tracker and Dedup on this laptop, so read those two as rough
Against sort-every-add 131 ns against 1,493 ns at a disorder window of 256 (3 allocations per add for the baseline)
Kafka crash restart 200 random crashes, no reading lost; the commit-every-fetched consumer loses readings in all 200
Combined proof three partitions through Stream and Aligned into one Buffer: global order, no late events (property test); FuzzAligned and FuzzStream run in CI

Reproduce:

go test -count=1 -cover ./...                        # tests and coverage
scripts/bench.sh                                     # -count 10 plus benchstat
go test -run xxx -fuzz FuzzTracker -fuzztime 20s     # any Fuzz target

scripts/bench.sh needs go install golang.org/x/perf/cmd/benchstat@latest. Save the raw runs with scripts/bench.sh before.txt to compare two commits with benchstat before.txt after.txt.

Limits

  • Events are held in memory only: a crash loses what is held. Keep the broker offset (Tracker) and replay.
  • Ordering by event time trusts the producers' clocks. If clocks drift, use a sequence number where you control the producer; this library does not repair skew.
  • Dedup is in memory and windowed: a duplicate after its window, or after a restart, is not caught. Keep processing idempotent where a repeat costs money.
  • Buffer and Keyed are not goroutine safe by design; use Safe, SafeKeyed or Pipe.
  • No exactly-once delivery and no checkpointing; this is not a stream engine.
  • v0.x: until v1.0.0 a minor version may change the API.

Searched 2026-10-24 on pkg.go.dev and GitHub. An honest map of what else exists:

Project What it is How it differs
jamesrom/order/chansort sliding-window channel sorter orders on a wall-clock window, no watermark or late policy, last updated 2022
teivah/resequencer resequencer sequence numbers, not event time; no license, last pushed 2022
reugn/go-streams stream processing framework windows over a pipeline, not a standalone reorder stage
eKuiper, StreamSQL stream engines with a watermark inside engines you run, not a library you import
Flink, Beam, Kafka Streams full stream processors JVM or a cluster; this is the one watermark idea in-process

Command line

go install github.com/A-RYAN-1/event-order/cmd/event-order@latest
tail -f app.log | event-order -field ts -delay 5s          # JSON lines, "ts" key
event-order -col 1 -sep , -delay 1m events.csv > sorted.csv  # CSV, first column

Times may be RFC 3339, Unix seconds or Unix milliseconds. Late lines are dropped, printed in place (-late emit) or sent to stderr (-late stderr); -max-buffered N caps memory; -stats prints counters. Measured 2026-10-23 on a 95 MB, 2,000,000-line JSON log shuffled out of order (-delay 1h): output fully sorted, none lost, 4.2 s, 13 MB peak memory.

Run it locally

Needs Go 1.23 or newer (pinned in go.mod).

git clone https://github.com/A-RYAN-1/event-order.git
cd event-order
go test -count=1 ./...
go run ./examples/sensors
cd examples/kafka && go test ./...    # examples are separate modules

Ready check: go test -count=1 ./... ends with ok. Before a push run gofmt -l . (it must print nothing), go vet ./... and go test -race ./... (the race detector needs cgo; where it is missing, CI runs it).

Environment variables

None. The library makes no network calls.

Deployment

Not deployed yet: it is a library, so there is nothing to host. It is published as a tagged Go module and pkg.go.dev renders it from the tag. CI runs on Linux, Windows and macOS, and a release workflow builds the command on a tag.

Status and roadmap

Public since 2026-10-28, v0.x (no release tagged yet), last checked 2026-10-28. Aligned, Stream, real-broker tests, the Watermill example and the benchmark script are merged; the changelog carries an untagged 0.2.0 entry.

Known limitations: see Limits.

Planned:

  • govulncheck and a coverage floor in CI; CodeQL and OpenSSF Scorecard are in place and start when the repository is public
  • Codecov (Go Report Card shut down in 2026; golangci-lint in CI covers the same checks)
  • OpenSSF Best Practices badge (passing level, 2026-10-28)
  • Bots, releases and repository settings for the public flip (labels, stale, DCO, GoReleaser, scripts/apply-settings.sh)
  • Community files: support, governance, architecture, FAQ
  • v0.3 contributor issues (milestone v0.3, open to everyone): Watermill middleware, Redis Streams example, OpenTelemetry adapter in a separate module, the A4 sequencer (ordering by sequence number), more benchmarks, docs
  • First tagged release and pkg.go.dev page (the maintainer decides when)

Contributing

Open an issue first for anything beyond a typo or an obvious bug. Tests are required, the module stays dependency-free, and every change goes through a pull request the maintainer approves. See CONTRIBUTING.md for the checks, the commit style and the benchmark routine, and SECURITY.md to report a problem.

License

MIT, see LICENSE. Contributions are licensed under the same terms; contributors keep their own copyright.

Credits

Designed and written by Aryan Deshmukh (A-RYAN-1).

Documentation

Overview

Package eventorder puts out-of-order events back in event-time order.

Events from sensors, message brokers and mobile clients arrive late and shuffled. A Buffer holds them until a watermark says no earlier event is expected, then releases them sorted by event time. The watermark is the largest event time seen so far minus a bounded delay, the same model stream processors use, without a stream processor.

Choosing a type

  • Buffer orders one stream. Add returns what is ready, AdvanceTo moves the watermark from a timer, Flush empties it at shutdown.
  • Keyed orders many streams (devices, users, partitions) independently, so one slow key never holds back the others.
  • Pipe wraps a channel and, with WithIdleTimeout, releases the tail when the source goes quiet.
  • Safe and SafeKeyed add a mutex for several goroutines.
  • Reorder wraps an iter.Seq of events, and Merge joins several ordered sequences into one.
  • Tracker and KeyedTracker say which broker offset is safe to commit while events are still held, see Committing offsets.
  • Dedup drops repeated events from an at-least-once source, see Duplicates.
  • Stream wires Keyed, KeyedTracker and Dedup for a broker consumer in the right order, see Consumer glue.
  • Aligned turns the watermarks of several sources into one safe watermark, see Several sources.

Buffer and Keyed are not safe for concurrent use; serialise calls with a mutex or a single goroutine, or use Safe, SafeKeyed or Pipe.

Choosing a delay

WithMaxDelay is how far out of order an event may arrive and still be placed correctly. Measure the largest gap between an event's time and the newest time seen when it arrives, and set the delay a little above it. A larger delay tolerates more disorder and holds events longer; a smaller one releases sooner and turns more events into late ones. WithMaxBuffered bounds memory when the delay is generous: the oldest events are released early and the watermark moves up to them.

When the disorder changes over time, WithDelayStrategy with PercentileDelay follows it: the delay becomes a percentile of the lags seen in the last window events, and WithMaxDelay stays as the floor. A rare very late event is then late instead of stretching the delay for everyone.

Late events

An event whose time is at or before the watermark when it arrives is late. The default policy drops it and counts it in Stats.Late; Emit returns it at once with Late set. OnLate sees every late event whatever the policy, for a dead-letter queue or a metric.

Committing offsets

A broker commit means everything before an offset is done. Events held in a Buffer are not done, so committing the newest offset seen loses them if the process crashes. Call Tracker.Track when an event enters the buffer and Tracker.Done when it has been processed or dropped as late (OnLate), and commit what Tracker.Commit returns: the lowest offset not yet done. After a restart the consumer re-reads from there, so nothing is lost; events released before the crash but after the commit are read again, so processing should tolerate a repeat. With one Keyed buffer and one KeyedTracker per partition, a rebalance revoke is Keyed.Flush(partition), Done for each flushed event, then KeyedTracker.Revoke(partition) for the offset to commit.

Duplicates

At-least-once delivery repeats events, often out of order. Dedup.Seen(t, id) reports whether id was seen within a window of event time, and bounds its memory with the window and an optional cap on ids. Call it before Buffer.Add with a window at least as long as the delay plus the longest retry. It lives in memory only: after a restart the earlier ids are gone.

Consumer glue

Using Dedup, Keyed and KeyedTracker together has an order that is easy to get wrong: track the offset first, drop duplicates, finish late events, mark released events done only after processing. Stream does that. Per message call Ingest, process the Records it returns, call Done for each, and commit Commit(key). Revoke(key, fn) handles a lost partition and Flush(fn) a clean shutdown. Watermark(key) gives the value to feed to an Aligned when the partitions are merged into one ordered stream.

Several sources

Ordering a merged stream by the fastest source's clock makes every event of a slower source late. Aligned keeps one watermark per source and releases by their minimum: report each source's watermark with Observe (for example its Buffer.Watermark) and pass the combined value to the downstream Buffer.AdvanceTo. Register every source up front so the others wait for it; a source that Observe meets first is not waited for. A silent source holds everything back, so mark it Idle or set WithSourceIdleTimeout; Lag names the source that is holding the watermark. The combined watermark never moves back, so a source that rejoins behind it only has late events of its own.

Compared with a stream processor

Flink and similar systems offer watermarks, allowed lateness and state backends for whole pipelines. This package is the ordering step alone, in memory, in one process. It does not persist what it holds: a crash loses the buffered events, so keep the broker offset and replay after a restart.

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func Merge 

func Merge[T any](sources ...iter.Seq[Event[T]]) iter.Seq[Event[T]]

Merge combines sources that are each already in event-time order into one sequence in event-time order, for example several partitions or devices after Reorder. Events with equal times come out in source order, then in the order within the source. The sources are read one event at a time, so memory is one event per source. Stopping the loop early stops reading every source.

A source that is not itself in order is not fixed: its events are passed along in the order they come, which can leave the output out of order. Run such a source through Reorder first.

Example 
package main

import (
	"fmt"
	"slices"
	"time"

	eventorder "github.com/A-RYAN-1/event-order"
)

func main() {
	at := func(s int, v string) eventorder.Event[string] {
		return eventorder.Event[string]{Time: time.Unix(int64(s), 0), Value: v}
	}
	north := slices.Values([]eventorder.Event[string]{at(1, "n1"), at(4, "n4")})
	south := slices.Values([]eventorder.Event[string]{at(2, "s2"), at(3, "s3")})
	for e := range eventorder.Merge(north, south) {
		fmt.Println(e.Value)
	}
}

Output:
n1
s2
s3
n4

func PercentileDelay 

func PercentileDelay(q float64, window int) func() DelayStrategy

PercentileDelay returns a strategy factory whose delay is the q-th quantile (0 to 1, for example 0.99) of the lags of the last window events. It adapts to a source whose disorder changes over time while ignoring rare outliers: at q = 0.99 about one event in a hundred may arrive later than the delay and be treated as late. Before any event is seen the delay is zero. Observe does no allocation once the window is filled; the window is clamped to at least 1 and q to [0, 1].

Example 
package main

import (
	"fmt"
	"time"

	eventorder "github.com/A-RYAN-1/event-order"
)

func main() {
	b := eventorder.New[string](eventorder.WithDelayStrategy(eventorder.PercentileDelay(0.99, 1000)))
	base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
	for i, s := range []string{"a", "b", "c"} {
		for _, e := range b.Add(base.Add(time.Duration(i)*time.Second), s) {
			fmt.Println(e.Value)
		}
	}
	// With no disorder seen the delay is zero, so each event is held only
	// until the next newer one arrives.

func Pipe 

func Pipe[T any](ctx context.Context, in <-chan Event[T], opts ...Option) <-chan Event[T]

Pipe reads events in arrival order from in and sends them in event-time order on the returned channel. When in is closed, what is still held is flushed in order and the output closes. When ctx is cancelled the output closes at once and held events are discarded. With WithIdleTimeout, a source that stays quiet for that long has everything it holds released.

The caller must drain the returned channel or cancel ctx; the goroutine stops when either happens.

Example 
package main

import (
	"context"
	"fmt"
	"time"

	eventorder "github.com/A-RYAN-1/event-order"
)

func main() {
	t0 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	in := make(chan eventorder.Event[string])
	out := eventorder.Pipe(context.Background(), in, eventorder.WithMaxDelay(5*time.Second))
	go func() {
		defer close(in)
		for _, r := range []struct {
			sec int
			val string
		}{{3, "c"}, {1, "a"}, {2, "b"}} {
			in <- eventorder.Event[string]{Time: t0.Add(time.Duration(r.sec) * time.Second), Value: r.val}
		}
	}()
	for ev := range out {
		fmt.Println(ev.Value)
	}
}

Output:
a
b
c

func Reorder 

func Reorder[T any](seq iter.Seq[Event[T]], opts ...Option) iter.Seq[Event[T]]

Reorder wraps a sequence of events in arrival order and yields them in event-time order, holding at most what opts allow. When the input ends, the rest is flushed in order. Stopping the loop early stops reading the input.

Types

type Aligned 

type Aligned[S comparable] struct {
	// contains filtered or unexported fields
}

Aligned combines the watermarks of several sources into one. Events from partitions, devices or upstream services arrive on separate streams, each with its own progress. Releasing events by the fastest stream's clock makes the slower streams' events late, so the safe watermark is the minimum over the sources: nothing earlier is expected from any of them.

Report each source's watermark with Observe, for example the value of that source's Buffer.Watermark, and feed the result to the Buffer that orders the merged stream with AdvanceTo. A source that falls silent would hold the minimum forever, so it can be marked Idle by hand, or automatically with WithSourceIdleTimeout, and is left out until it reports again. Lag shows which source is holding the watermark back.

The combined watermark never moves backwards. A source that rejoins behind it, or is added behind it, is not waited for: its events older than the combined watermark are late in the downstream Buffer, as for any late event. A source that has been registered but has not reported yet blocks the watermark until it reports, goes idle or is removed.

Aligned is safe for concurrent use.

func NewAligned 

func NewAligned[S comparable](opts ...AlignedOption) *Aligned[S]

NewAligned returns an Aligned with no sources.

func (*Aligned[S]) Idle 

func (a *Aligned[S]) Idle(src S) (combined time.Time, advanced, ok bool)

Idle leaves src out of the minimum until it next reports with Observe. It registers src if it is unknown. Returns as Observe does: leaving out a slow source can advance the combined watermark.

func (*Aligned[S]) Lag 

func (a *Aligned[S]) Lag() []SourceStatus[S]

Lag reports every source, the ones holding the watermark back first, then by watermark from lowest to highest.

func (*Aligned[S]) Len 

func (a *Aligned[S]) Len() int

Len is the number of sources known, active or not.

func (*Aligned[S]) Observe 

func (a *Aligned[S]) Observe(src S, wm time.Time) (combined time.Time, advanced, ok bool)

Observe records watermark wm for src and returns the combined watermark. advanced is true when the combined watermark moved up because of this call; then pass the returned value to Buffer.AdvanceTo. ok is false while no combined watermark exists yet. A watermark lower than the source's previous one is ignored, and a report from an idle source makes it active again.

func (*Aligned[S]) Register 

func (a *Aligned[S]) Register(src S)

Register declares a source before it reports, so the combined watermark waits for it. Observe registers a source on first use, which is too late to make the others wait. Registering a known source does nothing.

func (*Aligned[S]) Remove 

func (a *Aligned[S]) Remove(src S) (combined time.Time, advanced, ok bool)

Remove forgets src, for a partition that was revoked or a device that left. Returns as Observe does.

func (*Aligned[S]) Watermark 

func (a *Aligned[S]) Watermark() (time.Time, bool)

Watermark returns the combined watermark. Idle timeouts are evaluated here, so a call after a quiet period can advance it.

type AlignedOption 

type AlignedOption func(*alignedOptions)

AlignedOption configures an Aligned.

func WithAlignedClock 

func WithAlignedClock(now func() time.Time) AlignedOption

WithAlignedClock replaces time.Now as the wall clock used for the idle timeout, for tests.

func WithSourceIdleTimeout 

func WithSourceIdleTimeout(d time.Duration) AlignedOption

WithSourceIdleTimeout leaves a source out of the minimum once it has not reported for d of wall-clock time, so a stalled source does not hold the watermark. It counts from the source's last Observe, or from Register for a source that never reported. The default, zero, never times out.

type Buffer 

type Buffer[T any] struct {
	// contains filtered or unexported fields
}

Buffer reorders events by event time.

Example 
package main

import (
	"fmt"
	"time"

	eventorder "github.com/A-RYAN-1/event-order"
)

func main() {
	t0 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	b := eventorder.New[string](eventorder.WithMaxDelay(5 * time.Second))

	// Readings arrive shuffled; event times are 3s, 1s, 2s, then a much later one.
	for _, r := range []struct {
		sec int
		val string
	}{{3, "c"}, {1, "a"}, {2, "b"}, {20, "z"}} {
		for _, ev := range b.Add(t0.Add(time.Duration(r.sec)*time.Second), r.val) {
			fmt.Println(ev.Value)
		}
	}
}

Output:
a
b
c

func New 

func New[T any](opts ...Option) *Buffer[T]

New returns an empty Buffer.

func (*Buffer[T]) Add 

func (b *Buffer[T]) Add(t time.Time, v T) []Event[T]

Add inserts an event and returns the events that are now safe to process, in event-time order. The slice is empty when nothing can be released yet. Under the Emit late policy a late event is returned first with Late set.

func (*Buffer[T]) AddTo 

func (b *Buffer[T]) AddTo(dst []Event[T], t time.Time, v T) []Event[T]

AddTo is Add appending to dst, for a hot path that reuses one slice: 

out = b.AddTo(out[:0], t, v)

Once the slice has grown to the largest burst, Add costs no allocation.

func (*Buffer[T]) AdvanceTo 

func (b *Buffer[T]) AdvanceTo(t time.Time) []Event[T]

AdvanceTo moves the watermark up to t without an event, for example from a heartbeat or a timer when a source goes quiet. It never moves backwards. It returns the events released by the move.

func (*Buffer[T]) Flush 

func (b *Buffer[T]) Flush() []Event[T]

Flush releases everything still held, in order, for shutdown or end of input. The watermark moves to the last released event.

func (*Buffer[T]) Len 

func (b *Buffer[T]) Len() int

Len is the number of events currently held.

func (*Buffer[T]) OnLate 

func (b *Buffer[T]) OnLate(fn func(Event[T]))

OnLate sets a function called for every late event, whatever the policy, so dropped events can go to a dead-letter queue or a metric instead of vanishing. It runs inside Add; keep it fast. A nil fn clears it.

Example 
package main

import (
	"fmt"
	"time"

	eventorder "github.com/A-RYAN-1/event-order"
)

func main() {
	t0 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	b := eventorder.New[string]()
	b.OnLate(func(e eventorder.Event[string]) { fmt.Println("dead letter:", e.Value) })
	b.Add(t0.Add(10*time.Second), "now")
	b.Add(t0, "stale")
}

Output:
dead letter: stale

func (*Buffer[T]) Stats 

func (b *Buffer[T]) Stats() Stats

Stats returns the current counters.

func (*Buffer[T]) Watermark 

func (b *Buffer[T]) Watermark() time.Time

Watermark is the time up to which events have been released; the zero time until the first event or AdvanceTo.

type Dedup 

type Dedup[K comparable] struct {
	// contains filtered or unexported fields
}

Dedup drops repeated events from an at-least-once source. Brokers and retrying producers deliver the same event twice, often out of order, so a duplicate can arrive well after the original. Dedup remembers event ids for a window of event time and answers whether an id was already seen.

The window is measured against the newest event time seen, the same clock a Buffer uses: an id is forgotten once it is more than window behind. Set the window at least as large as the Buffer's delay plus how late a retry can arrive. A duplicate that arrives after its id was forgotten is not caught, and the set lives in memory only, so after a restart earlier ids are gone: keep the processing itself idempotent where a duplicate costs money.

Dedup is safe for concurrent use.

Example

A retrying producer sends some events twice. Dedup drops the second copy before it reaches the Buffer. 

package main

import (
	"fmt"
	"time"

	eventorder "github.com/A-RYAN-1/event-order"
)

func main() {
	t0 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	d := eventorder.NewDedup[string](time.Minute, 0)
	b := eventorder.New[string](eventorder.WithMaxDelay(5 * time.Second))

	for _, m := range []struct {
		id  string
		sec int
	}{{"a", 1}, {"b", 2}, {"a", 1}, {"c", 3}} {
		ts := t0.Add(time.Duration(m.sec) * time.Second)
		if d.Seen(ts, m.id) {
			continue
		}
		b.Add(ts, m.id)
	}
	for _, ev := range b.Flush() {
		fmt.Println(ev.Value)
	}
	fmt.Println("duplicates dropped:", d.Duplicates())
}

Output:
a
b
c
duplicates dropped: 1

func NewDedup 

func NewDedup[K comparable](window time.Duration, maxIDs int) *Dedup[K]

NewDedup returns a Dedup that remembers ids for window of event time. When maxIDs is above zero and more ids are held, the oldest are forgotten first, which bounds memory at the price of missing older duplicates.

func (*Dedup[K]) Duplicates 

func (d *Dedup[K]) Duplicates() uint64

Duplicates is the number of duplicates found so far.

func (*Dedup[K]) Len 

func (d *Dedup[K]) Len() int

Len is the number of ids currently remembered.

func (*Dedup[K]) Seen 

func (d *Dedup[K]) Seen(t time.Time, id K) bool

Seen reports whether id, an event at time t, is a duplicate of an event remembered earlier. A first sight returns false and is remembered. An event already older than the window is not remembered and returns false: it is too old to tell.

type DelayStrategy 

type DelayStrategy interface {
	Observe(lag time.Duration)
	Delay() time.Duration
}

DelayStrategy chooses the delay of a Buffer from the disorder it has seen, instead of a fixed WithMaxDelay. Observe is called for every event added, late ones included, with how far behind the newest event time it arrived (zero for an event that is the newest so far). Delay is read after each Observe and is the watermark delay to use from now on. A strategy belongs to one Buffer and is never called from two goroutines at once.

type Event 

type Event[T any] struct {
	Time  time.Time
	Value T
	// Late is true when the event arrived after the watermark had passed its
	// time. It is only ever set on events released under LatePolicy Emit.
	Late bool
}

Event is a value with the time it happened (not the time it arrived).

type Keyed 

type Keyed[K comparable, T any] struct {
	// contains filtered or unexported fields
}

Keyed keeps one Buffer per key, so events of different devices, users or partitions are reordered independently and one slow key does not hold back the others. Not safe for concurrent use.

Example 
package main

import (
	"fmt"
	"time"

	eventorder "github.com/A-RYAN-1/event-order"
)

func main() {
	t0 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	k := eventorder.NewKeyed[string, int](eventorder.WithMaxDelay(time.Second))
	k.Add("sensor-1", t0.Add(2*time.Second), 20)
	k.Add("sensor-1", t0.Add(1*time.Second), 10)
	k.Add("sensor-2", t0, 99) // its own clock: not held back by sensor-1
	k.FlushAll(func(key string, evs []eventorder.Event[int]) {
		if key == "sensor-1" {
			for _, e := range evs {
				fmt.Println(key, e.Value)
			}
		}
	})
}

Output:
sensor-1 10
sensor-1 20

func NewKeyed 

func NewKeyed[K comparable, T any](opts ...Option) *Keyed[K, T]

NewKeyed returns an empty Keyed; every key's Buffer is made with opts.

func (*Keyed[K, T]) Add 

func (k *Keyed[K, T]) Add(key K, t time.Time, v T) []Event[T]

Add inserts an event for key and returns that key's newly released events.

func (*Keyed[K, T]) AdvanceAll 

func (k *Keyed[K, T]) AdvanceAll(t time.Time, fn func(key K, events []Event[T]))

AdvanceAll moves every key watermark up to t, see Buffer.AdvanceTo, calling fn once per key that released events. Keys left empty are forgotten, so a periodic call with the current time keeps the number of tracked keys bounded by the keys that were active recently. Key order is not defined.

func (*Keyed[K, T]) AdvanceTo 

func (k *Keyed[K, T]) AdvanceTo(key K, t time.Time) []Event[T]

AdvanceTo moves one key's watermark, see Buffer.AdvanceTo. An unknown key releases nothing.

func (*Keyed[K, T]) Flush 

func (k *Keyed[K, T]) Flush(key K) []Event[T]

Flush releases and forgets one key. Call it when a key is finished, or the map of buffers grows with every key ever seen.

func (*Keyed[K, T]) FlushAll 

func (k *Keyed[K, T]) FlushAll(fn func(key K, events []Event[T]))

FlushAll releases every key, calling fn once per key with its events in order, and forgets them all. Key order is not defined.

func (*Keyed[K, T]) Len 

func (k *Keyed[K, T]) Len() int

Len is the number of keys currently tracked.

func (*Keyed[K, T]) OnLate 

func (k *Keyed[K, T]) OnLate(fn func(key K, e Event[T]))

OnLate sets a function called with the key for every late event, see Buffer.OnLate. It applies to existing and future keys. A nil fn clears it.

func (*Keyed[K, T]) Stats 

func (k *Keyed[K, T]) Stats(key K) (s Stats, ok bool)

Stats returns the counters of one key; ok is false for an unknown key.

type KeyedTracker 

type KeyedTracker[K comparable] struct {
	// contains filtered or unexported fields
}

KeyedTracker keeps one Tracker per key, usually per partition. Pair it with Keyed: Keyed.Flush(partition) on a revoke releases that partition's held events, process them and call Done, then Revoke returns the offset to commit before the partition goes to another consumer. Safe for concurrent use.

func NewKeyedTracker 

func NewKeyedTracker[K comparable]() *KeyedTracker[K]

NewKeyedTracker returns an empty KeyedTracker.

func (*KeyedTracker[K]) Commit 

func (k *KeyedTracker[K]) Commit(key K) (offset int64, ok bool)

Commit is Tracker.Commit for one key; ok is false for an unknown key.

func (*KeyedTracker[K]) Done 

func (k *KeyedTracker[K]) Done(key K, offset int64) bool

Done is Tracker.Done for one key; false for an unknown key.

func (*KeyedTracker[K]) Len 

func (k *KeyedTracker[K]) Len() int

Len is the number of keys currently tracked.

func (*KeyedTracker[K]) Revoke 

func (k *KeyedTracker[K]) Revoke(key K) (offset int64, ok bool)

Revoke forgets a key and returns the offset to commit for it, as Commit would have. Events still pending are not done, so the offset stops at the first of them and the next owner of the partition will read them again.

func (*KeyedTracker[K]) Track 

func (k *KeyedTracker[K]) Track(key K, offset int64) bool

Track is Tracker.Track for one key.

type LatePolicy 

type LatePolicy int

LatePolicy says what happens to an event that arrives after the watermark has passed its event time. 

const (
	// Drop discards late events and counts them in Stats.Late.
	Drop LatePolicy = iota
	// Emit returns late events immediately, flagged Late, ahead of the
	// in-order events released by the same call.
	Emit
)

type Option 

type Option func(*options)

Option configures a Buffer.

func WithDelayStrategy 

func WithDelayStrategy(newStrategy func() DelayStrategy) Option

WithDelayStrategy lets a DelayStrategy set the delay. newStrategy is called once per Buffer, so Keyed gives every key its own state. WithMaxDelay, when also given, is the floor: the delay used is the larger of the two. A nil newStrategy, or one that returns nil, leaves the fixed delay in place.

func WithIdleTimeout 

func WithIdleTimeout(d time.Duration) Option

WithIdleTimeout makes Pipe release everything it holds when no event has arrived for d of wall-clock time, so a source that goes quiet does not strand its last events. Zero, the default, waits for input or Flush. Buffer and Keyed ignore it: they have no clock, drive them with AdvanceTo.

func WithLatePolicy 

func WithLatePolicy(p LatePolicy) Option

WithLatePolicy chooses what happens to late events. The default is Drop.

func WithMaxBuffered 

func WithMaxBuffered(n int) Option

WithMaxBuffered caps how many events are held. When the cap is exceeded the oldest events are released early and the watermark moves up to them, so memory stays bounded at the price of more late events. Zero means no cap.

func WithMaxDelay 

func WithMaxDelay(d time.Duration) Option

WithMaxDelay sets how far out of order events may arrive. Larger values tolerate more disorder and hold events longer. The default is zero: only events with the same or later time than everything seen are in order.

type Record 

type Record[T any] struct {
	Time   time.Time
	Offset int64
	Value  T
}

Record is an event as Stream hands it back: the event time, the broker offset it came from and the value.

type Safe 

type Safe[T any] struct {
	// contains filtered or unexported fields
}

Safe is a Buffer that can be used from several goroutines. Calls are serialised by a mutex; the OnLate function runs while it is held, so it must not call back into the same Safe.

func NewSafe 

func NewSafe[T any](opts ...Option) *Safe[T]

NewSafe returns an empty Safe, configured like New.

func (*Safe[T]) Add 

func (s *Safe[T]) Add(t time.Time, v T) []Event[T]

Add is Buffer.Add under the lock.

func (*Safe[T]) AddTo 

func (s *Safe[T]) AddTo(dst []Event[T], t time.Time, v T) []Event[T]

AddTo is Buffer.AddTo under the lock.

func (*Safe[T]) AdvanceTo 

func (s *Safe[T]) AdvanceTo(t time.Time) []Event[T]

AdvanceTo is Buffer.AdvanceTo under the lock, for a timer goroutine.

func (*Safe[T]) Flush 

func (s *Safe[T]) Flush() []Event[T]

Flush is Buffer.Flush under the lock.

func (*Safe[T]) OnLate 

func (s *Safe[T]) OnLate(fn func(Event[T]))

OnLate is Buffer.OnLate under the lock.

func (*Safe[T]) Stats 

func (s *Safe[T]) Stats() Stats

Stats is Buffer.Stats under the lock.

type SafeKeyed 

type SafeKeyed[K comparable, T any] struct {
	// contains filtered or unexported fields
}

SafeKeyed is a Keyed that can be used from several goroutines.

func NewSafeKeyed 

func NewSafeKeyed[K comparable, T any](opts ...Option) *SafeKeyed[K, T]

NewSafeKeyed returns an empty SafeKeyed, configured like NewKeyed.

func (*SafeKeyed[K, T]) Add 

func (s *SafeKeyed[K, T]) Add(key K, t time.Time, v T) []Event[T]

Add is Keyed.Add under the lock.

func (*SafeKeyed[K, T]) AdvanceAll 

func (s *SafeKeyed[K, T]) AdvanceAll(t time.Time, fn func(key K, events []Event[T]))

AdvanceAll is Keyed.AdvanceAll under the lock; fn runs while it is held.

func (*SafeKeyed[K, T]) AdvanceTo 

func (s *SafeKeyed[K, T]) AdvanceTo(key K, t time.Time) []Event[T]

AdvanceTo is Keyed.AdvanceTo under the lock.

func (*SafeKeyed[K, T]) Flush 

func (s *SafeKeyed[K, T]) Flush(key K) []Event[T]

Flush is Keyed.Flush under the lock.

func (*SafeKeyed[K, T]) FlushAll 

func (s *SafeKeyed[K, T]) FlushAll(fn func(key K, events []Event[T]))

FlushAll is Keyed.FlushAll under the lock; fn runs while it is held.

func (*SafeKeyed[K, T]) Len 

func (s *SafeKeyed[K, T]) Len() int

Len is Keyed.Len under the lock.

func (*SafeKeyed[K, T]) OnLate 

func (s *SafeKeyed[K, T]) OnLate(fn func(key K, e Event[T]))

OnLate is Keyed.OnLate under the lock.

func (*SafeKeyed[K, T]) Stats 

func (s *SafeKeyed[K, T]) Stats(key K) (Stats, bool)

Stats is Keyed.Stats under the lock.

type SourceStatus 

type SourceStatus[S comparable] struct {
	Source S
	// Watermark is the source's latest reported watermark; zero when Reported
	// is false.
	Watermark time.Time
	Reported  bool
	// Behind is how far the source's watermark is behind the newest watermark
	// among the active sources.
	Behind time.Duration
	// Idle is true when the source is left out of the minimum, by hand or by
	// timeout.
	Idle bool
	// Holding is true when the source is what keeps the combined watermark
	// from advancing: active and either not yet reported or alone at the
	// minimum behind some other active source.
	Holding bool
}

SourceStatus is one source's state in the report from Lag.

type Stats 

type Stats struct {
	Buffered  int       // events currently held
	Released  uint64    // events released in order
	Late      uint64    // events that arrived late (dropped or emitted)
	Peak      int       // most events held at once, to size WithMaxBuffered
	Watermark time.Time // zero until the first event or AdvanceTo
}

Stats is a snapshot of a Buffer's counters.

type Stream 

type Stream[K comparable, T any] struct {
	// contains filtered or unexported fields
}

Stream wires the pieces a broker consumer needs into one place, in the order that is correct: duplicates are dropped first, the rest are held per key in a Keyed buffer, and a KeyedTracker follows which offsets are finished so that Commit never passes an event that is still held.

The flow for each message is Ingest, process the Records it returns, Done for each of them, and commit what Commit returns. A message that is a duplicate, late (older than the key's watermark) or has an offset the key already saw is finished by Stream itself; it never comes back as a Record.

Key is usually the partition. On a rebalance call Revoke for each lost partition, and on shutdown call Flush. Stream is safe for concurrent use, so worker goroutines may call Done and Commit.

func NewStream 

func NewStream[K comparable, T any](opts ...StreamOption) *Stream[K, T]

NewStream returns a Stream. The late hook, if any, is set with OnLate.

func (*Stream[K, T]) AdvanceAll 

func (s *Stream[K, T]) AdvanceAll(t time.Time, fn func(key K, records []Record[T]))

AdvanceAll releases what is due at time t in every key, see Keyed.AdvanceAll, calling fn once per key with its records. Call it periodically so quiet keys drain and are forgotten.

func (*Stream[K, T]) Commit 

func (s *Stream[K, T]) Commit(key K) (offset int64, ok bool)

Commit is the offset to commit for key: everything before it is processed. ok is false for a key with no messages yet.

func (*Stream[K, T]) Done 

func (s *Stream[K, T]) Done(key K, offset int64) bool

Done marks the message at offset as processed. It reports false for an offset that is not pending.

func (*Stream[K, T]) Flush 

func (s *Stream[K, T]) Flush(fn func(key K, r Record[T]))

Flush is for a clean shutdown. It releases everything still held, passes each record to fn and marks it done; Commit then returns the final offsets.

func (*Stream[K, T]) Ingest 

func (s *Stream[K, T]) Ingest(key K, offset int64, t time.Time, id string, v T) []Record[T]

Ingest takes one message: key and offset say where it came from, t is its event time, id its unique id for deduplication (empty to skip) and v the value. It returns the records that became ready, in event-time order for the key. Call Done for each once processed.

func (*Stream[K, T]) OnLate 

func (s *Stream[K, T]) OnLate(fn func(key K, r Record[T]))

OnLate sets a function called for every late message, for a dead-letter queue or a metric. The offset is already finished when it runs. A nil fn clears it. The function must not call back into the Stream.

func (*Stream[K, T]) Revoke 

func (s *Stream[K, T]) Revoke(key K, fn func(Record[T])) (offset int64, ok bool)

Revoke is for a partition about to be given up. It releases what key still holds, passes each record to fn (in order) and marks it done, forgets the key and returns the offset to commit. The key starts clean if it comes back.

func (*Stream[K, T]) Skip 

func (s *Stream[K, T]) Skip(key K, offset int64) bool

Skip records a message that is finished without being processed, for example a payload that cannot be decoded, so the commit position can move past it. It reports false for an offset not above the last one seen for the key.

func (*Stream[K, T]) Stats 

func (s *Stream[K, T]) Stats() StreamStats

Stats returns the counters.

func (*Stream[K, T]) Watermark 

func (s *Stream[K, T]) Watermark(key K) (time.Time, bool)

Watermark is key's watermark, the time up to which its events are released, for feeding an Aligned. ok is false for a key that holds nothing.

type StreamOption 

type StreamOption func(*streamConfig)

StreamOption configures a Stream.

func WithBuffer 

func WithBuffer(opts ...Option) StreamOption

WithBuffer sets the options of every key's Buffer: WithMaxDelay, WithMaxBuffered, WithDelayStrategy. WithLatePolicy is ignored, because Stream always drops late events and reports them to WithOnLate.

func WithDedup 

func WithDedup(window time.Duration, maxIDs int) StreamOption

WithDedup drops repeated ids, see NewDedup. Without it, ids are not looked at.

type StreamStats 

type StreamStats struct {
	// Duplicates are messages dropped because their id was seen in the window.
	Duplicates uint64
	// Late are messages dropped because they arrived at or before the
	// watermark of their key.
	Late uint64
	// Skipped are messages whose offset was not above the last one seen for
	// their key, for example a replay after a rebalance that was not Revoked.
	Skipped uint64
	// Keys is the number of keys currently held.
	Keys int
}

StreamStats are the counters of a Stream.

type Tracker 

type Tracker struct {
	// contains filtered or unexported fields
}

Tracker answers one question for a consumer that holds events in a Buffer before processing them: which broker offset is safe to commit?

A broker commit means "everything before this offset is done". Events held for reordering are not done, so committing the highest offset seen would lose them if the process crashes. Call Track(offset) when an event enters the buffer, Done(offset) when it has been processed (or dropped as late), and commit what Commit returns. Offsets may finish in any order; the commit position only moves past a gap when the gap is done.

A Tracker is for one partition, whose offsets are tracked in increasing order. It is safe for concurrent use, so workers may call Done.

Example

A consumer that holds events for reordering must not commit the offset of events still in the buffer. The Tracker gives the offset that is safe. 

package main

import (
	"fmt"
	"time"

	eventorder "github.com/A-RYAN-1/event-order"
)

func main() {
	t0 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	b := eventorder.New[int64](eventorder.WithMaxDelay(2 * time.Second))
	tr := eventorder.NewTracker()

	// Offsets 0, 1, 2 arrive; their event times are 3s, 1s, 2s.
	for offset, sec := range []int{3, 1, 2} {
		tr.Track(int64(offset))
		for _, ev := range b.Add(t0.Add(time.Duration(sec)*time.Second), int64(offset)) {
			tr.Done(ev.Value) // processed
		}
	}
	next, _ := tr.Commit()
	fmt.Println("safe to commit:", next, "pending:", tr.Pending())

	for _, ev := range b.Flush() { // shutdown: process the rest
		tr.Done(ev.Value)
	}
	next, _ = tr.Commit()
	fmt.Println("after flush:", next)
}

Output:
safe to commit: 0 pending: 3
after flush: 3

func NewTracker 

func NewTracker() *Tracker

NewTracker returns an empty Tracker.

func (*Tracker) Commit 

func (t *Tracker) Commit() (offset int64, ok bool)

Commit returns the offset to commit: the lowest tracked offset that is not done, or one past the last tracked offset when everything is done. ok is false until the first Track.

func (*Tracker) Done 

func (t *Tracker) Done(offset int64) bool

Done records that the event at offset has been processed or discarded. It reports false for an offset that is not tracked or is already done.

func (*Tracker) Pending 

func (t *Tracker) Pending() int

Pending is the number of tracked offsets that are not done.

func (*Tracker) Track 

func (t *Tracker) Track(offset int64) bool

Track records that the event at offset has entered the buffer. Offsets must increase; it reports false, and records nothing, for an offset not greater than the last one tracked.

Source Files

View all Source files

Directories

Path Synopsis
cmd
event-order command
Command event-order sorts a stream of timestamped lines into event-time order using a bounded delay, without reading the whole input into memory.
 Command event-order sorts a stream of timestamped lines into event-time order using a bounded delay, without reading the whole input into memory.
examples
sensors command
Command sensors feeds shuffled temperature readings into a Buffer and prints them in the order they happened.
 Command sensors feeds shuffled temperature readings into a Buffer and prints them in the order they happened.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.