@csiro-geoanalytics/utils-rxjs

RxJS Utilities

A suite of helper classes and utilities for RxJS.

SubscriptionManager

A collection that handles registration, subscription and disposal of multiple RxJS Observables. The class might be particularly useful in applications (e.g., Angular components) where one might need to subscribe and keep references to multiple Observables during the lifetime of a parent class.

The use of the SubscriptionManager class eliminates the need to store multiple class members to store references to individual Observables and the need to unsubscribe from them manually.

Example

@Injectable()
export class UserService implements OnDestroy
{
	private _jwtToken	: firebase.auth.IdTokenResult;
	private user		= new BehaviorSubject<User>(undefined);

	// Create an instance of the SubscriptionManager.
	private subscriptions = new SubscriptionManager();

	constructor(private afAuth: AngularFireAuth)
	{
		// Register multiple Observables.
		this.subscriptions.register(this.afAuth.authState, user => this.user.next(user));
		this.subscriptions.register
			this.afAuth.idTokenResult,
			{
				next		: token => this._jwtToken = token,
				error		: () => { console.error("An error has occurred."); }.
				complete	: () => {}
			}
		);

		// Subscribe to all at once.
		this.subscriptions.subscribeAll();
	}

	ngOnDestroy()
	{
		// Unsubscribe from all Observables.
		this.subscriptions.unsubscribeAll();
	}
}

NamedEventManager

The NamedEventManager class simplifies the handling of custom Subject-based events.

Usually, to define a custom event, one will have to instantiate a Subject class and expose it as an Observable for clients to use. This would result in two class members for each event, e.g.:

@Injectable()
export class LayoutService implements OnDestroy
{
	// A Subject object used to emit events.
	private  relayoutEvent = new Subject();

	// An Observable that clients can listen to.
	readonly relayout$ = this.relayoutEvent.asObservable();

	relayout()
	{
		// ...

		// Broadcast an event.
		this.relayoutEvent.next();
	}
}

Should there be multiple custom events, the same Subject instantiation process will need to be repeated. The NamedEventManager is named collection that simplifies this process and allows the use of enumerators for event definition.

Example

export const enum Events
{
	MapLayerDataLoaded,
	ProfileReload
	// ...
}

@Injectable()
export class SomeService implements OnDestroy
{
	private events = new NamedEventManager();

	constructor()
	{
		// Register events.
		this.events.registerBehaviorSubject<MapLayers.LineGroup[]>(Events.MapLayerDataLoaded);
		this.events.registerSubject<void>(Events.ProfileReload);
		this.events.registerSubject<void>("CustomNamedEvent");
	}

	ngOnDestroy()
	{
		// Unregister all events.
		this.events.unregisterAll();
	}

	getEvent$<T>(event: Events): Observable<T>
	{
		return (this.events.getSubject<T>(event) instanceof BehaviorSubject)
			? this.events.getObservable<T>(event)
				.pipe(
					// Skip the initial undefined value of a BehaviorSubject.
					filter(v => v !== undefined)
				)
			: this.events.getObservable<T>(event);
	}

	broadcast<T>(event: Events, value: T = null)
	{
		// Broadcast an event.
		this.events.emit(event, value);
	}

	getLastValue<T>(event: Events): T
	{
		// Get last value of an BehaviorSubject.
		return this.events.getLastValue<T>(event);
	}
}

The listener will subscribe to that event's Observable:

this.events.getObservable<T>("CustomNamedEvent")
	.pipe(
		// Do something with it.
	)
	.subscribe(...)

GenericRetryStrategy / GenericRetryOperation

GenericRetryStrategy

The GenericRetryStrategy is an Observable for the RxJS retryWhen operator that defines the rules for retry attempts. The default settings for the GenericRetryStrategy will retry operation for 3 times with a one-second delay between attempts. A scaling factor can be introduced to progressively increase the interval between attempts.

Example

of(observable$)
	.pipe(
		map(() => operationThatMightFail()),
		retryWhen(GenericRetryStrategy())
	)
	.subscribe(...);

The following example will retry the operation for up to 5 times with intervals increasing by 500 ms with each attempt starting from one second:

of(observable$)
	.pipe(
		map(() => operationThatMightFail()),
		retryWhen(
			GenericRetryStrategy({
				maxRetryAttempts	: 5,
				retryInterval		: 1000,
				scalingDuration		: 500
			})
		)
	)
	.subscribe(...);

GenericRetryOperation

The GenericRetryOperation function implements an RxJS-based approach to retry an operation until a certain condition is met according to the specified GenericRetryStrategy. A strategy defines the number of retry attempts, retry interval, and scaling factor allowing to increase retry interval progressively between attempts.

The GenericRetryOperation function is asynchronous and always returns immediately.

Example

GenericRetryOperation(
	() => {
		// Some condition that needs to be met.
		return true;
	},
	() => {
		// Success handler.
	},
	() => {
		// Error handler.
	},
	() => {
		// Complete handler.
	},
	{
		// Options (defaults).
		maxRetryAttempts	: 3,
		retryInterval		: 1000,
		scalingDuration		: 0
	}
);