RxJS Subscription Patterns

Overview

This guide documents the approved patterns for managing RxJS subscriptions in the CometChat Angular V5 UIKit. Following these patterns ensures that subscriptions are automatically cleaned up when components and services are destroyed, preventing memory leaks in long-running sessions.

✅ Approved Patterns

1. `takeUntilDestroyed()` — Primary Pattern

The preferred pattern for all new subscriptions. Uses Angular’s DestroyRef to automatically unsubscribe when the component or service is destroyed.

import { Component, inject, DestroyRef } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { CometChatMessageEvents } from '../events/CometChatMessageEvents';

@Component({ ... })
export class MyComponent {
  private readonly destroyRef = inject(DestroyRef);

  constructor() {
    CometChatMessageEvents.ccMessageRead
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe(message => {
        // handle message read
      });
  }
}

Why this is preferred:

No boilerplate ngOnDestroy needed for subscription cleanup
Works in both components and services
Angular manages the lifecycle automatically via DestroyRef
Composable with other RxJS operators

2. `toSignal()` — For Service Observables

Use toSignal() when you need to consume a service observable as an Angular Signal in a component. The subscription is automatically cleaned up when the component is destroyed.

import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { ConversationsService } from '../services/conversations.service';

@Component({ ... })
export class MyComponent {
  private conversationsService = inject(ConversationsService);

  // Automatically subscribes and unsubscribes — no manual cleanup needed
  conversations = toSignal(this.conversationsService.conversations$, {
    initialValue: [] as CometChat.Conversation[],
  });
}

Why this is preferred for service state:

Zero subscription management code
Integrates with Angular’s change detection (OnPush compatible)
Automatically handles initial value and completion

3. CometChat SDK Listeners — Service-Managed Pattern

CometChat SDK listeners (addMessageListener, addUserListener, etc.) must be registered and removed via the service layer, not directly in components.

// ✅ CORRECT — service manages SDK listener lifecycle
@Injectable({ providedIn: 'root' })
export class ConversationsService {
  private readonly listenerId = `conversations_${Date.now()}`;

  private setupMessageListener(): void {
    CometChat.addMessageListener(
      this.listenerId,
      new CometChat.MessageListener({
        onTextMessageReceived: (message) => { /* handle */ },
        onMediaMessageReceived: (message) => { /* handle */ },
      })
    );
  }

  private removeMessageListener(): void {
    try {
      CometChat.removeMessageListener(this.listenerId);
    } catch (error) {
      // log but don't throw — cleanup errors are non-critical
    }
  }

  cleanup(): void {
    this.removeMessageListener();
    // ... remove other listeners
  }
}

Why services manage SDK listeners:

SDK listeners are singleton-scoped — they must persist across component destroy/recreate cycles (e.g., tab switching)
Components should never call cleanup() in ngOnDestroy — this would tear down listeners on every navigation
Call cleanup() only for application-level events: user logout, account switching

❌ Anti-Patterns to Avoid

❌ Manual `destroy$` Subject

// ❌ DO NOT USE — verbose boilerplate, error-prone
@Component({ ... })
export class MyComponent implements OnDestroy {
  private destroy$ = new Subject<void>();

  constructor() {
    someObservable$
      .pipe(takeUntil(this.destroy$))
      .subscribe(value => { /* handle */ });
  }

  ngOnDestroy(): void {
    this.destroy$.next();    // easy to forget
    this.destroy$.complete(); // easy to forget
  }
}

Problem: Requires manual ngOnDestroy implementation. If next() or complete() is forgotten, subscriptions leak. Fix: Use takeUntilDestroyed(this.destroyRef) instead.

❌ Manual `Subscription.unsubscribe()`

// ❌ DO NOT USE — manual tracking is error-prone
@Component({ ... })
export class MyComponent implements OnDestroy {
  private subscription: Subscription | null = null;

  constructor() {
    this.subscription = someObservable$.subscribe(value => { /* handle */ });
  }

  ngOnDestroy(): void {
    this.subscription?.unsubscribe(); // easy to miss for multiple subscriptions
  }
}

Problem: Each subscription requires a separate property and manual cleanup. Scales poorly. Fix: Use takeUntilDestroyed(this.destroyRef) — no property or ngOnDestroy needed.

❌ Subscribing Without Any Cleanup

// ❌ MEMORY LEAK — subscription never cleaned up
@Component({ ... })
export class MyComponent {
  constructor() {
    someObservable$.subscribe(value => {
      // This subscription lives forever!
    });
  }
}

Problem: The subscription persists after the component is destroyed, accumulating over time. Fix: Always use takeUntilDestroyed(this.destroyRef) or toSignal().

Migration Guide

If you encounter the old takeUntil(this.destroy$) pattern, here is how to migrate:

Before

import { Component, OnDestroy } from '@angular/core';
import { Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

@Component({ ... })
export class MyComponent implements OnDestroy {
  private destroy$ = new Subject<void>();

  constructor() {
    someObservable$
      .pipe(takeUntil(this.destroy$))
      .subscribe(value => { /* handle */ });
  }

  ngOnDestroy(): void {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

After

import { Component, inject, DestroyRef } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

@Component({ ... })
export class MyComponent {
  private readonly destroyRef = inject(DestroyRef);

  constructor() {
    someObservable$
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe(value => { /* handle */ });
  }
  // No ngOnDestroy needed!
}

Steps:

Add DestroyRef to @angular/core imports
Add takeUntilDestroyed to @angular/core/rxjs-interop imports
Add private readonly destroyRef = inject(DestroyRef) field
Replace .pipe(takeUntil(this.destroy$)) with .pipe(takeUntilDestroyed(this.destroyRef))
Remove private destroy$ = new Subject<void>()
Remove this.destroy$.next() and this.destroy$.complete() from ngOnDestroy
Remove OnDestroy interface if no other cleanup is needed
Remove unused Subject and takeUntil imports

Summary

Pattern	Use Case	Cleanup
`takeUntilDestroyed(destroyRef)`	Any RxJS subscription in component/service	Automatic via DestroyRef
`toSignal(obs$)`	Service observable → Signal in component	Automatic via DestroyRef
Service-managed SDK listeners	CometChat SDK `addXxxListener`	Manual via `cleanup()` on logout
~~`takeUntil(destroy$)`~~	❌ Deprecated	Manual — avoid
~~`subscription.unsubscribe()`~~	❌ Deprecated	Manual — avoid

​Overview

​✅ Approved Patterns

​1. takeUntilDestroyed() — Primary Pattern

​2. toSignal() — For Service Observables

​3. CometChat SDK Listeners — Service-Managed Pattern

​❌ Anti-Patterns to Avoid

​❌ Manual destroy$ Subject

​❌ Manual Subscription.unsubscribe()

​❌ Subscribing Without Any Cleanup

​Migration Guide

​Before

​After

​Summary