Observables puzzle

puzzle.ts

import { combineLatest, from, Observable, Unsubscribable } from 'rxjs'
import { map, switchMap } from 'rxjs/operators'
import { Location } from 'vscode-languageserver-types'
import { ReferenceParams, TextDocumentPositionParams, TextDocumentRegistrationOptions } from '../../protocol'
import { compact, flatten } from '../../util'
import { FeatureProviderRegistry } from './registry'
import { flattenAndCompact } from './util'

export function getLocationsWithExtensionID<
    P extends TextDocumentPositionParams = TextDocumentPositionParams,
    L extends Location = Location
>(
    providersWithID: Observable<{ extensionID: string; provider: ProvideTextDocumentLocationSignature<P, L> }[]>,
    params: P
): Observable<{ extensionID: string; location: L }[]> {
    return providersWithID.pipe(
        switchMap(providersWithID =>
            combineLatest(
                providersWithID.map(({ provider, extensionID }) =>
                    provider(params).pipe(
                        map(r => flattenAndCompactNonNull([r]).map(l => ({ extensionID, location: l })))
                    )
                )
            ).pipe(map(flattenAndCompactNonNull))
        )
    )
}
      
export function getLocationsWithExtensionIDPromise<
    P extends TextDocumentPositionParams = TextDocumentPositionParams,
    L extends Location = Location
>(
    providersWithID: Observable<{ extensionID: string; provider: ProvideTextDocumentLocationSignature<P, L> }[]>,
    params: P
): Observable<{ extensionID: string; location: L }[]> {
    return providersWithID.pipe(
        switchMap(async providersWithID => {
            const resultsByProvider = await Promise.all(
                providersWithID.map(async ({ provider, extensionID }) => {
                    const providerRes = await provider(params).toPromise()

                    console.log('### extensionID', extensionID, 'providerRes: ', providerRes)

                    return flattenAndCompactNonNull([providerRes]).map(loc => ({
                        extensionID,
                        location: loc,
                    }))
                })
            )
            const flattenedResults: { extensionID: string; location: L }[] = []
            for (const providerResults of resultsByProvider) {
                for (const res of providerResults) {
                    flattenedResults.push(res)
                }
            }
            return flattenedResults
        })
    )
}


/** Flattens and compacts the argument. If it is null or if the result is empty, it returns null. */
function flattenAndCompactNonNull<T>(value: (T | T[] | null)[] | null): T[] {
    return value ? flatten(compact(value)) : []
}

test.ts

import * as assert from 'assert'
import { of } from 'rxjs'
import { TestScheduler } from 'rxjs/testing'
import { Location, Position, Range } from 'vscode-languageserver-types'
import {
    getLocation,
    getLocations,
    getLocationsWithExtensionID,
    ProvideTextDocumentLocationSignature,
} from './location'
import { FIXTURE } from './registry.test'

const scheduler = () => new TestScheduler((a, b) => assert.deepStrictEqual(a, b))

const FIXTURE_LOCATION: Location = {
    uri: 'file:///f',
    range: Range.create(Position.create(1, 2), Position.create(3, 4)),
}
const FIXTURE_LOCATIONS: Location | Location[] | null = [FIXTURE_LOCATION, FIXTURE_LOCATION]

describe('getLocationsWithExtensionID', () => {
    it('wraps single result in array', () =>
        scheduler().run(({ cold, expectObservable }) => {
            const res = getLocationsWithExtensionID(
                cold<{ extensionID: string; provider: ProvideTextDocumentLocationSignature }[]>('-a-|', {
                    a: [{ extensionID: 'test', provider: () => of(FIXTURE_LOCATION) }],
                }),
                FIXTURE.TextDocumentPositionParams
            )
            expectObservable(res).toBe('-a-|', {
                a: [{ extensionID: 'test', location: FIXTURE_LOCATION }],
            })
        }))
})

                  
describe('getLocationsWithExtensionIDPromise', () => {
    it('wraps single result in array', () =>
        scheduler().run(({ cold, expectObservable }) => {
            const res = getLocationsWithExtensionIDPromise(
                cold<{ extensionID: string; provider: ProvideTextDocumentLocationSignature }[]>('-a-|', {
                    a: [{ extensionID: 'test', provider: () => of(FIXTURE_LOCATION) }],
                }),
                FIXTURE.TextDocumentPositionParams
            )
            expectObservable(res).toBe('', {})
        }))
})

The two methods getLocationsWithExtensionID and getLocationsWithExtensionIDPromise appear equivalent. Indeed, if you invoke both , the returned Observable from each resolves to the same value. However, the Observables are not actually equivalent, as shown by the two test functions:

• getLocationsWithExtensionID returns a value we'd expect
• getLocationsWithExtensionIDPromise returns what looks like an Observable that never emits.

Why does the use of Promises result in an Observable that is incompatible with expectObservable(...).toBe(...)? Is it because Promise resolution is non-deterministic (or at least does not conform to the RxJS Observables scheduler)?

This is yet another friction point in using Observables with Promises. What's considered best practice here? I suppose the test function fur getLocationsWithExtensionIDPromise could just be rewritten to manually examine what the Observable emits, but is that best practice? Is it sub-optimal to insert Promises into Observables when we don't need to (as suggested in Option 4 of this post), because we lose out on some precision for examining the results?

Rx Observables have the concept of a "scheduler". When subscribing to an Observable, you can pass a scheduler object, and there are different implementations of it (synchronous, async, next tick, etc). You usually never need to know about this because Rx picks the right scheduler automatically.

A special scheduler is the test scheduler, which allows to pretend events are async, but actually run them all synchronously. This allows writing tests for asynchronous events that still run fast.

However, Promises don't have the concept of a scheduler, and they are not lazy. So if a Promise is used somewhere in the Observable chain, there is no way for the test scheduler to force it to be synchronous.

This doesn't mean the Observable isn't testable, it's just not testable with the test scheduler. Depending on what you actually care about, there are different ways to test. For example, if you don't care about the timing between events, you can just .pipe(toArray()).toPromise() and await the result, then do a synchronous assertion on the array you received. Example for this approach: https://github.com/sourcegraph/javascript-typescript-langserver/blob/82d81f8359968062f35823c0518c5b103c2048bc/src/test/typescript-service-helpers.ts#L2805-L2815

If you do care about timing, you can subscribe to the Observable with a sinon spy, then trigger events and after each step assert the spy was called with the value you expected. Examples for this approach: https://github.com/felixfbecker/abortable-rx/blob/master/src/index.test.ts