feature: add setLoading rxjs operator function

packages/rxjs/src/operators/set-loading.operator.spec.ts

@@ -0,0 +1,31 @@
+import { BehaviorSubject, of, throwError } from 'rxjs';
+import { delay } from 'rxjs/operators';
+import { flipFlag } from './set-loading.operator';
+
+describe('set-loading.operator.ts', () => {
+  let isLoading$: BehaviorSubject<boolean>;
+
+  beforeEach(() => {
+    isLoading$ = new BehaviorSubject<boolean>(false);
+  });
+
+  it('should set loading state to false after source completes', () => {
+    of('test')
+      .pipe(flipFlag(isLoading$))
+      .subscribe({
+        complete: () => {
+          expect(isLoading$.getValue()).toBe(false);
+        }
+      });
+  });
+
+  it('should set loading state to false after source errors', () => {
+    throwError('error')
+      .pipe(delay(10), flipFlag(isLoading$), delay(10))
+      .subscribe({
+        error: () => {
+          expect(isLoading$.getValue()).toBe(false);
+        }
+      });
+  });
+});

packages/rxjs/src/operators/set-loading.operator.ts

@@ -0,0 +1,12 @@
+import { BehaviorSubject, defer, Observable, OperatorFunction } from 'rxjs';
+import { finalize } from 'rxjs/operators';
+
+export function flipFlag<T>(flagSubject$: BehaviorSubject<boolean>): OperatorFunction<T, T> {
+  const initialSubjectValue: boolean = flagSubject$.getValue();
+
+  return (source: Observable<T>): Observable<T> =>
+    defer(() => {
+      flagSubject$.next(!initialSubjectValue);
+      return source.pipe(finalize(() => flagSubject$.next(!!initialSubjectValue)));
+    });
+}