Simplified users and event consumers state.

src/Squidex/app/features/administration/guards/user-must-exist.guard.spec.ts

@@ -9,8 +9,7 @@ import { Router } from '@angular/router';
 import { of } from 'rxjs';
 import { IMock, Mock, Times } from 'typemoq';
 
-import { UserDto } from './../services/users.service';
-import { UsersState } from './../state/users.state';
+import { SnapshotUser, UsersState } from './../state/users.state';
 import { UserMustExistGuard } from './user-must-exist.guard';
 
 describe('UserMustExistGuard', () => {
@@ -32,7 +31,7 @@ describe('UserMustExistGuard', () => {
 
     it('should load user and return true when found', () => {
         usersState.setup(x => x.select('123'))
-            .returns(() => of(<UserDto>{}));
+            .returns(() => of(<SnapshotUser>{}));
 
         let result: boolean;
 

src/Squidex/app/features/administration/pages/users/users-page.component.ts

@@ -7,7 +7,6 @@
 
 import { Component, OnInit } from '@angular/core';
 import { FormControl } from '@angular/forms';
-import { onErrorResumeNext } from 'rxjs/operators';
 
 import { UserDto, UsersState } from '../../declarations';
 
@@ -25,31 +24,31 @@ export class UsersPageComponent implements OnInit {
     }
 
     public ngOnInit() {
-        this.usersState.load().pipe(onErrorResumeNext()).subscribe();
+        this.usersState.load();
     }
 
     public reload() {
-        this.usersState.load(true).pipe(onErrorResumeNext()).subscribe();
+        this.usersState.load(true);
     }
 
     public search() {
-        this.usersState.search(this.usersFilter.value).pipe(onErrorResumeNext()).subscribe();
+        this.usersState.search(this.usersFilter.value);
     }
 
     public goPrev() {
-        this.usersState.goPrev().pipe(onErrorResumeNext()).subscribe();
+        this.usersState.goPrev();
     }
 
     public goNext() {
-        this.usersState.goNext().pipe(onErrorResumeNext()).subscribe();
+        this.usersState.goNext();
     }
 
     public lock(user: UserDto) {
-        this.usersState.lock(user).pipe(onErrorResumeNext()).subscribe();
+        this.usersState.lock(user);
     }
 
     public unlock(user: UserDto) {
-        this.usersState.unlock(user).pipe(onErrorResumeNext()).subscribe();
+        this.usersState.unlock(user);
     }
 
     public trackByUser(index: number, userInfo: { user: UserDto }) {

src/Squidex/app/features/administration/state/event-consumers.state.spec.ts

@@ -6,7 +6,6 @@
  */
 
 import { of, throwError } from 'rxjs';
-import { onErrorResumeNext } from 'rxjs/operators';
 import { IMock, It, Mock, Times } from 'typemoq';
 
 import { DialogService } from '@app/shared';
@@ -46,7 +45,7 @@ describe('EventConsumersState', () => {
     });
 
     it('should show notification on load when reload is true', () => {
-        eventConsumersState.load(true).subscribe();
+        eventConsumersState.load(true);
 
         expect().nothing();
 
@@ -57,7 +56,7 @@ describe('EventConsumersState', () => {
         eventConsumersService.setup(x => x.getEventConsumers())
             .returns(() => throwError({}));
 
-        eventConsumersState.load(true, false).pipe(onErrorResumeNext()).subscribe();
+        eventConsumersState.load(true, false);
 
         expect().nothing();
 
@@ -68,7 +67,7 @@ describe('EventConsumersState', () => {
         eventConsumersService.setup(x => x.getEventConsumers())
             .returns(() => throwError({}));
 
-        eventConsumersState.load(true, true).pipe(onErrorResumeNext()).subscribe();
+        eventConsumersState.load(true, true);
 
         expect().nothing();
 
@@ -79,7 +78,7 @@ describe('EventConsumersState', () => {
         eventConsumersService.setup(x => x.putStart(oldConsumers[1].name))
             .returns(() => of({}));
 
-        eventConsumersState.start(oldConsumers[1]).subscribe();
+        eventConsumersState.start(oldConsumers[1]);
 
         const es_1 = eventConsumersState.snapshot.eventConsumers.at(1);
 
@@ -90,7 +89,7 @@ describe('EventConsumersState', () => {
         eventConsumersService.setup(x => x.putStop(oldConsumers[0].name))
             .returns(() => of({}));
 
-        eventConsumersState.stop(oldConsumers[0]).subscribe();
+        eventConsumersState.stop(oldConsumers[0]);
 
         const es_1 = eventConsumersState.snapshot.eventConsumers.at(0);
 
@@ -101,7 +100,7 @@ describe('EventConsumersState', () => {
         eventConsumersService.setup(x => x.putReset(oldConsumers[0].name))
             .returns(() => of({}));
 
-        eventConsumersState.reset(oldConsumers[0]).subscribe();
+        eventConsumersState.reset(oldConsumers[0]);
 
         const es_1 = eventConsumersState.snapshot.eventConsumers.at(0);
 

src/Squidex/app/features/administration/state/event-consumers.state.ts

@@ -10,6 +10,7 @@ import { Observable } from 'rxjs';
 import { distinctUntilChanged, map, share } from 'rxjs/operators';
 
 import {
+    array,
     DialogService,
     ImmutableArray,
     State
@@ -19,12 +20,14 @@ import { EventConsumerDto, EventConsumersService } from './../services/event-con
 
 interface Snapshot {
     // The list of event consumers.
-    eventConsumers: ImmutableArray<EventConsumerDto>;
+    eventConsumers: EventConsumersList;
 
     // Indicates if event consumers are loaded.
     isLoaded?: boolean;
 }
 
+type EventConsumersList = ImmutableArray<EventConsumerDto>;
+
 @Injectable()
 export class EventConsumersState extends State<Snapshot> {
     public eventConsumers =
@@ -42,21 +45,21 @@ export class EventConsumersState extends State<Snapshot> {
         super({ eventConsumers: ImmutableArray.empty() });
     }
 
-    public load(isReload = false, silent = false): Observable<EventConsumerDto[]> {
+    public load(isReload = false, silent = false): Observable<EventConsumersList> {
         if (!isReload) {
             this.resetState();
         }
 
-        const stream = this.eventConsumersService.getEventConsumers().pipe(share());
+        const stream =
+            this.eventConsumersService.getEventConsumers().pipe(
+                map(dtos => array(dtos)), share());
 
-        stream.subscribe(dtos => {
+        stream.subscribe(eventConsumers => {
             if (isReload && !silent) {
                 this.dialogs.notifyInfo('Event Consumers reloaded.');
             }
 
             this.next(s => {
-                const eventConsumers = ImmutableArray.of(dtos);
-
                 return { ...s, eventConsumers, isLoaded: true };
             });
 
@@ -70,39 +73,41 @@ export class EventConsumersState extends State<Snapshot> {
     }
 
     public start(eventConsumer: EventConsumerDto): Observable<any> {
-        const stream = this.eventConsumersService.putStart(eventConsumer.name).pipe(share());
+        const stream =
+            this.eventConsumersService.putStart(eventConsumer.name).pipe(
+                map(_ => setStopped(eventConsumer, false), share()));
 
-        stream.subscribe(() => {
-            this.replaceEventConsumer(setStopped(eventConsumer, false));
-        }, error => {
-            this.dialogs.notifyError(error);
-        });
+        this.updateState(stream);
 
         return stream;
     }
 
-    public stop(eventConsumer: EventConsumerDto): Observable<any> {
-        const stream = this.eventConsumersService.putStop(eventConsumer.name).pipe(share());
+    public stop(eventConsumer: EventConsumerDto): Observable<EventConsumerDto> {
+        const stream =
+            this.eventConsumersService.putStop(eventConsumer.name).pipe(
+                map(_ => setStopped(eventConsumer, true), share()));
 
-        stream.subscribe(() => {
-            this.replaceEventConsumer(setStopped(eventConsumer, true));
-        }, error => {
-            this.dialogs.notifyError(error);
-        });
+        this.updateState(stream);
 
         return stream;
     }
 
     public reset(eventConsumer: EventConsumerDto): Observable<any> {
-        const stream = this.eventConsumersService.putReset(eventConsumer.name).pipe(share());
+        const stream =
+            this.eventConsumersService.putReset(eventConsumer.name).pipe(
+                map(_ => reset(eventConsumer), share()));
+
+        this.updateState(stream);
 
-        stream.subscribe(() => {
-            this.replaceEventConsumer(reset(eventConsumer));
+        return stream;
+    }
+
+    private updateState(stream: Observable<EventConsumerDto>) {
+        stream.subscribe(updated => {
+            this.replaceEventConsumer(updated);
         }, error => {
             this.dialogs.notifyError(error);
         });
-
-        return stream;
     }
 
     private replaceEventConsumer(eventConsumer: EventConsumerDto) {

src/Squidex/app/features/administration/state/users.state.spec.ts

@@ -10,7 +10,7 @@ import { IMock, It, Mock, Times } from 'typemoq';
 
 import { AuthService, DialogService } from '@app/shared';
 
-import { UsersState } from './users.state';
+import { SnapshotUser, UsersState } from './users.state';
 
 import {
     UserDto,
@@ -45,7 +45,7 @@ describe('UsersState', () => {
             .returns(() => of(new UsersDto(200, oldUsers)));
 
         usersState = new UsersState(authService.object, dialogs.object, usersService.object);
-        usersState.load().subscribe();
+        usersState.load();
     });
 
     it('should load users', () => {
@@ -60,7 +60,7 @@ describe('UsersState', () => {
     });
 
     it('should show notification on load when reload is true', () => {
-        usersState.load(true).subscribe();
+        usersState.load(true);
 
         expect().nothing();
 
@@ -68,7 +68,7 @@ describe('UsersState', () => {
     });
 
     it('should replace selected user when reloading', () => {
-        usersState.select('id1').subscribe();
+        usersState.select('id1');
 
         const newUsers = [
             new UserDto('id1', 'mail1@mail.de_new', 'name1_new', ['Permission1_New'], false),
@@ -78,19 +78,19 @@ describe('UsersState', () => {
         usersService.setup(x => x.getUsers(10, 0, undefined))
             .returns(() => of(new UsersDto(200, newUsers)));
 
-        usersState.load().subscribe();
+        usersState.load();
 
         expect(usersState.snapshot.selectedUser).toEqual({ isCurrentUser: false, user: newUsers[0] });
     });
 
     it('should return user on select and not load when already loaded', () => {
-        let selectedUser: UserDto;
+        let selectedUser: SnapshotUser;
 
         usersState.select('id1').subscribe(x => {
             selectedUser = x!;
         });
 
-        expect(selectedUser!).toEqual(oldUsers[0]);
+        expect(selectedUser!.user).toEqual(oldUsers[0]);
         expect(usersState.snapshot.selectedUser).toEqual({ isCurrentUser: false, user: oldUsers[0] });
 
         usersService.verify(x => x.getUser(It.isAnyString()), Times.never());
@@ -100,20 +100,20 @@ describe('UsersState', () => {
         usersService.setup(x => x.getUser('id3'))
             .returns(() => of(newUser));
 
-        let selectedUser: UserDto;
+        let selectedUser: SnapshotUser;
 
         usersState.select('id3').subscribe(x => {
             selectedUser = x!;
         });
 
-        expect(selectedUser!).toEqual(newUser);
+        expect(selectedUser!.user).toEqual(newUser);
         expect(usersState.snapshot.selectedUser).toEqual({ isCurrentUser: false, user: newUser });
 
         usersService.verify(x => x.getUser('id3'), Times.once());
     });
 
     it('should return null on select when unselecting user', () => {
-        let selectedUser: UserDto;
+        let selectedUser: SnapshotUser;
 
         usersState.select(null).subscribe(x => {
             selectedUser = x!;
@@ -129,11 +129,11 @@ describe('UsersState', () => {
         usersService.setup(x => x.getUser('unknown'))
             .returns(() => throwError({}));
 
-        let selectedUser: UserDto;
+        let selectedUser: SnapshotUser;
 
         usersState.select('unknown').subscribe(x => {
             selectedUser = x!;
-        }).unsubscribe();
+        });
 
         expect(selectedUser!).toBeNull();
         expect(usersState.snapshot.selectedUser).toBeNull();
@@ -143,8 +143,8 @@ describe('UsersState', () => {
         usersService.setup(x => x.lockUser('id1'))
             .returns(() => of({}));
 
-        usersState.select('id1').subscribe();
-        usersState.lock(oldUsers[0]).subscribe();
+        usersState.select('id1');
+        usersState.lock(oldUsers[0]);
 
         const user_1 = usersState.snapshot.users.at(0);
 
@@ -156,8 +156,8 @@ describe('UsersState', () => {
         usersService.setup(x => x.unlockUser('id2'))
             .returns(() => of({}));
 
-        usersState.select('id2').subscribe();
-        usersState.unlock(oldUsers[1]).subscribe();
+        usersState.select('id2');
+        usersState.unlock(oldUsers[1]);
 
         const user_1 = usersState.snapshot.users.at(1);
 
@@ -171,8 +171,8 @@ describe('UsersState', () => {
         usersService.setup(x => x.putUser('id1', request))
             .returns(() => of({}));
 
-        usersState.select('id1').subscribe();
-        usersState.update(oldUsers[0], request).subscribe();
+        usersState.select('id1');
+        usersState.update(oldUsers[0], request);
 
         const user_1 = usersState.snapshot.users.at(0);
 
@@ -188,7 +188,7 @@ describe('UsersState', () => {
         usersService.setup(x => x.postUser(request))
             .returns(() => of(newUser));
 
-        usersState.create(request).subscribe();
+        usersState.create(request);
 
         expect(usersState.snapshot.users.values).toEqual([
             { isCurrentUser: false, user: newUser },
@@ -202,8 +202,8 @@ describe('UsersState', () => {
         usersService.setup(x => x.getUsers(10, 10, undefined))
             .returns(() => of(new UsersDto(200, [])));
 
-        usersState.goNext().subscribe();
-        usersState.goPrev().subscribe();
+        usersState.goNext();
+        usersState.goPrev();
 
         expect().nothing();
 
@@ -215,7 +215,7 @@ describe('UsersState', () => {
         usersService.setup(x => x.getUsers(10, 0, 'my-query'))
             .returns(() => of(new UsersDto(0, [])));
 
-        usersState.search('my-query').subscribe();
+        usersState.search('my-query');
 
         expect(usersState.snapshot.usersQuery).toEqual('my-query');
 

src/Squidex/app/features/administration/state/users.state.ts

@@ -7,15 +7,15 @@
 
 import { Injectable } from '@angular/core';
 import { Observable, of } from 'rxjs';
-import { catchError, distinctUntilChanged, map, switchMap, tap } from 'rxjs/operators';
+import { catchError, distinctUntilChanged, map, share, switchMap } from 'rxjs/operators';
 
 import '@app/framework/utils/rxjs-extensions';
 
 import {
+    array,
     AuthService,
     DialogService,
     ImmutableArray,
-    notify,
     Pager,
     State
 } from '@app/shared';
@@ -27,7 +27,7 @@ import {
     UsersService
 } from './../services/users.service';
 
-interface SnapshotUser {
+export interface SnapshotUser {
     // The user.
     user: UserDto;
 
@@ -37,7 +37,7 @@ interface SnapshotUser {
 
 interface Snapshot {
     // The current users.
-    users: ImmutableArray<SnapshotUser>;
+    users: UsersList;
 
     // The pagination information.
     usersPager: Pager;
@@ -52,6 +52,9 @@ interface Snapshot {
     selectedUser?: SnapshotUser | null;
 }
 
+export type UsersList = ImmutableArray<SnapshotUser>;
+export type UsersResult = { total: number, users: UsersList };
+
 @Injectable()
 export class UsersState extends State<Snapshot> {
     public users =
@@ -78,25 +81,28 @@ export class UsersState extends State<Snapshot> {
         super({ users: ImmutableArray.empty(), usersPager: new Pager(0) });
     }
 
-    public select(id: string | null): Observable<UserDto | null> {
-        return this.loadUser(id).pipe(
-            tap(selectedUser => {
-                this.next(s => ({ ...s, selectedUser }));
-            }),
-            map(x => x && x.user));
+    public select(id: string | null): Observable<SnapshotUser | null> {
+        const stream = this.loadUser(id).pipe(share());
+
+        stream.subscribe(selectedUser => {
+            this.next(s => ({ ...s, selectedUser }));
+        });
+
+        return stream;
     }
 
     private loadUser(id: string | null) {
-        return !id ?
-            of(null) :
-            of(this.snapshot.users.find(x => x.user.id === id)).pipe(
-                switchMap(user => {
-                    if (!user) {
-                        return this.usersService.getUser(id).pipe(map(x => this.createUser(x)), catchError(() => of(null)));
-                    } else {
-                        return of(user);
-                    }
-                }));
+        if (!id) {
+            return of(null);
+        }
+
+        const found = this.snapshot.users.find(x => x.user.id === id);
+
+        if (found) {
+            return of(found);
+        }
+
+        return this.usersService.getUser(id).pipe(map(x => this.createUser(x)), catchError(() => of(null)));
     }
 
     public load(isReload = false): Observable<any> {
@@ -107,80 +113,96 @@ export class UsersState extends State<Snapshot> {
         return this.loadInternal(isReload);
     }
 
-    private loadInternal(isReload = false): Observable<any> {
-        return this.usersService.getUsers(
+    private loadInternal(isReload = false): Observable<UsersResult> {
+        const stream =
+            this.usersService.getUsers(
                 this.snapshot.usersPager.pageSize,
                 this.snapshot.usersPager.skip,
                 this.snapshot.usersQuery).pipe(
-            tap(dtos => {
-                if (isReload) {
-                    this.dialogs.notifyInfo('Users reloaded.');
-                }
+                map(({ total, items }) => ({ total, users: array(items.map(x => this.createUser(x))) })), share());
 
-                this.next(s => {
-                    const users = ImmutableArray.of(dtos.items.map(x => this.createUser(x)));
-                    const usersPager = s.usersPager.setCount(dtos.total);
+        stream.subscribe(({ total, users }) => {
+            if (isReload) {
+                this.dialogs.notifyInfo('Users reloaded.');
+            }
 
-                    let selectedUser = s.selectedUser;
+            this.next(s => {
+                const usersPager = s.usersPager.setCount(total);
 
-                    if (selectedUser) {
-                        selectedUser = users.find(x => x.user.id === selectedUser!.user.id) || selectedUser;
-                    }
+                let selectedUser = s.selectedUser;
 
-                    return { ...s, users, usersPager, selectedUser, isLoaded: true };
-                });
-            }),
-            notify(this.dialogs));
+                if (selectedUser) {
+                    selectedUser = users.find(x => x.user.id === selectedUser!.user.id) || selectedUser;
+                }
+
+                return { ...s, users, usersPager, selectedUser, isLoaded: true };
+            });
+
+        }, error => {
+            this.dialogs.notifyError(error);
+        });
+
+        return stream;
     }
 
     public create(request: CreateUserDto): Observable<UserDto> {
-        return this.usersService.postUser(request).pipe(
-            tap(dto => {
-                this.next(s => {
-                    const users = s.users.pushFront(this.createUser(dto));
-                    const usersPager = s.usersPager.incrementCount();
+        const stream = this.usersService.postUser(request).pipe(share());
+
+        stream.subscribe(dto => {
+            this.next(s => {
+                const users = s.users.pushFront(this.createUser(dto));
+                const usersPager = s.usersPager.incrementCount();
+
+                return { ...s, users, usersPager };
+            });
+        });
 
-                    return { ...s, users, usersPager };
-                });
-            }));
+        return stream;
     }
 
-    public update(user: UserDto, request: UpdateUserDto): Observable<any> {
-        return this.usersService.putUser(user.id, request).pipe(
-            tap(() => {
-                this.replaceUser(update(user, request));
-            }));
+    public update(user: UserDto, request: UpdateUserDto): Observable<UserDto> {
+        const stream =
+            this.usersService.putUser(user.id, request).pipe(
+                map(_ => update(user, request)), share());
+
+        this.updateState(stream, false);
+
+        return stream;
     }
 
-    public lock(user: UserDto): Observable<any> {
-        return this.usersService.lockUser(user.id).pipe(
-            tap(() => {
-                this.replaceUser(setLocked(user, true));
-            }),
-            notify(this.dialogs));
+    public lock(user: UserDto): Observable<UserDto> {
+        const stream =
+            this.usersService.lockUser(user.id).pipe(
+                map(_ => setLocked(user, true)), share());
+
+        this.updateState(stream, true);
+
+        return stream;
     }
 
-    public unlock(user: UserDto): Observable<any> {
-        return this.usersService.unlockUser(user.id).pipe(
-            tap(() => {
-                this.replaceUser(setLocked(user, false));
-            }),
-            notify(this.dialogs));
+    public unlock(user: UserDto): Observable<UserDto> {
+        const stream =
+            this.usersService.unlockUser(user.id).pipe(
+                map(_ => setLocked(user, false)), share());
+
+        this.updateState(stream, true);
+
+        return stream;
     }
 
-    public search(query: string): Observable<any> {
+    public search(query: string): Observable<UsersResult> {
         this.next(s => ({ ...s, usersPager: new Pager(0), usersQuery: query }));
 
         return this.loadInternal();
     }
 
-    public goNext(): Observable<any> {
+    public goNext(): Observable<UsersResult> {
         this.next(s => ({ ...s, usersPager: s.usersPager.goNext() }));
 
         return this.loadInternal();
     }
 
-    public goPrev(): Observable<any> {
+    public goPrev(): Observable<UsersResult> {
         this.next(s => ({ ...s, usersPager: s.usersPager.goPrev() }));
 
         return this.loadInternal();
@@ -190,7 +212,11 @@ export class UsersState extends State<Snapshot> {
         return this.next(s => {
             const users = s.users.map(u => u.user.id === user.id ? this.createUser(user, u) : u);
 
-            const selectedUser = s.selectedUser && s.selectedUser.user.id === user.id ? users.find(x => x.user.id === user.id) : s.selectedUser;
+            const selectedUser =
+                s.selectedUser &&
+                s.selectedUser.user.id !== user.id ?
+                s.selectedUser :
+                users.find(x => x.user.id === user.id);
 
             return { ...s, users, selectedUser };
         });
@@ -200,6 +226,16 @@ export class UsersState extends State<Snapshot> {
         return this.authState.user!.id;
     }
 
+    private updateState(stream: Observable<UserDto>, notify: boolean) {
+        stream.subscribe(dto => {
+            this.replaceUser(dto);
+        }, error => {
+            if (notify) {
+                this.dialogs.notifyError(error);
+            }
+        });
+    }
+
     private createUser(user: UserDto, current?: SnapshotUser): SnapshotUser {
         if (!user) {
             return null!;

src/Squidex/app/framework/utils/immutable-array.ts

@@ -17,6 +17,11 @@ function freeze<T>(items: T[]): T[] {
     return items;
 }
 
+
+export function array<V>(items?: V[]): ImmutableArray<V> {
+    return ImmutableArray.of(items);
+}
+
 export class ImmutableArray<T> implements Iterable<T> {
     private static readonly EMPTY = new ImmutableArray<any>([]);
     private readonly items: T[];