fix: Missing unsubscriptions on ChannelService reset

Author: szuperaz

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

@@ -1,10 +1,11 @@
 import { fakeAsync, TestBed, tick } from '@angular/core/testing';
 import { Subject } from 'rxjs';
-import { first } from 'rxjs/operators';
+import { first, take } from 'rxjs/operators';
 import {
   Channel,
   ChannelMemberResponse,
   ChannelOptions,
+  ChannelResponse,
   ChannelSort,
   Event,
   SendMessageAPIResponse,
@@ -184,10 +185,24 @@ describe('ChannelService', () => {
     expect(activeChannelSpy).toHaveBeenCalledWith(undefined);
     expect(messageToQuoteSpy).toHaveBeenCalledWith(undefined);
     expect(latestMessagesSpy).toHaveBeenCalledWith({});
+
+    channelsSpy.calls.reset();
+    events$.next({
+      eventType: 'message.new',
+      event: {
+        channel: {
+          id: 'channel',
+        } as ChannelResponse<DefaultStreamChatGenerics>,
+      } as Event<DefaultStreamChatGenerics>,
+    });
+
+    expect(channelsSpy).not.toHaveBeenCalled();
   });
 
   it('should deselect active channel', async () => {
     await init();
+    let activeChannel!: Channel<DefaultStreamChatGenerics>;
+    service.activeChannel$.pipe(take(1)).subscribe((c) => (activeChannel = c!));
     const messagesSpy = jasmine.createSpy();
     service.activeChannelMessages$.subscribe(messagesSpy);
     const activeChannelSpy = jasmine.createSpy();
@@ -206,6 +221,11 @@ describe('ChannelService', () => {
     expect(activeChannelSpy).toHaveBeenCalledWith(undefined);
     expect(messageToQuoteSpy).toHaveBeenCalledWith(undefined);
     expect(latestMessagesSpy).toHaveBeenCalledWith({});
+
+    messagesSpy.calls.reset();
+    (activeChannel as MockChannel).handleEvent('message.new', mockMessage());
+
+    expect(messagesSpy).not.toHaveBeenCalled();
   });
 
   it('should tell if user #hasMoreChannels$', async () => {
@@ -1521,6 +1541,7 @@ describe('ChannelService', () => {
       service['options'],
       service['shouldSetActiveChannel']
     );
+
     expect(service.reset).toHaveBeenCalledWith();
   });
 });

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

@@ -4,6 +4,7 @@ import {
   combineLatest,
   Observable,
   ReplaySubject,
+  Subscription,
 } from 'rxjs';
 import { filter, first, map, shareReplay } from 'rxjs/operators';
 import {
@@ -259,6 +260,7 @@ export class ChannelService<
   );
   private _shouldMarkActiveChannelAsRead = true;
   private shouldSetActiveChannel: boolean | undefined;
+  private clientEventsSubscription: Subscription | undefined;
 
   private channelListSetter = (
     channels: (Channel<T> | ChannelResponse<T>)[]
@@ -393,6 +395,7 @@ export class ChannelService<
     if (!activeChannel) {
       return;
     }
+    this.stopWatchForActiveChannelEvents(activeChannel);
     this.activeChannelMessagesSubject.next([]);
     this.activeChannelSubject.next(undefined);
     this.activeParentMessageIdSubject.next(undefined);
@@ -489,7 +492,7 @@ export class ChannelService<
     this.sort = sort || { last_message_at: -1, updated_at: -1 };
     this.shouldSetActiveChannel = shouldSetActiveChannel;
     const result = await this.queryChannels(this.shouldSetActiveChannel);
-    this.chatClientService.events$.subscribe(
+    this.clientEventsSubscription = this.chatClientService.events$.subscribe(
       (notification) => void this.handleNotification(notification)
     );
     return result;
@@ -501,6 +504,7 @@ export class ChannelService<
   reset() {
     this.deselectActiveChannel();
     this.channelsSubject.next(undefined);
+    this.clientEventsSubscription?.unsubscribe();
   }
 
   /**
@@ -795,7 +799,7 @@ export class ChannelService<
       case 'connection.recovered': {
         this.ngZone.run(() => {
           this.reset();
-          this.init(
+          void this.init(
             this.filters!,
             this.sort,
             this.options,