Merge 0b4126f into 3026c9d

package.json

@@ -1,7 +1,7 @@
 {
 	"name": "ngx-crud",
 	"description": "CRUD services in Angular with effortless aborting, caching and observing",
-	"version": "11.3.0",
+	"version": "12.4.0-alpha.1",
 	"homepage": "https://ngx-crud.com",
 	"license": "MIT",
 	"type": "module",

src/abort/abort.service.ts

@@ -1,6 +1,6 @@
 import { HttpContextToken, HttpRequest } from '@angular/common/http';
 import { Injectable } from '@angular/core';
-import { filter, from, timer, Observable, Subject, Subscription } from 'rxjs';
+import { Observable, Subject, Subscription, filter, from, timer } from 'rxjs';
 import { Context, Store } from './abort.interface';
 import { stripUrlParams } from '../common';
 

src/cache/cache.service.ts

@@ -1,6 +1,6 @@
 import { HttpContextToken, HttpRequest, HttpResponse } from '@angular/common/http';
 import { Injectable } from '@angular/core';
-import { filter, from, timer, Observable, Subscription } from 'rxjs';
+import { Observable, Subscription, filter, from, timer } from 'rxjs';
 import { Context, Store } from './cache.interface';
 import { stripUrlParams } from '../common';
 

src/observe/observe.interceptor.ts

@@ -9,7 +9,7 @@ import
 } from '@angular/common/http';
 import { Injectable } from '@angular/core';
 import { Observable, throwError } from 'rxjs';
-import { catchError, filter, finalize, tap } from 'rxjs/operators';
+import { catchError, filter, tap } from 'rxjs/operators';
 import { Context } from './observe.interface';
 import { ObserveService } from './observe.service';
 
@@ -30,18 +30,22 @@ export class ObserveInterceptor implements HttpInterceptor
 
 	handle<T>(request : HttpRequest<T>, next : HttpHandler) : Observable<HttpEvent<T>>
 	{
-		this.observeService.start();
+		this.observeService.start(request);
 		return next
 			.handle(this.observeService.before(request))
 			.pipe(
 				filter(event => event instanceof HttpResponse),
-				tap((response : HttpResponse<T>) => this.observeService.after(request, response)),
+				tap((response : HttpResponse<T>) =>
+				{
+					this.observeService.after(request, response);
+					this.observeService.complete(request.urlWithParams);
+				}),
 				catchError((response : HttpErrorResponse) =>
 				{
 					this.observeService.after(request, response);
+					this.observeService.error(request.urlWithParams);
 					return throwError(() => response);
-				}),
-				finalize(() => this.observeService.end(request))
+				})
 			);
 	}
 }

src/observe/observe.interface.ts

@@ -1,5 +1,13 @@
 import { HttpErrorResponse, HttpRequest, HttpResponse } from '@angular/common/http';
+import { Subject, Subscription } from 'rxjs';
 import { UniversalMethod } from '../common';
+import { ObserveStatus } from './observe.type';
+
+export interface Store
+{
+	status : Subject<ObserveStatus>;
+	timer : Subscription;
+}
 
 export interface Context
 {

src/observe/observe.service.ts

@@ -1,9 +1,10 @@
 import { HttpContextToken, HttpErrorResponse, HttpRequest, HttpResponse } from '@angular/common/http';
 import { Optional, Inject, Injectable } from '@angular/core';
-import { timer, Subject, Subscription } from 'rxjs';
-import { Context, ObserveAfterEffect, ObserveBeforeEffect } from './observe.interface';
-import { ObserveStatus } from './observe.type';
+import { Observable, Subject, Subscription, filter, from, timer, mergeMap } from 'rxjs';
+import { ObserveAfterEffect, ObserveBeforeEffect, Context, Store } from './observe.interface';
 import { OBSERVE_EFFECT } from './observe.token';
+import { ObserveStatus } from './observe.type';
+import { stripUrlParams } from '../common';
 
 @Injectable()
 export class ObserveService
@@ -15,8 +16,7 @@ export class ObserveService
 	};
 
 	protected token : HttpContextToken<Context> = new HttpContextToken<Context>(() => this.defaultContext);
