Query Operator -- manage your asynchronous data query

apps/rxjs.dev/content/guide/operators.md

@@ -166,6 +166,7 @@ These are Observable creation operators that also have join functionality -- emi
 - [`mergeScan`](/api/operators/mergeScan)
 - [`pairwise`](/api/operators/pairwise)
 - [`partition`](/api/operators/partition)
+- [`query`](/api/operators/query)
 - [`scan`](/api/operators/scan)
 - [`switchScan`](/api/operators/switchScan)
 - [`switchMap`](/api/operators/switchMap)

packages/rxjs/spec-dtslint/operators/query-spec.ts

@@ -0,0 +1,13 @@
+import { of, throwError } from 'rxjs';
+import { query } from '../../src/internal/operators/query';
+import { QueryResult } from '../../src/internal/types';
+
+it('should infer QueryResult<T> when T is string', () => {
+  const result = of('hello').pipe(query());
+  // $ExpectType Observable<QueryResult<string>>
+});
+
+it('should handle error types correctly', () => {
+  const result = throwError(() => new Error()).pipe(query());
+  // $ExpectType Observable<QueryResult<unknown>>
+});

packages/rxjs/spec/operators/query-spec.ts

@@ -0,0 +1,60 @@
+import { expect } from 'chai';
+import { TestScheduler } from 'rxjs/testing';
+import { query } from '../../src/internal/operators/query';
+import { QueryResult } from '../../src/internal/types';
+import { observableMatcher } from '../helpers/observableMatcher';
+
+/** @test {query} */
+describe('query operator', () => {
+  let testScheduler: TestScheduler;
+
+  beforeEach(() => {
+    testScheduler = new TestScheduler(observableMatcher);
+  });
+
+  it('should emit loading and then data', () => {
+    testScheduler.run(({ cold, expectObservable }) => {
+      const source$ = cold('  --a|', { a: 'value' });
+      const expected = '      i-a|';
+
+      const expectedValues = {
+        i: { isLoading: true, data: null, error: null },
+        a: { isLoading: false, data: 'value', error: null },
+      };
+
+      const result$ = source$.pipe(query());
+
+      expectObservable(result$).toBe(expected, expectedValues);
+    });
+  });
+
+  it('should emit loading and then error', () => {
+    testScheduler.run(({ cold, expectObservable }) => {
+      const source$ = cold('  --#', {}, 'BOOM');
+      const expected = '    i-(e|)';
+      const expectedValues: Record<string, QueryResult<any>> = {
+        i: { isLoading: true, data: null, error: null },
+        e: { isLoading: false, data: null, error: 'BOOM' },
+      };
+
+      const result$ = source$.pipe(query());
+
+      expectObservable(result$).toBe(expected, expectedValues);
+    });
+  });
+
+  it('should complete after emitting data', () => {
+    testScheduler.run(({ cold, expectObservable }) => {
+      const source$ = cold('  a---|', { a: 123 });
+      const expected = '    (ia)|';
+      const expectedValues = {
+        i: { isLoading: true, data: null, error: null },
+        a: { isLoading: false, data: 123, error: null },
+      };
+
+      const result$ = source$.pipe(query());
+
+      expectObservable(result$).toBe(expected, expectedValues);
+    });
+  });
+});

packages/rxjs/src/index.ts

@@ -140,6 +140,7 @@ export { min } from './internal/operators/min.js';
 export { observeOn } from './internal/operators/observeOn.js';
 export { onErrorResumeNextWith } from './internal/operators/onErrorResumeNextWith.js';
 export { pairwise } from './internal/operators/pairwise.js';
+export { query } from './internal/operators/query.js';
 export { raceWith } from './internal/operators/raceWith.js';
 export { reduce } from './internal/operators/reduce.js';
 export type { RepeatConfig } from './internal/operators/repeat.js';

packages/rxjs/src/internal/operators/query.ts

@@ -0,0 +1,31 @@
+import { Observable, of } from 'rxjs';
+import { QueryResult } from '../types';
+import { startWith, catchError, map } from 'rxjs/operators';
+
+/**
+ * Transforms an observable into a query result observable.
+ *
+ * The query result observable will emit one of the following states:
+ *
+ * - `isLoading: true, data: null, error: null` when the source observable is
+ *   executing.
+ * - `isLoading: false, data: T, error: null` when the source observable
+ *   completes successfully.
+ * - `isLoading: false, data: null, error: Error` when the source observable
+ *   throws an error.
+ *
+ * The first state is sent immediately using `startWith`, the others are sent
+ * when the source observable completes or throws an error.
+ *
+ * @param source$ The source observable to transform
+ * @returns An observable that emits the query result
+ */
+export function query<T>() {
+  return (source$: Observable<T>): Observable<QueryResult<T>> => {
+    return source$.pipe(
+      map((data) => ({ isLoading: false, data, error: null })),
+      catchError((error) => of({ isLoading: false, data: null, error })),
+      startWith({ isLoading: true, data: null, error: null })
+    );
+  };
+}

packages/rxjs/src/internal/types.ts

@@ -72,6 +72,17 @@ export interface TimeInterval<T> {
   interval: number;
 }
 
+/**
+ * The result of a query.
+ * 
+ * @see {@link query}
+ */
+export interface QueryResult<T> {
+  isLoading: boolean;
+  data: T | null;
+  error: any;
+}
+
 /* SUBSCRIPTION INTERFACES */
 
 export interface Unsubscribable {