refactor(inquirer): remove rxjs runtime dependency

Author: SBoudrias

packages/inquirer/README.md

@@ -117,7 +117,7 @@ yarn node packages/inquirer/examples/checkbox.js
 
 Launch the prompt interface (inquiry session)
 
-- **questions** (Array) containing [Question Object](#question) (using the [reactive interface](#reactive-interface), you can also pass a `Rx.Observable` instance)
+- **questions** a [Question Object](#question), an array or map of questions, or an RxJS-compatible Observable of questions
 - **answers** (object) contains values of already answered questions. Inquirer will avoid asking answers already provided here. Defaults `{}`.
 - returns a **Promise**
 
@@ -328,9 +328,7 @@ The `postfix` property is useful if you want to provide an extension.
 
 ## Reactive interface
 
-Internally, Inquirer uses the [JS reactive extension](https://github.com/ReactiveX/rxjs) to handle events and async flows.
-
-This mean you can take advantage of this feature to provide more advanced flows. For example, you can dynamically add questions to be asked:
+`inquirer.prompt()` accepts an RxJS-compatible Observable of questions. This supports dynamic flows where questions are emitted over time:
 
 ```js
 const prompts = new Rx.Subject();

packages/inquirer/inquirer.test.ts

@@ -9,7 +9,7 @@ import stream from 'node:stream';
 import tty from 'node:tty';
 import readline from 'node:readline';
 import { vi, expect, beforeEach, afterEach, describe, it, expectTypeOf } from 'vitest';
-import { of } from 'rxjs';
+import { from as observableFrom, map, of, Subject } from 'rxjs';
 import { AbortPromptError, createPrompt } from '@inquirer/core';
 import type { InquirerReadline } from '@inquirer/type';
 import inquirer from './src/index.ts';
@@ -254,6 +254,50 @@ describe('inquirer.prompt(...)', () => {
       expect(answers).toEqual({ q1: true, q2: false });
       expectTypeOf(answers).toEqualTypeOf<{ q1: any; q2: any }>();
     });
+
+    it('takes an Observable that emits questions over time', async () => {
+      const questions = new Subject<Question & { answer: boolean }>();
+      const processEvents: unknown[] = [];
+
+      const promise = inquirer.prompt(questions);
+      promise.ui.process.subscribe((answer) => {
+        processEvents.push(answer);
+      });
+
+      questions.next({
+        type: 'stub',
+        name: 'q1',
+        message: 'message',
+        answer: true,
+      });
+      questions.next({
+        type: 'stub',
+        name: 'q2',
+        message: 'message',
+        answer: false,
+      });
+      questions.complete();
+
+      await expect(promise).resolves.toEqual({ q1: true, q2: false });
+      expect(processEvents).toEqual([
+        { name: 'q1', answer: true },
+        { name: 'q2', answer: false },
+      ]);
+    });
+
+    it('rejects when an Observable question source errors', async () => {
+      const questions = new Subject<Question>();
+      const error = new Error('Question source failed');
+      const onError = vi.fn();
+
+      const promise = inquirer.prompt(questions);
+      promise.ui.process.subscribe({ error: onError });
+
+      questions.error(error);
+
+      await expect(promise).rejects.toBe(error);
+      expect(onError).toHaveBeenCalledWith(error);
+    });
   });
 
   it("should close and create a new readline instances each time it's called", async () => {
@@ -638,6 +682,33 @@ describe('inquirer.prompt(...)', () => {
     expect(spy).toHaveBeenCalledWith({ name: 'name', answer: 'doe' });
   });
 
+  it('should expose an RxJS-compatible Reactive interface', async () => {
+    const processEvents: string[] = [];
+
+    const promise = inquirer.prompt([
+      {
+        type: 'stub',
+        name: 'name1',
+        message: 'message',
+        answer: 'bar',
+      },
+      {
+        type: 'stub',
+        name: 'name',
+        message: 'message',
+        answer: 'doe',
+      },
+    ]);
+    observableFrom(promise.ui.process)
+      .pipe(map(({ name, answer }) => `${name}:${String(answer)}`))
+      .subscribe((answer) => {
+        processEvents.push(answer);
+      });
+
+    await promise;
+    expect(processEvents).toEqual(['name1:bar', 'name:doe']);
+  });
+
   it('should expose the UI', async () => {
     const promise = inquirer.prompt([]);
     expect(promise.ui.answers).toBeTypeOf('object');

packages/inquirer/package.json

@@ -78,12 +78,12 @@
     "@inquirer/prompts": "^8.4.3",
     "@inquirer/type": "^4.0.5",
     "mute-stream": "^4.0.0",
-    "run-async": "^4.0.6",
-    "rxjs": "^7.8.2"
+    "run-async": "^4.0.6"
   },
   "devDependencies": {
     "@repo/tsconfig": "workspace:*",
     "@types/mute-stream": "^0.0.4",
+    "rxjs": "^7.8.2",
     "typescript": "^6.0.2"
   },
   "peerDependencies": {

packages/inquirer/src/types.ts

@@ -12,7 +12,7 @@ import {
   select,
 } from '@inquirer/prompts';
 import type { Context, DistributiveMerge, Prettify } from '@inquirer/type';
-import { Observable } from 'rxjs';
+import type { Observable } from './utils/observable.ts';
 
 export type Answers<Key extends string = string> = Record<Key, any>;
 

packages/inquirer/src/ui/prompt.ts

@@ -1,17 +1,5 @@
 /* eslint-disable @typescript-eslint/no-explicit-any, @typescript-eslint/no-unsafe-return, @typescript-eslint/no-unsafe-assignment */
 import readline from 'node:readline';
-import {
-  defer,
-  EMPTY,
-  from,
-  of,
-  concatMap,
-  filter,
-  reduce,
-  isObservable,
-  Observable,
-  lastValueFrom,
-} from 'rxjs';
 import runAsync from 'run-async';
 import MuteStream from 'mute-stream';
 import { AbortPromptError } from '@inquirer/core';
@@ -24,6 +12,13 @@ import type {
   PromptSession,
   StreamOptions,
 } from '../types.ts';
+import {
+  EMPTY,
+  createObservableController,
+  isObservableLike,
+  observableToAsyncIterable,
+} from '../utils/observable.ts';
+import type { InteropObservable } from '../utils/observable.ts';
 
 export const _ = {
   set: (obj: Record<string, unknown>, path: string = '', value: unknown): void => {
@@ -139,6 +134,8 @@ class UnknownPromptTypeError extends Error {
   }
 }
 
+type AnswerEvent = { name: string; answer: unknown };
+
 class TTYError extends Error {
   override name = 'TTYError';
   isTtyError = true;
@@ -208,7 +205,7 @@ function isPromptConstructor(
 export default class PromptsRunner<A extends Answers> {
   private prompts: PromptCollection;
   answers: Partial<A> = {};
-  process: Observable<any> = EMPTY;
+  process: InteropObservable<AnswerEvent> = EMPTY;
   private abortController: AbortController = new AbortController();
   private opt: StreamOptions;
 
@@ -219,58 +216,48 @@ export default class PromptsRunner<A extends Answers> {
 
   async run(questions: PromptSession<A>, answers?: Partial<A>): Promise<A> {
     this.abortController = new AbortController();
+    const processController = createObservableController<AnswerEvent>();
+    this.process = processController.observable;
 
     // Keep global reference to the answers
     this.answers = typeof answers === 'object' ? { ...answers } : {};
 
-    let obs: Observable<Question<A>>;
+    try {
+      for await (const question of this.getQuestions(questions)) {
+        if (await this.shouldRun(question)) {
+          const answer = await this.fetchAnswer(question);
+          _.set(this.answers, answer.name, answer.answer);
+          processController.next(answer);
+        }
+      }
+
+      processController.complete();
+      // oxlint-disable-next-line typescript/no-unsafe-type-assertion
+      return this.answers as A;
+    } catch (error: unknown) {
+      processController.error(error);
+      throw error;
+    } finally {
+      this.close();
+    }
+  }
+
+  private async *getQuestions(
+    questions: PromptSession<A>,
+  ): AsyncGenerator<Question<A>, void> {
     if (isQuestionArray(questions)) {
-      obs = from(questions);
-    } else if (isObservable(questions)) {
-      obs = questions;
+      yield* questions;
+    } else if (isObservableLike(questions)) {
+      yield* observableToAsyncIterable(questions);
     } else if (isQuestionMap(questions)) {
       // Case: Called with a set of { name: question }
-      obs = from(
-        Object.entries(questions).map(([name, question]): Question<A> => {
-          return Object.assign({}, question, { name });
-        }),
-      );
+      for (const [name, question] of Object.entries(questions)) {
+        yield Object.assign({}, question, { name });
+      }
     } else {
       // Case: Called with a single question config
-      obs = from([questions]);
+      yield questions;
     }
-
-    this.process = obs.pipe(
-      concatMap((question) =>
-        of(question).pipe(
-          concatMap((question) =>
-            from(
-              this.shouldRun(question).then((shouldRun: boolean | void) => {
-                if (shouldRun) {
-                  return question;
-                }
-                return undefined;
-              }),
-            ).pipe(filter((val) => val != null)),
-          ),
-          concatMap((question) => defer(() => from(this.fetchAnswer(question)))),
-        ),
-      ),
-    );
-
-    return (
-      lastValueFrom(
-        this.process.pipe(
-          reduce((answersObj, answer: { name: string; answer: unknown }) => {
-            _.set(answersObj, answer.name, answer.answer);
-            return answersObj;
-          }, this.answers),
-        ),
-      )
-        // oxlint-disable-next-line typescript/no-unsafe-type-assertion
-        .then(() => this.answers as A)
-        .finally(() => this.close())
-    );
   }
 
   private prepareQuestion = async (question: Question<A>) => {