RxJS is a powerful library for reactive programming, and one of its key features being the ability to create custom operators. In this guide, we'll look at some handy custom operators and show how to implement them. ## 1. withLoading Create a custom `switchMap` that seamlessly manages loading states for each stream. ```typescript import { Observable, of } from 'rxjs'; import { map, catchError, startWith, finalize } from 'rxjs/operators'; interface WithLoadingResult { loading: boolean; data?: T; error?: any; } export function withLoading() { return (source: Observable) => source.pipe( startWith({ loading: true }), map((data) => ({ loading: false, data })), catchError((error) => of({ loading: false, error })), finalize(() => ({})) ); } // Usage someObservable$.pipe(withLoading()).subscribe(({ loading, data, error }) => { if (loading) { console.log('Loading...'); } else if (error) { console.error('Error:', error); } else { console.log('Data:', data); } }); ``` ## 2. debounceAndDistinct Effectively manage user input, minimizing unnecessary API requests. We can achieve this by combining debounce and distinctUntilChanged ```typescript import { Observable } from 'rxjs'; import { debounceTime, distinctUntilChanged } from 'rxjs/operators'; export function debounceAndDistinct(time: number = 300) { return (source: Observable) => source.pipe( debounceTime(time), distinctUntilChanged() ); } // Usage searchInput$.pipe(debounceAndDistinct()).subscribe((value) => { // Perform search with debounced and distinct value }); ``` ## 3. retryWithBackoff Apply a backoff strategy for retrying failed requests more intelligently. ```typescript import { Observable, throwError, timer } from 'rxjs'; import { mergeMap, retryWhen } from 'rxjs/operators'; export function retryWithBackoff( maxRetries: number = 3, backoffTime: number = 1000 ) { return (source: Observable) => source.pipe( retryWhen((errors) => errors.pipe( mergeMap((error, index) => { const retryAttempt = index + 1; if (retryAttempt > maxRetries) { return throwError(error); } console.log(`Retry attempt ${retryAttempt}: retrying in ${backoffTime}ms`); return timer(backoffTime * retryAttempt); }) ) ) ); } // Usage apiCall$.pipe(retryWithBackoff()).subscribe( (data) => console.log('Success:', data), (error) => console.error('Error:', error) ); ``` ## 4. cachingOperator Cache API responses to reduce server load and improve efficiency. ```typescript import { Observable, of } from 'rxjs'; import { tap, shareReplay } from 'rxjs/operators'; export function cachingOperator(cacheTime: number = 60000) { let cachedData: T; let cachedTime: number; return (source: Observable) => new Observable((observer) => { if (cachedData && Date.now() - cachedTime < cacheTime) { observer.next(cachedData); observer.complete(); } else { source .pipe( tap((data) => { cachedData = data; cachedTime = Date.now(); }), shareReplay(1) ) .subscribe(observer); } }); } // Usage apiCall$.pipe(cachingOperator()).subscribe((data) => console.log('Data:', data)); ``` ## 5. progressiveLoading Load data incrementally, emitting partial results to enhance perceived performance. ```typescript import { Observable } from 'rxjs'; import { expand, take, map } from 'rxjs/operators'; export function progressiveLoading( pageSize: number = 10, maxItems: number = 100 ) { return (source: Observable) => source.pipe( expand((items, index) => items.length < maxItems ? source.pipe( map((newItems) => [...items, ...newItems.slice(0, pageSize)]) ) : [] ), take(Math.ceil(maxItems / pageSize)) ); } // Usage apiCall$.pipe(progressiveLoading()).subscribe((partialData) => { console.log('Partial data:', partialData); }); ``` ## 6. errorHandlingOperator Centralize error management for consistent and streamlined error handling. ```typescript import { Observable, throwError } from 'rxjs'; import { catchError } from 'rxjs/operators'; export function errorHandlingOperator() { return (source: Observable) => source.pipe( catchError((error) => { console.error('An error occurred:', error); // You can add custom error handling logic here return throwError('Something went wrong. Please try again later.'); }) ); } // Usage apiCall$.pipe(errorHandlingOperator()).subscribe( (data) => console.log('Data:', data), (error) => console.error('Handled error:', error) ); ``` ## 7. optimisticUpdate Immediately update the UI before API confirmation, with a rollback in case of failure. ```typescript import { Observable, of } from 'rxjs'; import { map, catchError, switchMap } from 'rxjs/operators'; export function optimisticUpdate( updateFn: (data: T) => T, apiCall: (data: T) => Observable ) { return (source: Observable) => source.pipe( map((data) => ({ optimistic: updateFn(data), original: data })), switchMap(({ optimistic, original }) => apiCall(optimistic).pipe( map(() => optimistic), catchError(() => { console.warn('API call failed. Reverting to original data.'); return of(original); }) ) ) ); } // Usage const updateTodo = (todo: Todo): Todo => ({ ...todo, completed: true }); const apiUpdateTodo = (todo: Todo): Observable => // API call implementation todoStream$.pipe(optimisticUpdate(updateTodo, apiUpdateTodo)) .subscribe((updatedTodo) => console.log('Updated todo:', updatedTodo)); ``` ## 8. throttleAndBuffer Use throttle combined with buffer to batch updates and manage data streams effectively. ```typescript import { Observable } from 'rxjs'; import { buffer, throttleTime } from 'rxjs/operators'; export function throttleAndBuffer(time: number = 1000) { return (source: Observable) => source.pipe( buffer(source.pipe(throttleTime(time))), filter((batch) => batch.length > 0) ); } // Usage dataStream$.pipe(throttleAndBuffer()).subscribe((batch) => { console.log('Batched updates:', batch); }); ``` ## 9. conditionalMerge Merge multiple observables based on dynamic conditions for flexible data combination. ```typescript import { Observable, merge } from 'rxjs'; import { filter } from 'rxjs/operators'; export function conditionalMerge( ...sources: Array<[Observable, (value: T) => boolean]> ) { return merge( ...sources.map(([source, condition]) => source.pipe(filter(condition)) ) ); } // Usage const source1$ = of(1, 2, 3, 4); const source2$ = of('a', 'b', 'c', 'd'); conditionalMerge( [source1$, (value) => value % 2 === 0], [source2$, (value) => ['a', 'c'].includes(value)] ).subscribe((value) => console.log('Merged value:', value)); ``` ## 10. smartPolling Implement adaptive polling intervals that respond to data changes or user activity. ```typescript import { Observable, timer } from 'rxjs'; import { switchMap, tap, distinctUntilChanged } from 'rxjs/operators'; export function smartPolling( pollFn: () => Observable, baseInterval: number = 5000, maxInterval: number = 60000 ) { let currentInterval = baseInterval; let lastValue: T; return new Observable((observer) => { const subscription = timer(0, baseInterval) .pipe( switchMap(() => pollFn()), tap((value) => { if (JSON.stringify(value) !== JSON.stringify(lastValue)) { currentInterval = baseInterval; } else { currentInterval = Math.min(currentInterval * 2, maxInterval); } lastValue = value; }), distinctUntilChanged() ) .subscribe(observer); return () => subscription.unsubscribe(); }); } // Usage const pollApi = (): Observable => // API call implementation smartPolling(pollApi).subscribe((data) => console.log('Polled data:', data)); ``` ## 11. filterOnlyPresent Filters out null or undefined values from the stream, ensuring only valid data is emitted. ```typescript import { Observable } from 'rxjs'; import { filter } from 'rxjs/operators'; export function filterOnlyPresent() { return (source: Observable) => source.pipe( filter((value): value is NonNullable => value !== null && value !== undefined) ); } // Usage someStream$.pipe(filterOnlyPresent()).subscribe((value) => { console.log('Non-null value:', value); }); ``` ## 12. filterOnlyPropertyPresent Emits only when a specified property in the returned object is neither null nor undefined. ```typescript import { Observable } from 'rxjs'; import { filter } from 'rxjs/operators'; export function filterOnlyPropertyPresent(prop: keyof T) { return (source: Observable) => source.pipe( filter((value) => value[prop] !== null && value[prop] !== undefined) ); } // Usage interface User { id: number; name: string | null; } userStream$.pipe(filterOnlyPropertyPresent('name')).subscribe((user) => { console.log('User with name:', user); }); ``` These custom operators can greatly improve your RxJS workflows, making your code more efficient and cleaner. Be sure to thoroughly test these operators in your specific scenarios and adjust them as needed.

tunjiadeyeri

Custom RxJS Operators to Improve your Angular Apps

Join Our Newsletter. No Spam, Only the good stuff.