| 31 | } |
| 32 | |
| 33 | run(): void { |
| 34 | const updatedValues = new Map<Hash, [Multiplicity, T]>() |
| 35 | |
| 36 | // Compute the new multiplicity for each value |
| 37 | for (const message of this.inputMessages()) { |
| 38 | for (const [value, diff] of message.getInner()) { |
| 39 | const hashedValue = hash(this.#by(value)) |
| 40 | |
| 41 | const oldMultiplicity = |
| 42 | updatedValues.get(hashedValue)?.[0] ?? |
| 43 | this.#values.get(hashedValue) ?? |
| 44 | 0 |
| 45 | const newMultiplicity = oldMultiplicity + diff |
| 46 | updatedValues.set(hashedValue, [newMultiplicity, value]) |
| 47 | } |
| 48 | } |
| 49 | |
| 50 | const result: Array<[KeyValue<number, GetValue<T>>, number]> = [] |
| 51 | |
| 52 | // Check which values became visible or disappeared |
| 53 | for (const [ |
| 54 | hashedValue, |
| 55 | [newMultiplicity, value], |
| 56 | ] of updatedValues.entries()) { |
| 57 | const oldMultiplicity = this.#values.get(hashedValue) ?? 0 |
| 58 | |
| 59 | if (newMultiplicity === 0) { |
| 60 | this.#values.delete(hashedValue) |
| 61 | } else { |
| 62 | this.#values.set(hashedValue, newMultiplicity) |
| 63 | } |
| 64 | |
| 65 | if (oldMultiplicity <= 0 && newMultiplicity > 0) { |
| 66 | // The value wasn't present in the stream |
| 67 | // but with this change it is now present in the stream |
| 68 | result.push([[hash(this.#by(value)), value[1]], 1]) |
| 69 | } else if (oldMultiplicity > 0 && newMultiplicity <= 0) { |
| 70 | // The value was present in the stream |
| 71 | // but with this change it is no longer present in the stream |
| 72 | result.push([[hash(this.#by(value)), value[1]], -1]) |
| 73 | } |
| 74 | } |
| 75 | |
| 76 | if (result.length > 0) { |
| 77 | this.output.sendData(new MultiSet(result)) |
| 78 | } |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | /** |