| 24 | } |
| 25 | |
| 26 | run(): void { |
| 27 | // Collect all input messages and update the index |
| 28 | const keysTodo = new Set<K>() |
| 29 | for (const message of this.inputMessages()) { |
| 30 | for (const [item, multiplicity] of message.getInner()) { |
| 31 | const [key, value] = item |
| 32 | this.#index.addValue(key, [value, multiplicity]) |
| 33 | keysTodo.add(key) |
| 34 | } |
| 35 | } |
| 36 | |
| 37 | // For each key, compute the reduction and delta |
| 38 | const result: Array<[[K, V2], number]> = [] |
| 39 | for (const key of keysTodo) { |
| 40 | const curr = this.#index.get(key) |
| 41 | const currOut = this.#indexOut.get(key) |
| 42 | const out = this.#f(curr) |
| 43 | |
| 44 | // Create maps for current and previous outputs using values directly as keys |
| 45 | const newOutputMap = new Map<V2, number>() |
| 46 | const oldOutputMap = new Map<V2, number>() |
| 47 | |
| 48 | // Process new output |
| 49 | for (const [value, multiplicity] of out) { |
| 50 | const existing = newOutputMap.get(value) ?? 0 |
| 51 | newOutputMap.set(value, existing + multiplicity) |
| 52 | } |
| 53 | |
| 54 | // Process previous output |
| 55 | for (const [value, multiplicity] of currOut) { |
| 56 | const existing = oldOutputMap.get(value) ?? 0 |
| 57 | oldOutputMap.set(value, existing + multiplicity) |
| 58 | } |
| 59 | |
| 60 | // First, emit removals for old values that are no longer present |
| 61 | for (const [value, multiplicity] of oldOutputMap) { |
| 62 | if (!newOutputMap.has(value)) { |
| 63 | // Remove the old value entirely |
| 64 | result.push([[key, value], -multiplicity]) |
| 65 | this.#indexOut.addValue(key, [value, -multiplicity]) |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | // Then, emit additions for new values that are not present in old |
| 70 | for (const [value, multiplicity] of newOutputMap) { |
| 71 | if (!oldOutputMap.has(value)) { |
| 72 | // Add the new value only if it has non-zero multiplicity |
| 73 | if (multiplicity !== 0) { |
| 74 | result.push([[key, value], multiplicity]) |
| 75 | this.#indexOut.addValue(key, [value, multiplicity]) |
| 76 | } |
| 77 | } |
| 78 | } |
| 79 | |
| 80 | // Finally, emit multiplicity changes for values that were present and are still present |
| 81 | for (const [value, newMultiplicity] of newOutputMap) { |
| 82 | const oldMultiplicity = oldOutputMap.get(value) |
| 83 | if (oldMultiplicity !== undefined) { |