-	protected status : Subject<ObserveStatus> = new Subject<ObserveStatus>();
-	protected timer : Subscription = new Subscription();
+	protected store : Map<string, Store> = new Map();
 
 	constructor(@Optional() @Inject(OBSERVE_EFFECT) protected observeEffect : ObserveBeforeEffect | ObserveAfterEffect)
 	{
@@ -27,9 +27,20 @@ export class ObserveService
 		return this.token;
 	}
 
-	start() : this
+	start<T>(request : HttpRequest<T>) : this
 	{
-		this.status.next('STARTED');
+		const context : Context = request.context.get(this.getToken());
+
+		if (this.has(request))
+		{
+			this.store.get(request.urlWithParams).timer.unsubscribe();
+		}
+		this.store.set(request.urlWithParams,
+		{
+			status: new Subject<ObserveStatus>(),
+			timer: context.lifetime > 0 ? timer(context.lifetime).subscribe(() => this.complete(request.urlWithParams)) : new Subscription()
+		});
+		this.store.get(request.urlWithParams).status.next('STARTED');
 		return this;
 	}
 
@@ -51,23 +62,61 @@ export class ObserveService
 		return this;
 	}
 
-	end<T>(request : HttpRequest<T>) : this
+	has<T>(request : HttpRequest<T>) : boolean
 	{
-		const context : Context = request.context.get(this.getToken());
+		return this.store.has(request.urlWithParams);
+	}
+
+	error(urlWithParams : string) : this
+	{
+		if (this.store.has(urlWithParams))
+		{
+			this.store.get(urlWithParams).status.next('ERRORED');
+			this.store.get(urlWithParams).timer.unsubscribe();
+		}
+		return this;
+	}
+
+	complete(urlWithParams : string) : this
+	{
+		if (this.store.has(urlWithParams))
+		{
+			this.store.get(urlWithParams).status.next('COMPLETED');
+			this.store.get(urlWithParams).timer.unsubscribe();
+		}
+		return this;
+	}
 
-		this.timer.unsubscribe();
-		this.timer = context.lifetime > 0 ? timer(context.lifetime).subscribe(() => this.completeAll()) : new Subscription();
+	completeMany(url : string) : this
+	{
+		this.store.forEach((store, urlWithParams) => stripUrlParams(urlWithParams) === url ? this.complete(urlWithParams) : null);
 		return this;
 	}
 
 	completeAll() : this
 	{
-		this.status.next('COMPLETED');
+		this.store.forEach((store, urlWithParams) => this.complete(urlWithParams));
 		return this;
 	}
 
-	observeAll() : Subject<ObserveStatus>
+	observe(urlWithParams : string) : Observable<ObserveStatus>
+	{
+		return from(this.store).pipe(
+			filter(value => value[0] === urlWithParams),
+			mergeMap(value => value[1].status)
+		);
+	}
+
+	observeMany(url : string) : Observable<ObserveStatus>
+	{
+		return from(this.store).pipe(
+			filter(value => stripUrlParams(value[0]) === url),
+			mergeMap(value => value[1].status)
+		);
+	}
+
+	observeAll() : Observable<ObserveStatus>
 	{
-		return this.status;
+		return from(this.store).pipe(mergeMap(value => value[1].status));
 	}
 }

src/observe/observe.type.ts

@@ -1 +1 @@
-export type ObserveStatus = 'STARTED' | 'COMPLETED';
+export type ObserveStatus = 'STARTED' | 'COMPLETED' | 'ERRORED';

tests/abort.service.spec.ts

@@ -1,10 +1,10 @@
 import { HttpClientModule } from '@angular/common/http';
+import { take } from 'rxjs/operators';
 import { inject, TestBed } from '@angular/core/testing';
 import { expect } from 'chai';
 import { CrudModule, AbortService } from '../src';
 import { TestService } from './test.service';
 import { mockRequest } from './test.helper';
