Merge pull request #390 from GetStream/fix-reconnect

feat: Enhance reconnect #350

Author: szuperaz

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

@@ -1702,35 +1702,35 @@ describe('ChannelService', () => {
 
   it('should reset state after connection recovered', async () => {
     await init();
-    spyOn(service, 'init');
-    spyOn(service, 'reset');
+    mockChatClient.queryChannels.calls.reset();
     events$.next({ eventType: 'connection.recovered' } as ClientEvent);
 
-    expect(service.init).toHaveBeenCalledWith(
-      service['filters']!,
-      service['sort'],
-      service['options'],
-      service['shouldSetActiveChannel']
+    expect(mockChatClient.queryChannels).toHaveBeenCalledWith(
+      jasmine.any(Object),
+      jasmine.any(Object),
+      jasmine.any(Object)
     );
-
-    expect(service.reset).toHaveBeenCalledWith();
   });
 
   it(`shouldn't do duplicate state reset after connection recovered`, async () => {
     await init();
-    spyOn(service, 'init');
-    spyOn(service, 'reset');
+    mockChatClient.queryChannels.calls.reset();
     events$.next({ eventType: 'connection.recovered' } as ClientEvent);
     events$.next({ eventType: 'connection.recovered' } as ClientEvent);
 
-    expect(service.init).toHaveBeenCalledOnceWith(
-      service['filters']!,
-      service['sort'],
-      service['options'],
-      service['shouldSetActiveChannel']
-    );
+    expect(mockChatClient.queryChannels).toHaveBeenCalledTimes(1);
+  });
 
-    expect(service.reset).toHaveBeenCalledOnceWith();
+  it('should reset pagination options after reconnect', async () => {
+    await init(undefined, undefined, { offset: 20 });
+    mockChatClient.queryChannels.calls.reset();
+    events$.next({ eventType: 'connection.recovered' } as ClientEvent);
+
+    expect(mockChatClient.queryChannels).toHaveBeenCalledWith(
+      jasmine.any(Object),
+      jasmine.any(Object),
+      jasmine.objectContaining({ offset: 0 })
+    );
   });
 
   it('should load message into state', async () => {
@@ -1823,4 +1823,39 @@ describe('ChannelService', () => {
       'Error removing message pin'
     );
   });
+
+  it('should deselect active channel if active channel is not present after state reconnect', fakeAsync(async () => {
+    await init();
+    let activeChannel!: Channel<DefaultStreamChatGenerics>;
+    service.activeChannel$.subscribe((c) => (activeChannel = c!));
+    let channels!: Channel<DefaultStreamChatGenerics>[];
+    service.channels$.subscribe((c) => (channels = c!));
+    channels = channels.filter((c) => c.id !== activeChannel.id);
+    const spy = jasmine.createSpy();
+    service.activeChannel$.subscribe(spy);
+    spy.calls.reset();
+    mockChatClient.queryChannels.and.resolveTo(channels);
+    spyOn(service, 'deselectActiveChannel').and.callThrough();
+    events$.next({ eventType: 'connection.recovered' } as ClientEvent);
+    tick();
+
+    expect(spy).toHaveBeenCalledWith(undefined);
+    expect(service.deselectActiveChannel).toHaveBeenCalledWith();
+  }));
+
+  it(`shouldn't deselect active channel if active channel is present after state reconnect`, fakeAsync(async () => {
+    await init();
+    let channels!: Channel<DefaultStreamChatGenerics>[];
+    service.channels$.subscribe((c) => (channels = c!));
+    const spy = jasmine.createSpy();
+    service.activeChannel$.subscribe(spy);
+    spy.calls.reset();
+    mockChatClient.queryChannels.and.resolveTo(channels);
+    spyOn(service, 'deselectActiveChannel').and.callThrough();
+    events$.next({ eventType: 'connection.recovered' } as ClientEvent);
+    tick();
+
+    expect(spy).not.toHaveBeenCalled();
+    expect(service.deselectActiveChannel).not.toHaveBeenCalled();
+  }));
 });

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

@@ -1,4 +1,4 @@
-import { TestBed } from '@angular/core/testing';
+import { fakeAsync, TestBed, tick } from '@angular/core/testing';
 import { Subject } from 'rxjs';
 import { first } from 'rxjs/operators';
 import {
@@ -225,18 +225,19 @@ describe('ChannelService - threads', () => {
     expect(activeParentMessageSpy).toHaveBeenCalledWith(undefined);
   });
 
-  it('should deselect thread after reconnect', async () => {
+  it('should deselect thread after reconnect', fakeAsync(async () => {
     await init();
     let parentMessage!: StreamMessage;
     service.activeChannelMessages$.subscribe((m) => (parentMessage = m[0]));
     await service.setAsActiveParentMessage(parentMessage);
     const spy = jasmine.createSpy();
     service.activeParentMessage$.subscribe(spy);
     spy.calls.reset();
-    connectionState$.next('online');
+    events$.next({ eventType: 'connection.recovered' } as ClientEvent);
+    tick();
 
     expect(spy).toHaveBeenCalledWith(undefined);
-  });
+  }));
 
   it('should not add readBy field to messages', async () => {
     await init();

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

@@ -6,7 +6,7 @@ import {
   ReplaySubject,
   Subscription,
 } from 'rxjs';
-import { filter, first, map, shareReplay } from 'rxjs/operators';
+import { first, map, shareReplay } from 'rxjs/operators';
 import {
   Attachment,
   Channel,
@@ -359,12 +359,6 @@ export class ChannelService<
     this.messageToQuote$ = this.messageToQuoteSubject.asObservable();
     this.jumpToMessage$ = this.jumpToMessageSubject.asObservable();
 
-    this.chatClientService.connectionState$
-      .pipe(filter((s) => s === 'online'))
-      .subscribe(() => {
-        void this.setAsActiveParentMessage(undefined);
-      });
-
     this.usersTypingInChannel$ =
       this.usersTypingInChannelSubject.asObservable();
     this.usersTypingInThread$ = this.usersTypingInThreadSubject.asObservable();
@@ -956,14 +950,13 @@ export class ChannelService<
             return;
           }
           this.isStateRecoveryInProgress = true;
-          this.reset();
           try {
-            await this.init(
-              this.filters!,
-              this.sort,
-              this.options,
-              this.shouldSetActiveChannel
-            );
+            if (this.options) {
+              this.options.offset = 0;
+            }
+            await this.queryChannels(false, true);
+            // Thread messages are not refetched so active thread gets deselected to avoid displaying stale messages
+            void this.setAsActiveParentMessage(undefined);
             this.isStateRecoveryInProgress = false;
           } catch {
             this.isStateRecoveryInProgress = false;
@@ -1247,23 +1240,35 @@ export class ChannelService<
     this.activeChannelSubscriptions = [];
   }
 
-  private async queryChannels(shouldSetActiveChannel: boolean) {
+  private async queryChannels(
+    shouldSetActiveChannel: boolean,
+    recoverState = false
+  ) {
     try {
       const channels = await this.chatClientService.chatClient.queryChannels(
         this.filters!,
-        this.sort,
+        this.sort || {},
         this.options
       );
       channels.forEach((c) => this.watchForChannelEvents(c));
-      const prevChannels = this.channelsSubject.getValue() || [];
+      const prevChannels = recoverState
+        ? []
+        : this.channelsSubject.getValue() || [];
       this.channelsSubject.next([...prevChannels, ...channels]);
+      const currentActiveChannel = this.activeChannelSubject.getValue();
       if (
         channels.length > 0 &&
-        !this.activeChannelSubject.getValue() &&
+        !currentActiveChannel &&
         shouldSetActiveChannel
       ) {
         this.setAsActiveChannel(channels[0]);
       }
+      if (
+        recoverState &&
+        !channels.find((c) => c.cid === currentActiveChannel?.cid)
+      ) {
+        this.deselectActiveChannel();
+      }
       this.hasMoreChannelsSubject.next(channels.length >= this.options!.limit!);
       return channels;
     } catch (error) {