Class LiveAggregation<S, Out>

constructor

• new LiveAggregation<
      S extends SeriesSchema,
      Out extends SeriesSchema = SeriesSchema,
  >(
      source: LiveSource<S>,
      sequence: Sequence,
      mapping:
          | Readonly<
              Record<
                  string,
                  | AggregateReducer
                  | Readonly<
                      {
                          from: ValueColumnsForSchema<S>[number]["name"];
                          kind?: ScalarKind | undefined;
                          using: AggregateFunctionsForKind<
                              Extract<
                                  ValueColumnsForSchema<(...)>[number],
                                  ColumnDef<(...)[(...)], ScalarKind>,
                              >["kind"],
                          >;
                      },
                  >,
              >,
          >
          | Readonly<
              Record<
                  string,
                  | AggregateReducer
                  | Readonly<
                      {
                          from: ValueColumnsForSchema<S>[number]["name"];
                          kind?: ScalarKind | undefined;
                          using: AggregateFunctionsForKind<
                              Extract<
                                  ValueColumnsForSchema<(...)>[number],
                                  ColumnDef<(...)[(...)], ScalarKind>,
                              >["kind"],
                          >;
                      },
                  >,
              >,
          >,
      options?: LiveAggregationOptions,
  ): LiveAggregation<S, Out>

  Type Parameters

  • S extends SeriesSchema
  • Out extends SeriesSchema = SeriesSchema

  Parameters

  • source: LiveSource<S>
  • sequence: Sequence
  • mapping:
        | Readonly<
            Record<
                string,
                | AggregateReducer
                | Readonly<
                    {
                        from: ValueColumnsForSchema<S>[number]["name"];
                        kind?: ScalarKind | undefined;
                        using: AggregateFunctionsForKind<
                            Extract<
                                ValueColumnsForSchema<(...)>[number],
                                ColumnDef<(...)[(...)], ScalarKind>,
                            >["kind"],
                        >;
                    },
                >,
            >,
        >
        | Readonly<
            Record<
                string,
                | AggregateReducer
                | Readonly<
                    {
                        from: ValueColumnsForSchema<S>[number]["name"];
                        kind?: ScalarKind | undefined;
                        using: AggregateFunctionsForKind<
                            Extract<
                                ValueColumnsForSchema<(...)>[number],
                                ColumnDef<(...)[(...)], ScalarKind>,
                            >["kind"],
                        >;
                    },
                >,
            >,
        >
  • Optionaloptions: LiveAggregationOptions

  Returns LiveAggregation<S, Out>

Readonlyname

Readonlyschema

closedCount

• get closedCount(): number

  Returns number

hasOpenBucket

• get hasOpenBucket(): boolean

  Returns boolean

length

• get length(): number

  Returns number

at

• at(index: number): EventForSchema<Out> | undefined

  Parameters

  • index: number

  Returns EventForSchema<Out> | undefined

closed

• closed(): TimeSeries<Out>

  Returns TimeSeries<Out>

cumulative

• cumulative<const Targets extends string>(
      spec: {
          [K in string]:
              | "sum"
              | "min"
              | "max"
              | "count"
              | ((acc: number, value: number) => number)
      },
  ): LiveView<
      readonly [
          Out[0],
          ReplaceSmoothedColumn<ValueColumnsForSchema<Out>, Targets>,
      ],
  >

  Type Parameters

  • const Targets extends string

  Parameters

  • spec: {
        [K in string]:
            | "sum"
            | "min"
            | "max"
            | "count"
            | ((acc: number, value: number) => number)
    }

  Returns LiveView<
      readonly [
          Out[0],
          ReplaceSmoothedColumn<ValueColumnsForSchema<Out>, Targets>,
      ],
  >

diff

• diff<const Target extends string>(
      columns: Target | readonly Target[],
      options?: { drop?: boolean },
  ): LiveView<
      readonly [
          Out[0],
          ReplaceSmoothedColumn<ValueColumnsForSchema<Out>, Target>,
      ],
  >

  Type Parameters

  • const Target extends string

  Parameters

  • columns: Target | readonly Target[]
  • Optionaloptions: { drop?: boolean }

  Returns LiveView<
      readonly [
          Out[0],
          ReplaceSmoothedColumn<ValueColumnsForSchema<Out>, Target>,
      ],
  >

dispose

• dispose(): void

  Returns void

fill

• fill(
      strategy: LiveFillStrategy | LiveFillMapping<Out>,
      options?: { limit?: number },
  ): LiveView<Out>

  Parameters

  • strategy: LiveFillStrategy | LiveFillMapping<Out>
  • Optionaloptions: { limit?: number }

  Returns LiveView<Out>

filter

• filter(predicate: (event: EventForSchema<Out>) => boolean): LiveView<Out>

  Parameters

  • predicate: (event: EventForSchema<Out>) => boolean

  Returns LiveView<Out>

map

