{"version":3,"file":"adaptive-publisher.d.ts","sourceRoot":"","sources":["../../../src/harness/utils/adaptive-publisher.ts"],"names":[],"mappings":"AAAA,MAAM,WAAW,wBAAwB,CAAC,MAAM,EAAE,OAAO;IACxD,QAAQ,IAAI,MAAM,CAAC;IACnB,MAAM,CAAC,QAAQ,EAAE,MAAM,GAAG,SAAS,EAAE,OAAO,EAAE,MAAM,GAAG,OAAO,GAAG,SAAS,CAAC;IAC3E,OAAO,CAAC,MAAM,EAAE,OAAO,GAAG,MAAM,CAAC;IACjC,OAAO,CAAC,MAAM,EAAE,OAAO,GAAG,IAAI,CAAC;IAC/B,OAAO,CAAC,KAAK,EAAE,OAAO,GAAG,IAAI,CAAC;IAC9B,aAAa,CAAC,EAAE,MAAM,CAAC;IACvB,oBAAoB,CAAC,EAAE,MAAM,CAAC;CAC9B;AAED;;;;;;GAMG;AACH,qBAAa,iBAAiB,CAAC,MAAM,EAAE,OAAO;;IAU7C,YAAY,OAAO,EAAE,wBAAwB,CAAC,MAAM,EAAE,OAAO,CAAC,EAI7D;IAED,SAAS,IAAI,IAAI,CAShB;IAED,KAAK,CAAC,KAAK,UAAQ,GAAG,IAAI,CAuBzB;IAED,OAAO,IAAI,IAAI,CAId;CAaD","sourcesContent":["export interface AdaptivePublisherOptions<TValue, TUpdate> {\n\tsnapshot(): TValue;\n\tupdate(previous: TValue | undefined, current: TValue): TUpdate | undefined;\n\tmeasure(update: TUpdate): number;\n\tpublish(update: TUpdate): void;\n\tonError(error: unknown): void;\n\tminIntervalMs?: number;\n\ttargetBytesPerSecond?: number;\n}\n\n/**\n * Publishes the latest state without queuing intermediate mutations.\n *\n * The first dirty state after idle is immediate. Each publication then buys a\n * delay proportional to its encoded size, with a minimum interval that also\n * bounds event count. A single trailing timer guarantees eventual publication.\n */\nexport class AdaptivePublisher<TValue, TUpdate> {\n\treadonly #options: AdaptivePublisherOptions<TValue, TUpdate>;\n\treadonly #minIntervalMs: number;\n\treadonly #targetBytesPerSecond: number;\n\t#published: TValue | undefined;\n\t#dirty = false;\n\t#nextEmitAt = 0;\n\t#timer: ReturnType<typeof setTimeout> | undefined;\n\t#disposed = false;\n\n\tconstructor(options: AdaptivePublisherOptions<TValue, TUpdate>) {\n\t\tthis.#options = options;\n\t\tthis.#minIntervalMs = options.minIntervalMs ?? 100;\n\t\tthis.#targetBytesPerSecond = options.targetBytesPerSecond ?? 100 * 1024;\n\t}\n\n\tmarkDirty(): void {\n\t\tif (this.#disposed) return;\n\t\tthis.#dirty = true;\n\t\tconst wait = this.#nextEmitAt - Date.now();\n\t\tif (wait <= 0) {\n\t\t\tthis.flush();\n\t\t\treturn;\n\t\t}\n\t\tthis.#armTimer(wait);\n\t}\n\n\tflush(force = false): void {\n\t\tif (this.#disposed || !this.#dirty) return;\n\t\tconst now = Date.now();\n\t\tif (!force && now < this.#nextEmitAt) {\n\t\t\tthis.#armTimer(this.#nextEmitAt - now);\n\t\t\treturn;\n\t\t}\n\t\tif (this.#timer !== undefined) clearTimeout(this.#timer);\n\t\tthis.#timer = undefined;\n\t\tconst current = this.#options.snapshot();\n\t\tconst update = this.#options.update(this.#published, current);\n\t\tif (update === undefined) {\n\t\t\tthis.#published = current;\n\t\t\tthis.#dirty = false;\n\t\t\treturn;\n\t\t}\n\t\tconst encodedBytes = this.#options.measure(update);\n\t\tthis.#published = current;\n\t\tthis.#dirty = false;\n\t\tthis.#nextEmitAt = now + Math.max(this.#minIntervalMs, (encodedBytes * 1000) / this.#targetBytesPerSecond);\n\t\t// Commit before delivery. A consumer may apply the update and then throw or\n\t\t// reenter the producer; retaining the old baseline would duplicate that delta.\n\t\tthis.#options.publish(update);\n\t}\n\n\tdispose(): void {\n\t\tif (this.#timer !== undefined) clearTimeout(this.#timer);\n\t\tthis.#timer = undefined;\n\t\tthis.#disposed = true;\n\t}\n\n\t#armTimer(wait: number): void {\n\t\tif (this.#timer !== undefined) return;\n\t\tthis.#timer = setTimeout(() => {\n\t\t\tthis.#timer = undefined;\n\t\t\ttry {\n\t\t\t\tthis.flush();\n\t\t\t} catch (error) {\n\t\t\t\tthis.#options.onError(error);\n\t\t\t}\n\t\t}, wait);\n\t}\n}\n"]}