rxjs-mergemap-array

rxjs-mergemap-array provides a single RxJS operator, mergeMapArray, for a narrow but recurring reactive-programming problem: you have an observable that emits arrays whose elements can be added, removed, or reordered over time, and you want to run a projection (an inner observable) for each unique element exactly once, keeping it alive for as long as that element remains in the array — regardless of its position.

Unlike a plain mergeMap over a flattened stream, it deduplicates by a configurable equality function, tracks additions and removals via an internal scan-based diff, and reassembles the output array in the same order as the latest input array, only emitting once every currently-present element has produced at least one value (similar to combineLatest semantics). This makes it useful for cases like subscribing to a live list of IDs and doing continuous per-ID work (e.g. fetching or watching a record) without restarting that work when the ID’s position in the list changes, and without leaking subscriptions when an ID drops out of the list.

What You Get

• mergeMapArray operator - a single exported OperatorFunction<T[], R[]> that projects each unique array element to an inner observable and emits the projected array in input order.
• Configurable equality - an optional isEqual comparator argument lets you key by identity, a field, or any custom equality instead of relying on strict ===.
• Automatic subscription lifecycle - inner observables for an element are unsubscribed via takeUntil as soon as that element is removed from the input array, avoiding leaked work or leaked subscriptions.
• Order-preserving, combineLatest-style emission - output only emits once every currently-present element has emitted at least once, and reorders the output array to match new input order without re-running finished projections.
• Dual ESM/CJS build with types - published via @sanity/pkg-utils as dist/index.js (ESM) and dist/index.cjs (CJS) with bundled .d.ts declarations, rxjs@7.x as the only peer dependency and zero runtime dependencies.

Common Use Cases

• Live list of watched records - subscribing to a stream of database record IDs and running a continuous per-ID subscription (e.g. a realtime document watch) that survives reordering of the ID list.
• Per-item async enrichment - given an array of items that changes over time, run an async lookup (HTTP call, cache read) once per unique item and keep the enriched array in sync as items are added or removed.
• Deduplicated background work over a changing set - starting a long-lived task (e.g. a WebSocket subscription) for each unique key in a set that changes membership, stopping the task only when the key actually leaves the set.
• Reactive UI state built from an ordered ID list - maintaining a derived, ordered array of computed values that tracks a source array of IDs, useful for list-backed UI state built entirely with RxJS rather than a framework’s own reconciliation.

Under The Hood

Architecture The entire implementation lives in one file, src/mergeMapArray.ts (re-exported via src/index.ts), and takes the form of a single closure-returning operator function rather than a class or layered module — appropriate for its narrow scope. Internally it composes RxJS primitives directly: a first scan over the shared, share()-multicasted input computes a State<T> of {current, added, removed} by diffing each new array against the previous one (using the caller-supplied isEqual), a second mergeMap+scan pipeline turns added/removed events into per-element inner subscriptions (unsubscribed via takeUntil when the matching removed event fires), and a final scan reassembles the ordered output array by walking the latest input array and looking up each element’s last-known emitted value. There is no dependency injection or configuration layering; the only inputs are the project and isEqual functions passed by the caller, so changing the core diff/reassembly scan logic would directly break both change-detection and the reordering guarantee that is the operator’s main value proposition.

Tech Stack Written in TypeScript, published as an ESM-first ("type": "module") package with a generated CJS build, built via @sanity/pkg-utils’s pkg build --strict --clean --check and typed via tsconfig.dist.json. It has zero runtime dependencies and declares rxjs@7.x only as a peerDependency, keeping the published bundle minimal. Linting uses ESLint with @typescript-eslint and eslint-plugin-simple-import-sort; formatting uses Prettier via the shared @sanity/prettier-config; lint-staged runs formatting on commit. No CI workflow file was present in the shallow clone of the repository root, so automated enforcement of lint/test/build on push could not be confirmed directly.

Code Quality A single test file, src/mergeMapArray.test.ts, exercises the operator with Vitest using toMatchInlineSnapshot assertions across several scenarios: empty-array input, order-preserving emission, unsubscription on removal, and duplicate elements within the same input array — each verified against precise, timed sequences built with Subject, delay, and concat. Naming is clear and intention-revealing (added, removed, mapped, uniqueBy), and the public API is fully generic and typed (mergeMapArray<T, R>). There is no explicit error-handling path — the operator assumes well-formed observable and array input, and the test suite does not cover malformed-input or error-propagation scenarios, so quality here rests on type safety and behavioral tests rather than defensive coding.

What Makes It Unique The operator addresses a fairly specific gap in RxJS’s built-in operator set: none of mergeMap, switchMap, combineLatest, or groupBy alone give you “one live subscription per unique array element, kept alive across reorders, emitted back out in current array order.” mergeMapArray combines diffing (similar in spirit to keyed list-reconciliation used in virtual-DOM libraries), deduplication-by-equality, and combineLatest-style gating into a single operator built entirely from stock RxJS primitives with no extra runtime dependency. It is not a novel algorithm so much as a well-scoped, correctly-handled composition of an underused pattern, published as a minimal, dependency-free utility rather than folded into a larger library.