• map(fn: (event: EventForSchema<Out>) => EventForSchema<Out>): LiveView<Out>

  Parameters

  • fn: (event: EventForSchema<Out>) => EventForSchema<Out>

  Returns LiveView<Out>

on

• on(type: "event", fn: CloseListener): () => void

  Parameters

  • type: "event"
  • fn: CloseListener

  Returns () => void

• on(type: "bucket", fn: BucketListener): this

  Parameters

  • type: "bucket"
  • fn: BucketListener

  Returns this

• on(type: "close", fn: CloseListener): this

  Parameters

  • type: "close"
  • fn: CloseListener

  Returns this

• on(type: "update", fn: UpdateListener): this

  Parameters

  • type: "update"
  • fn: UpdateListener

  Returns this

pctChange

• pctChange<const Target extends string>(
      columns: Target | readonly Target[],
      options?: { drop?: boolean },
  ): LiveView<
      readonly [
          Out[0],
          ReplaceSmoothedColumn<ValueColumnsForSchema<Out>, Target>,
      ],
  >

  Type Parameters

  • const Target extends string

  Parameters

  • columns: Target | readonly Target[]
  • Optionaloptions: { drop?: boolean }

  Returns LiveView<
      readonly [
          Out[0],
          ReplaceSmoothedColumn<ValueColumnsForSchema<Out>, Target>,
      ],
  >

rate

• rate<const Target extends string>(
      columns: Target | readonly Target[],
      options?: { drop?: boolean },
  ): LiveView<
      readonly [
          Out[0],
          ReplaceSmoothedColumn<ValueColumnsForSchema<Out>, Target>,
      ],
  >

  Type Parameters

  • const Target extends string

  Parameters

  • columns: Target | readonly Target[]
  • Optionaloptions: { drop?: boolean }

  Returns LiveView<
      readonly [
          Out[0],
          ReplaceSmoothedColumn<ValueColumnsForSchema<Out>, Target>,
      ],
  >

rolling

• rolling<
      const M extends
          | Readonly<
              Record<
                  string,
                  | AggregateReducer
                  | Readonly<
                      {
                          from: ValueColumnsForSchema<Out>[number]["name"];
                          kind?: ScalarKind | undefined;
                          using: AggregateFunctionsForKind<
                              Extract<
                                  ValueColumnsForSchema<(...)>[number],
                                  ColumnDef<(...)[(...)], ScalarKind>,
                              >["kind"],
                          >;
                      },
                  >,
              >,
          >
          | {
              readonly [K in string
              | number
              | symbol]: K extends ValueColumnsForSchema<Out>[number]["name"]
                  ?
                      | Readonly<
                          {
                              from: ValueColumnsForSchema<Out>[number]["name"];
                              kind?: ScalarKind;
                              using: AggregateFunctionsForKind<
                                  Extract<
                                      ValueColumnsForSchema<(...)>[number],
                                      ColumnDef<(...)[(...)], ScalarKind>,
                                  >["kind"],
                              >;
                          },
                      >
                      | AggregateFunctionsForKind<
                          ColumnKindByName<ValueColumnsForSchema<Out>, K & string>,
                      >
                  : string extends K
                      ? | AggregateReducer
                      | Readonly<
                          {
                              from: ValueColumnsForSchema<Out>[number]["name"];
                              kind?: ScalarKind | undefined;
                              using: AggregateFunctionsForKind<
                                  Extract<(...)[(...)], ColumnDef<(...), (...)>>["kind"],
                              >;
                          },
                      >
                      : Readonly<
                          {
                              from: ValueColumnsForSchema<Out>[number]["name"];
                              kind?: ScalarKind;
                              using: AggregateFunctionsForKind<
                                  Extract<
                                      ValueColumnsForSchema<(...)>[number],
                                      ColumnDef<(...)[(...)], ScalarKind>,
                                  >["kind"],
                              >;
                          },
                      >
          },
  >(
      windowSize: RollingWindow,
      mapping: M,
      options?: LiveRollingOptions,
  ): LiveRollingAggregation<
      Out,
      readonly [Out[0], AggregateColumns<ValueColumnsForSchema<Out>, M>],
  >

  Type Parameters

  • const M extends
        | Readonly<
            Record<
                string,
                | AggregateReducer
                | Readonly<
                    {
                        from: ValueColumnsForSchema<Out>[number]["name"];
                        kind?: ScalarKind | undefined;
                        using: AggregateFunctionsForKind<
                            Extract<
                                ValueColumnsForSchema<(...)>[number],
                                ColumnDef<(...)[(...)], ScalarKind>,
                            >["kind"],
                        >;
                    },
                >,
            >,
        >
        | {
            readonly [K in string
            | number
            | symbol]: K extends ValueColumnsForSchema<Out>[number]["name"]
                ?
                    | Readonly<
                        {
                            from: ValueColumnsForSchema<Out>[number]["name"];
                            kind?: ScalarKind;
                            using: AggregateFunctionsForKind<
                                Extract<
                                    ValueColumnsForSchema<(...)>[number],
                                    ColumnDef<(...)[(...)], ScalarKind>,
                                >["kind"],
                            >;
                        },
                    >
                    | AggregateFunctionsForKind<
                        ColumnKindByName<ValueColumnsForSchema<Out>, K & string>,
                    >
                : string extends K
                    ? | AggregateReducer
                    | Readonly<
                        {
                            from: ValueColumnsForSchema<Out>[number]["name"];
                            kind?: ScalarKind | undefined;
                            using: AggregateFunctionsForKind<
                                Extract<(...)[(...)], ColumnDef<(...), (...)>>["kind"],
                            >;
                        },
                    >
                    : Readonly<
                        {
                            from: ValueColumnsForSchema<Out>[number]["name"];
                            kind?: ScalarKind;
                            using: AggregateFunctionsForKind<
                                Extract<
                                    ValueColumnsForSchema<(...)>[number],
                                    ColumnDef<(...)[(...)], ScalarKind>,
                                >["kind"],
                            >;
                        },
                    >
        }

  Parameters

  • windowSize: RollingWindow
  • mapping: M
  • Optionaloptions: LiveRollingOptions

  Returns LiveRollingAggregation<
      Out,
      readonly [Out[0], AggregateColumns<ValueColumnsForSchema<Out>, M>],
  >