-import { take } from 'rxjs/operators';
 
 before(() =>
 {

tests/cache.service.spec.ts

@@ -1,7 +1,7 @@
 import { HttpClientModule } from '@angular/common/http';
+import { concatMap, delay, take, tap } from 'rxjs/operators';
 import { inject, TestBed } from '@angular/core/testing';
 import { expect } from 'chai';
-import { concatMap, delay, take, tap } from 'rxjs/operators';
 import { CrudModule, CacheService } from '../src';
 import { TestService } from './test.service';
 import { mockRequest } from './test.helper';

tests/observe.service.spec.ts

@@ -1,4 +1,6 @@
 import { HttpClientModule } from '@angular/common/http';
+import { EMPTY } from 'rxjs';
+import { catchError, take } from 'rxjs/operators';
 import { inject, TestBed } from '@angular/core/testing';
 import { expect } from 'chai';
 import { CrudModule, ObserveService, OBSERVE_EFFECT } from '../src';
@@ -56,6 +58,7 @@ describe('ObserveService', () =>
 		{
 			testService
 				.enableObserve('GET')
+				.setParam('observe', '1')
 				.find()
 				.subscribe(
 				{
@@ -75,6 +78,113 @@ describe('ObserveService', () =>
 		})();
 	});
 
+	it('natural error', done =>
+	{
+		inject(
+		[
+			ObserveService,
+			TestService
+		], (observeService : ObserveService, testService : TestService) =>
+		{
+			testService
+				.clone()
+				.setApiRoute('/error')
+				.enableObserve()
+				.find()
+				.pipe(catchError(() => EMPTY))
+				.subscribe();
+			observeService
+				.observe('https://jsonplaceholder.typicode.com/error')
+				.pipe(take(1))
+				.subscribe(
+				{
+					next: observeStatus =>
+					{
+						if (observeStatus === 'ERRORED')
+						{
+							testService.clear();
+							done();
+						}
+					},
+					error: () =>
+					{
+						testService.clear();
+						done('error');
+					}
+				});
+		})();
+	});
+
+	it('observe', done =>
+	{
+		inject(
+		[
+			ObserveService,
+			TestService
+		], (observeService : ObserveService, testService : TestService) =>
+		{
+			testService
+				.enableObserve()
+				.setParam('observe', '2')
+				.find()
+				.subscribe();
+			observeService
+				.observe('https://jsonplaceholder.typicode.com/posts?observe=2')
+				.pipe(take(1))
+				.subscribe(
+				{
+					next: observeStatus =>
+					{
+						if (observeStatus === 'COMPLETED')
+						{
+							testService.clear();
+							done();
+						}
+					},
+					error: () =>
+					{
+						testService.clear();
+						done('error');
+					}
+				});
+		})();
+	});
+
+	it('observe many', done =>
+	{
+		inject(
+		[
+			ObserveService,
+			TestService
+		], (observeService : ObserveService, testService : TestService) =>
+		{
+			testService
+				.enableObserve()
+				.setParam('observe', '3')
+				.find()
+				.subscribe();
+			observeService
+				.observeMany('https://jsonplaceholder.typicode.com/posts')
+				.pipe(take(1))
+				.subscribe(
+				{
+					next: observeStatus =>
+					{
+						if (observeStatus === 'COMPLETED')
+						{
+							testService.clear();
+							done();
+						}
+					},
+					error: () =>
+					{
+						testService.clear();
+						done('error');
+					}
+				});
+		})();
+	});
+
 	it('observe all', done =>
 	{
 		inject(
@@ -84,17 +194,22 @@ describe('ObserveService', () =>
 		], (observeService : ObserveService, testService : TestService) =>
 		{
 			testService
-				.enableObserve('GET')
+				.enableObserve()
+				.setParam('observe', '4')
 				.find()
 				.subscribe();
 			observeService
 				.observeAll()
+				.pipe(take(1))
 				.subscribe(
 				{
-					next: () =>
+					next: observeStatus =>
 					{
-						testService.clear();
-						done();
+						if (observeStatus === 'COMPLETED')
+						{
+							testService.clear();
+							done();
+						}
 					},
 					error: () =>
 					{