Merge pull request #446 from GetStream/fix-network-error-during-channel-init

fix: Can't recover from error during channel init

Author: szuperaz

projects/stream-chat-angular/src/lib/channel-list/channel-list.component.spec.ts

@@ -82,20 +82,29 @@ describe('ChannelListComponent', () => {
   it('should display error indicator, if error happened', () => {
     expect(queryChatdownContainer()).toBeNull();
 
-    channelServiceMock.channels$.error(new Error('error'));
+    channelServiceMock.channelQueryState$.next({
+      state: 'error',
+      error: new Error('error'),
+    });
     fixture.detectChanges();
 
     expect(queryChatdownContainer()).not.toBeNull();
   });
 
   it('should display loading indicator, if loading', () => {
+    channelServiceMock.channelQueryState$.next({
+      state: 'in-progress',
+    });
     fixture.detectChanges();
 
     expect(queryChatdownContainer()).toBeNull();
     expect(queryLoadingIndicator()).not.toBeNull();
 
     const channels = generateMockChannels();
     channelServiceMock.channels$.next(channels);
+    channelServiceMock.channelQueryState$.next({
+      state: 'success',
+    });
     fixture.detectChanges();
 
     expect(queryLoadingIndicator()).toBeNull();

projects/stream-chat-angular/src/lib/channel-list/channel-list.component.ts

@@ -6,8 +6,8 @@ import {
   TemplateRef,
   ViewChild,
 } from '@angular/core';
-import { Observable, of, Subscription } from 'rxjs';
-import { catchError, map, startWith } from 'rxjs/operators';
+import { Observable, Subscription } from 'rxjs';
+import { map } from 'rxjs/operators';
 import { Channel } from 'stream-chat';
 import { ChannelService } from '../channel.service';
 import { CustomTemplatesService } from '../custom-templates.service';
@@ -45,14 +45,11 @@ export class ChannelListComponent implements AfterViewInit, OnDestroy {
     this.isOpen$ = this.channelListToggleService.isOpen$;
     this.channels$ = this.channelService.channels$;
     this.hasMoreChannels$ = this.channelService.hasMoreChannels$;
-    this.isError$ = this.channels$.pipe(
-      map(() => false),
-      catchError(() => of(true)),
-      startWith(false)
+    this.isError$ = this.channelService.channelQueryState$.pipe(
+      map((s) => !this.isLoadingMoreChannels && s?.state === 'error')
     );
-    this.isInitializing$ = this.channels$.pipe(
-      map((channels) => !channels),
-      catchError(() => of(false))
+    this.isInitializing$ = this.channelService.channelQueryState$.pipe(
+      map((s) => !this.isLoadingMoreChannels && s?.state === 'in-progress')
     );
     this.subscriptions.push(
       this.customTemplatesService.channelPreviewTemplate$.subscribe(

projects/stream-chat-angular/src/lib/channel.service.spec.ts

@@ -44,7 +44,7 @@ describe('ChannelService', () => {
   let init: (
     c?: Channel<DefaultStreamChatGenerics>[],
     sort?: ChannelSort<DefaultStreamChatGenerics>,
-    options?: ChannelOptions,
+    options?: ChannelOptions & { keepAliveChannels$OnError?: boolean },
     mockChannelQuery?: Function,
     shouldSetActiveChannel?: boolean
   ) => Promise<Channel<DefaultStreamChatGenerics>[]>;
@@ -84,7 +84,7 @@ describe('ChannelService', () => {
     init = async (
       channels?: Channel<DefaultStreamChatGenerics>[],
       sort?: ChannelSort<DefaultStreamChatGenerics>,
-      options?: ChannelOptions,
+      options?: ChannelOptions & { keepAliveChannels$OnError?: boolean },
       mockChannelQuery?: Function,
       shouldSetActiveChannel?: boolean
     ) => {
@@ -118,7 +118,7 @@ describe('ChannelService', () => {
     expect(mockChatClient.queryChannels).toHaveBeenCalledWith(
       jasmine.any(Object),
       jasmine.any(Object),
-      options
+      jasmine.objectContaining(options)
     );
   });
 
@@ -179,6 +179,70 @@ describe('ChannelService', () => {
     );
   });
 
+  it('should handle errors during channel load', fakeAsync(() => {
+    const spy = jasmine.createSpy();
+    service.channels$.subscribe(spy);
+    const activeChannelSpy = jasmine.createSpy();
+    service.activeChannel$.subscribe(activeChannelSpy);
+
+    try {
+      void init(undefined, undefined, { keepAliveChannels$OnError: true }, () =>
+        mockChatClient.queryChannels.and.rejectWith('there was an error')
+      );
+      tick();
+      // eslint-disable-next-line no-empty
+    } catch (error) {}
+
+    const channels = generateMockChannels();
+    mockChatClient.queryChannels.and.resolveTo(channels);
+    spy.calls.reset();
+    activeChannelSpy.calls.reset();
+    const notificationService = TestBed.inject(NotificationService);
+    const notificationSpy = jasmine.createSpy();
+    notificationService.notifications$.subscribe(notificationSpy);
+    notificationSpy.calls.reset();
+    events$.next({ eventType: 'connection.recovered' } as ClientEvent);
+
+    tick();
+
+    expect(spy).toHaveBeenCalledWith(channels);
+    expect(activeChannelSpy).toHaveBeenCalledWith(channels[0]);
+    expect(notificationSpy).toHaveBeenCalledWith([]);
+  }));
+
+  it('should emit channel query state correctly', async () => {
+    const spy = jasmine.createSpy();
+    service.channelQueryState$.subscribe(spy);
+
+    await init();
+
+    let calls = spy.calls.all();
+
+    /* eslint-disable  @typescript-eslint/no-unsafe-member-access */
+    expect(calls[1].args[0].state).toBe('in-progress');
+
+    expect(calls[2].args[0].state).toBe('success');
+
+    spy.calls.reset();
+
+    await expectAsync(
+      init(undefined, undefined, undefined, () =>
+        mockChatClient.queryChannels.and.rejectWith('there was an error')
+      )
+    ).toBeRejected();
+
+    calls = spy.calls.all();
+
+    expect(calls[0].args[0]?.state).toBe('in-progress');
+
+    expect(calls[1].args[0]).toEqual({
+      state: 'error',
+      error: 'there was an error',
+    });
+
+    /* eslint-enable  @typescript-eslint/no-unsafe-member-access */
+  });
+
   it('should not set active channel if #shouldSetActiveChannel is false', async () => {
     const activeChannelSpy = jasmine.createSpy();
     service.activeChannel$.subscribe(activeChannelSpy);
@@ -204,13 +268,16 @@ describe('ChannelService', () => {
     service.jumpToMessage$.subscribe(jumpToMessageSpy);
     const pinnedMessagesSpy = jasmine.createSpy();
     service.activeChannelPinnedMessages$.subscribe(pinnedMessagesSpy);
+    const channelsQueryStateSpy = jasmine.createSpy();
+    service.channelQueryState$.subscribe(channelsQueryStateSpy);
     messagesSpy.calls.reset();
     activeChannelSpy.calls.reset();
     channelsSpy.calls.reset();
     messageToQuoteSpy.calls.reset();
     latestMessagesSpy.calls.reset();
     jumpToMessageSpy.calls.reset();
     pinnedMessagesSpy.calls.reset();
+    channelsQueryStateSpy.calls.reset();
     service.reset();
 
     expect(messagesSpy).toHaveBeenCalledWith([]);
@@ -219,6 +286,7 @@ describe('ChannelService', () => {
     expect(messageToQuoteSpy).toHaveBeenCalledWith(undefined);
     expect(latestMessagesSpy).toHaveBeenCalledWith({});
     expect(pinnedMessagesSpy).toHaveBeenCalledWith([]);
+    expect(channelsQueryStateSpy).toHaveBeenCalledWith(undefined);
 
     channelsSpy.calls.reset();
     events$.next({

projects/stream-chat-angular/src/lib/channel.service.ts

@@ -29,6 +29,7 @@ import { NotificationService } from './notification.service';
 import { getReadBy } from './read-by';
 import {
   AttachmentUpload,
+  ChannelQueryState,
   DefaultStreamChatGenerics,
   MessageReactionType,
   StreamMessage,
@@ -73,6 +74,10 @@ export class ChannelService<
    *  Our platform documentation covers the topic of [channel events](https://getstream.io/chat/docs/javascript/event_object/?language=javascript#events) in depth.
    */
   channels$: Observable<Channel<T>[] | undefined>;
+  /**
+   * The result of the latest channel query request.
+   */
+  channelQueryState$: Observable<ChannelQueryState | undefined>;
   /**
    * Emits the currently active channel.
    *
@@ -264,7 +269,9 @@ export class ChannelService<
   }>({});
   private filters: ChannelFilters<T> | undefined;
   private sort: ChannelSort<T> | undefined;
-  private options: ChannelOptions | undefined;
+  private options:
+    | (ChannelOptions & { keepAliveChannels$OnError?: boolean })
+    | undefined;
   private readonly messagePageSize = 25;
   private messageToQuoteSubject = new BehaviorSubject<
     StreamMessage<T> | undefined
@@ -279,6 +286,9 @@ export class ChannelService<
   private shouldSetActiveChannel: boolean | undefined;
   private clientEventsSubscription: Subscription | undefined;
   private isStateRecoveryInProgress = false;
+  private channelQueryStateSubject = new BehaviorSubject<
+    ChannelQueryState | undefined
+  >(undefined);
 
   private channelListSetter = (
     channels: (Channel<T> | ChannelResponse<T>)[]
@@ -305,6 +315,7 @@ export class ChannelService<
   private parentMessageSetter = (message: StreamMessage<T> | undefined) => {
     this.activeParentMessageIdSubject.next(message?.id);
   };
+  private dismissErrorNotification?: Function;
 
   constructor(
     private chatClientService: ChatClientService<T>,
@@ -366,6 +377,7 @@ export class ChannelService<
       this.latestMessageDateByUserByChannelsSubject.asObservable();
     this.activeChannelPinnedMessages$ =
       this.activeChannelPinnedMessagesSubject.asObservable();
+    this.channelQueryState$ = this.channelQueryStateSubject.asObservable();
   }
 
   /**
@@ -554,31 +566,33 @@ export class ChannelService<
   async init(
     filters: ChannelFilters<T>,
     sort?: ChannelSort<T>,
-    options?: ChannelOptions,
+    options?: ChannelOptions & { keepAliveChannels$OnError?: boolean },
     shouldSetActiveChannel: boolean = true
   ) {
     this.filters = filters;
-    this.options = options || {
+    this.options = {
       offset: 0,
       limit: 25,
       state: true,
       presence: true,
       watch: true,
       message_limit: this.messagePageSize,
+      ...options,
     };
     this.sort = sort || { last_message_at: -1, updated_at: -1 };
     this.shouldSetActiveChannel = shouldSetActiveChannel;
+    this.clientEventsSubscription = this.chatClientService.events$.subscribe(
+      (notification) => void this.handleNotification(notification)
+    );
     try {
       const result = await this.queryChannels(this.shouldSetActiveChannel);
-      this.clientEventsSubscription = this.chatClientService.events$.subscribe(
-        (notification) => void this.handleNotification(notification)
-      );
       return result;
     } catch (error) {
-      this.notificationService.addPermanentNotification(
-        'streamChat.Error loading channels',
-        'error'
-      );
+      this.dismissErrorNotification =
+        this.notificationService.addPermanentNotification(
+          'streamChat.Error loading channels',
+          'error'
+        );
       throw error;
     }
   }
@@ -589,7 +603,10 @@ export class ChannelService<
   reset() {
     this.deselectActiveChannel();
     this.channelsSubject.next(undefined);
+    this.channelQueryStateSubject.next(undefined);
     this.clientEventsSubscription?.unsubscribe();
+    this.dismissErrorNotification?.();
+    this.dismissErrorNotification = undefined;
   }
 
   /**
@@ -973,7 +990,11 @@ export class ChannelService<
             if (this.options) {
               this.options.offset = 0;
             }
-            await this.queryChannels(false, true);
+            // If channel list is not inited, we set the active channel
+            const shoulSetActiveChannel =
+              this.shouldSetActiveChannel &&
+              !this.activeChannelSubject.getValue();
+            await this.queryChannels(shoulSetActiveChannel || false, true);
             // Thread messages are not refetched so active thread gets deselected to avoid displaying stale messages
             void this.setAsActiveParentMessage(undefined);
             this.isStateRecoveryInProgress = false;
@@ -1264,6 +1285,7 @@ export class ChannelService<
     recoverState = false
   ) {
     try {
+      this.channelQueryStateSubject.next({ state: 'in-progress' });
       const channels = await this.chatClientService.chatClient.queryChannels(
         this.filters!,
         this.sort || {},
@@ -1274,13 +1296,14 @@ export class ChannelService<
         ? []
         : this.channelsSubject.getValue() || [];
       this.channelsSubject.next([...prevChannels, ...channels]);
-      const currentActiveChannel = this.activeChannelSubject.getValue();
+      let currentActiveChannel = this.activeChannelSubject.getValue();
       if (
         channels.length > 0 &&
         !currentActiveChannel &&
         shouldSetActiveChannel
       ) {
         this.setAsActiveChannel(channels[0]);
+        currentActiveChannel = this.activeChannelSubject.getValue();
       }
       if (
         recoverState &&
@@ -1289,9 +1312,23 @@ export class ChannelService<
         this.deselectActiveChannel();
       }
       this.hasMoreChannelsSubject.next(channels.length >= this.options!.limit!);
+      this.channelQueryStateSubject.next({ state: 'success' });
+      if (
+        this.options?.keepAliveChannels$OnError &&
+        this.dismissErrorNotification
+      ) {
+        this.dismissErrorNotification();
+      }
       return channels;
     } catch (error) {
-      this.channelsSubject.error(error);
+      if (!this.options?.keepAliveChannels$OnError) {
+        this.channelsSubject.error(error);
+      }
+      this.channelQueryStateSubject.next({
+        state: 'error',
+        // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+        error,
+      });
       throw error;
     }
   }

projects/stream-chat-angular/src/lib/channel/channel.component.ts

@@ -1,6 +1,6 @@
 import { Component } from '@angular/core';
-import { Observable, of, Subscription } from 'rxjs';
-import { catchError, map, startWith } from 'rxjs/operators';
+import { combineLatest, Observable, Subscription } from 'rxjs';
+import { map } from 'rxjs/operators';
 import { ChannelService } from '../channel.service';
 import { ThemeService } from '../theme.service';
 
@@ -24,14 +24,21 @@ export class ChannelComponent {
     private channelService: ChannelService,
     private themeService: ThemeService
   ) {
-    this.isError$ = this.channelService.channels$.pipe(
-      map(() => false),
-      catchError(() => of(true)),
-      startWith(false)
+    this.isError$ = combineLatest([
+      this.channelService.channelQueryState$,
+      this.channelService.activeChannel$,
+    ]).pipe(
+      map(([state, activeChannel]) => {
+        return !activeChannel && state?.state === 'error';
+      })
     );
-    this.isInitializing$ = this.channelService.channels$.pipe(
-      map((channels) => !channels),
-      catchError(() => of(false))
+    this.isInitializing$ = combineLatest([
+      this.channelService.channelQueryState$,
+      this.channelService.activeChannel$,
+    ]).pipe(
+      map(([state, activeChannel]) => {
+        return !activeChannel && state?.state === 'in-progress';
+      })
     );
     this.isActiveThread$ = this.channelService.activeParentMessageId$.pipe(
       map((id) => !!id)