• rolling<
      const M extends
          Readonly<
              Record<
                  string,
                  | AggregateReducer
                  | Readonly<
                      {
                          from: ValueColumnsForSchema<Out>[number]["name"];
                          kind?: ScalarKind | undefined;
                          using: AggregateFunctionsForKind<
                              Extract<
                                  ValueColumnsForSchema<Out>[number],
                                  ColumnDef<(...)[(...)]["name"], ScalarKind>,
                              >["kind"],
                          >;
                      },
                  >,
              >,
          >,
  >(
      windowSize: RollingWindow,
      mapping: M,
      options?: LiveRollingOptions,
  ): LiveRollingAggregation<
      Out,
      readonly [Out[0], AggregateColumns<ValueColumnsForSchema<Out>, M>],
  >

  Type Parameters

  • const M extends Readonly<
        Record<
            string,
            | AggregateReducer
            | Readonly<
                {
                    from: ValueColumnsForSchema<Out>[number]["name"];
                    kind?: ScalarKind | undefined;
                    using: AggregateFunctionsForKind<
                        Extract<
                            ValueColumnsForSchema<Out>[number],
                            ColumnDef<(...)[(...)]["name"], ScalarKind>,
                        >["kind"],
                    >;
                },
            >,
        >,
    >

  Parameters

  • windowSize: RollingWindow
  • mapping: M
  • Optionaloptions: LiveRollingOptions

  Returns LiveRollingAggregation<
      Out,
      readonly [Out[0], AggregateColumns<ValueColumnsForSchema<Out>, M>],
  >

select

• select<const Keys extends readonly (keyof EventDataForSchema<Out>)[]>(
      ...keys: Keys,
  ): LiveView<
      readonly [
          Out[0],
          PickSelectedColumns<ValueColumnsForSchema<Out>, Keys[number]>,
      ],
  >

  Type Parameters

  • const Keys extends readonly (keyof EventDataForSchema<Out>)[]

  Parameters

  • ...keys: Keys

  Returns LiveView<
      readonly [
          Out[0],
          PickSelectedColumns<ValueColumnsForSchema<Out>, Keys[number]>,
      ],
  >

snapshot

• snapshot(): TimeSeries<Out>

  Returns TimeSeries<Out>

stats

• stats(): {
      bucketsClosed: number;
      eventsObserved: number;
      openBuckets: number;
      openBucketStart?: number;
  }

  Pipeline stats snapshot — cumulative counters since construction plus current bucket-state. Cheap O(1).

  • eventsObserved: total source events that contributed to a bucket. Includes events replayed at construction from a non-empty source. Late events (those whose bucket was already closed under the watermark + grace window) are silently dropped and do NOT increment this counter — match the silent-drop shape in LiveSeries.stats for late-event handling. Never decreases.
  • bucketsClosed: total buckets finalized (= length of the closed-event stream, also exposed via length). Never decreases.
  • openBuckets: current count of pending (unclosed) buckets. Equal to #pending.size. Drops back as the watermark advances and buckets close.
  • openBucketStart: earliest open bucket's start ms, or undefined if no buckets are pending.

  Note on field omission vs LiveRollingAggregation.stats: this shape doesn't carry an emissions field because every closed bucket pushes exactly one event to the output stream — the count would be identical to bucketsClosed. openBuckets and openBucketStart give the bucket-lifecycle observability users actually reach for.

  Returns {
      bucketsClosed: number;
      eventsObserved: number;
      openBuckets: number;
      openBucketStart?: number;
  }

window

• window(size: RollingWindow): LiveView<Out>

  Parameters

  • size: RollingWindow

  Returns LiveView<Out>