fix: make event bus as signleton and failure events to be fired

src/commands/base.command.ts

@@ -10,7 +10,7 @@ export type ReturnTypeOfRun<T extends BaseCommand> = ReturnType<T['run']>;
 export abstract class BaseCommand {
 	protected readonly logger: Logger;
 
-	protected readonly eventBus = new ResilienceEventBus();
+	protected readonly eventBus = ResilienceEventBus.getInstance();
 
 	protected readonly strategies: Strategy[];
 
@@ -42,13 +42,11 @@ export abstract class BaseCommand {
 		if (error instanceof TimeoutException) {
 			this.logger.debug('Command timed out');
 			this.eventBus.emit(ResilienceEventType.Timeout, this);
-			return error;
 		}
 
 		if (error instanceof CircuitOpenedException) {
 			this.logger.debug('Command short-circuited');
 			this.eventBus.emit(ResilienceEventType.ShortCircuit, this);
-			return error;
 		}
 
 		this.logger.debug('Command failed');

src/commands/resilience.command.ts

@@ -1,5 +1,5 @@
 import { BaseCommand, ReturnTypeOfRun } from './base.command';
-import { defer, lastValueFrom, Observable } from 'rxjs';
+import { catchError, defer, lastValueFrom, Observable, tap } from 'rxjs';
 import { ResilienceEventType } from '../enum';
 
 export abstract class ResilienceCommand extends BaseCommand {
@@ -8,19 +8,19 @@ export abstract class ResilienceCommand extends BaseCommand {
 	public execute(...args: Parameters<this['run']>): ReturnTypeOfRun<this> {
 		this.eventBus.emit(ResilienceEventType.Emit, this);
 
-		try {
-			let observable: Observable<ReturnTypeOfRun<this>> = defer(() => this.run(...args));
+		let observable: Observable<ReturnTypeOfRun<this>> = defer(() => this.run(...args));
 
-			for (const strategy of this.strategies) {
-				observable = strategy.process(observable, this, ...args);
-			}
-
-			return lastValueFrom(observable).then(result => {
-				this.onSuccess();
-				return result;
-			}) as ReturnTypeOfRun<this>;
-		} catch (error) {
-			throw this.onFailure(error);
+		for (const strategy of this.strategies) {
+			observable = strategy.process(observable, this, ...args);
 		}
+
+		return lastValueFrom(
+			observable.pipe(
+				catchError(error => {
+					throw this.onFailure(error);
+				}),
+				tap(() => this.onSuccess())
+			)
+		) as ReturnTypeOfRun<this>;
 	}
 }

src/resilience.event-bus.ts

@@ -8,6 +8,8 @@ export class ResilienceEventBus {
 		return ResilienceEventBus.instance;
 	}
 
+	private constructor() {}
+
 	private readonly emitter = new EventEmitter();
 
 	public on<K extends keyof ResilienceEvents>(event: K, fn: (args: ResilienceEvents[K]) => void) {

src/strategies/circuit-breaker.strategy.ts

@@ -67,7 +67,7 @@ export class CircuitBreakerStrategy extends Strategy<CircuitBreakerOptions> {
 						state.openedAt = 0;
 					}
 				}
-				
+
 				state.lastRequestTimeMs = Date.now();
 
 				if (isOpen()) {

test/commands/resilience.command.spec.ts

@@ -1,14 +1,26 @@
-import { ResilienceCommand } from '../../src';
+import { ResilienceCommand, ResilienceEventBus } from '../../src';
 import { retryStrategy } from './fixtures/strategy.fixture';
 
 class TestCommand extends ResilienceCommand {
-	protected strategies = [retryStrategy];
-
 	private count = 0;
 
+	private isError = false;
+
+	public setCount(count: number) {
+		this.count = count;
+	}
+
+	public setIsError(isError: boolean) {
+		this.isError = isError;
+	}
+
 	public async run() {
 		this.count += 1;
 
+		if (this.isError) {
+			throw new Error('Test');
+		}
+
 		if (this.count !== 3) {
 			throw new Error('Test');
 		}
@@ -18,10 +30,35 @@ class TestCommand extends ResilienceCommand {
 }
 
 describe('Resilience Command', () => {
+	const command = new TestCommand([retryStrategy]);
+	const eventBus = ResilienceEventBus.getInstance();
+
+	let callback = jest.fn();
+
+	beforeEach(() => {
+		command.setCount(0);
+		command.setIsError(false);
+		callback = jest.fn();
+	});
+
 	it('should be able to retry a promise', async () => {
-		const command = new TestCommand([retryStrategy]);
+		eventBus.on('success', callback);
+
 		const value = await command.execute();
 
 		expect(value).toBe(1000);
+		expect(callback).toHaveBeenCalled();
+	});
+
+	it('should emit the failure', async () => {
+		command.setIsError(true);
+		eventBus.on('failure', callback);
+
+		try {
+			await command.execute();
+		} catch (error) {
+			expect(error.message).toBe('Test');
+			expect(callback).toHaveBeenCalled();
+		}
 	});
 });