RxJS in Angular — Chapter 5 | Subject, BehaviorSubject & ReplaySubject — The Two-Way Radio

"Subject, BehaviorSubject & ReplaySubject — The Two-Way Radio"

---

👋 Welcome to Chapter 5!

So far, everything we've done has been one-directional: an Observable produces data and you subscribe to receive it.

But what if YOU want to push data into a stream? What if you need to broadcast data to multiple components — like a shared shopping cart?

Enter: Subject 🎙️

---

📻 The Two-Way Radio Analogy

A regular Observable is like a radio broadcast station:

• The station sends signals
• You tune in (subscribe) to receive
• You can't talk back to the station

A Subject is like a two-way radio:

• You can SEND data into it
• AND you can RECEIVE data from it
• Multiple people can listen to the same channel

This makes Subject perfect for sharing data between components and event broadcasting.

---

🔴 What is a Subject?

A Subject is both an Observable AND an Observer at the same time.

• As an Observable — you can subscribe to it
• As an Observer — you can push values into it using .next()

import { Subject } from 'rxjs';

// Create a Subject
const mySubject = new Subject<string>();

// Subscribe to it (like a regular Observable)
mySubject.subscribe(value => console.log('Subscriber A got:', value));
mySubject.subscribe(value => console.log('Subscriber B got:', value));

// Push values INTO it (this is the special part!)
mySubject.next('Hello!');
mySubject.next('How are you?');

// Output:
// Subscriber A got: Hello!
// Subscriber B got: Hello!
// Subscriber A got: How are you?
// Subscriber B got: How are you?

Both subscribers got both values! This is called multicasting — one source, many listeners.

---

⚠️ Subject's Big Limitation

A regular Subject only delivers values to subscribers that are currently listening.

If you subscribe AFTER a value was emitted, you missed it.

const subject = new Subject<number>();

subject.next(1);  // Emitted before anyone subscribed!
subject.next(2);  // Also emitted before anyone subscribed!

// Subscribe now — too late for 1 and 2
subject.subscribe(v => console.log('Got:', v));

subject.next(3);  // ✅ This one gets received

// Output: Got: 3
// 1 and 2 are gone forever

This is like tuning into a radio broadcast after a song already played. You missed it.

That's why we have BehaviorSubject and ReplaySubject.

---

🎯 BehaviorSubject — Remembers the Last Value

BehaviorSubject always remembers its latest value and gives it immediately to any new subscriber.

Think of it like a whiteboard 📋 — it always shows the most recent thing written on it, even if you just walked into the room.

Rules of BehaviorSubject:

1. Must be initialized with a starting value
2. Always has a current value (accessible via .value)
3. New subscribers get the current value immediately upon subscribing
4. Emits new values to all current subscribers

import { BehaviorSubject } from 'rxjs';

// Must provide an initial value
const userStatus = new BehaviorSubject<string>('offline');

console.log('Current value:', userStatus.value); // 'offline'

// Subscribe now
userStatus.subscribe(status => console.log('Subscriber 1:', status));
// Output immediately: Subscriber 1: offline (gets current value!)

// Update the value
userStatus.next('online');
// Output: Subscriber 1: online

// New subscriber AFTER 'online' was emitted
userStatus.subscribe(status => console.log('Subscriber 2:', status));
// Output immediately: Subscriber 2: online (gets latest value, not 'offline'!)

---

🛒 Real-World Example — Shopping Cart Service

BehaviorSubject is PERFECT for shared state like a shopping cart:

cart.service.ts

import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable } from 'rxjs';
import { map } from 'rxjs/operators';

export interface CartItem {
  id: number;
  name: string;
  price: number;
  quantity: number;
}

@Injectable({
  providedIn: 'root'
})
export class CartService {

  // BehaviorSubject: starts with empty array
  // Private so only this service can push values
  private cartItemsSubject = new BehaviorSubject<CartItem[]>([]);

  // Public Observable: components can SUBSCRIBE but NOT push values
  // This is the security pattern — expose Observable, not Subject
  cartItems$ = this.cartItemsSubject.asObservable();

  // Derived Observables — automatically calculated from cartItems$
  cartCount$ = this.cartItems$.pipe(
    map(items => items.reduce((total, item) => total + item.quantity, 0))
  );

  cartTotal$ = this.cartItems$.pipe(
    map(items => items.reduce((total, item) => total + (item.price * item.quantity), 0))
  );

  // Add item to cart
  addItem(newItem: CartItem): void {
    const currentItems = this.cartItemsSubject.value; // Get current cart

    const existingItem = currentItems.find(item => item.id === newItem.id);

    if (existingItem) {
      // Item already in cart — increase quantity
      const updated = currentItems.map(item =>
        item.id === newItem.id
          ? { ...item, quantity: item.quantity + 1 }
          : item
      );
      this.cartItemsSubject.next(updated); // Push new value!
    } else {
      // New item — add to cart
      this.cartItemsSubject.next([...currentItems, newItem]);
    }
  }

  // Remove item from cart
  removeItem(itemId: number): void {
    const updated = this.cartItemsSubject.value
      .filter(item => item.id !== itemId);
    this.cartItemsSubject.next(updated);
  }

  // Clear cart
  clearCart(): void {
    this.cartItemsSubject.next([]);
  }
}

cart-icon.component.ts (in the header)

@Component({
  selector: 'app-cart-icon',
  template: `
    <div class="cart-icon">
      🛒
      <span class="badge" *ngIf="(cartService.cartCount$ | async) as count">
        {{ count }}
      </span>
    </div>
  `
})
export class CartIconComponent {
  constructor(public cartService: CartService) {}
  // No ngOnDestroy needed — async pipe handles it!
}

cart-total.component.ts (in the checkout area)

@Component({
  selector: 'app-cart-total',
  template: `
    <div class="cart-summary">
      <p>Items: {{ cartService.cartCount$ | async }}</p>
      <p class="total">Total: ৳{{ cartService.cartTotal$ | async }}</p>
      <button (click)="cartService.clearCart()">Clear Cart 🗑️</button>
    </div>
  `
})
export class CartTotalComponent {
  constructor(public cartService: CartService) {}
}

product-card.component.ts (on product pages)

@Component({
  selector: 'app-product-card',
  template: `
    <div class="card">
      <h3>{{ product.name }}</h3>
      <p>৳{{ product.price }}</p>
      <button (click)="addToCart()">Add to Cart 🛒</button>
    </div>
  `
})
export class ProductCardComponent {
  @Input() product!: CartItem;

  constructor(private cartService: CartService) {}

  addToCart(): void {
    this.cartService.addItem(this.product);
    // The cart icon in the header updates AUTOMATICALLY!
    // The cart total updates AUTOMATICALLY!
    // BehaviorSubject broadcasts to ALL subscribers!
  }
}

When a user adds a product, the cart icon in the header and the total in the checkout area both update instantly — without any EventEmitter, @Input/@Output, or complex communication! 🎉

---

🔄 ReplaySubject — Remembers Multiple Past Values

ReplaySubject is like a DVR recording 📼 — it saves the last N values and plays them back to any new subscriber.

import { ReplaySubject } from 'rxjs';

// Remember the last 3 values
const replay = new ReplaySubject<number>(3);

replay.next(1);
replay.next(2);
replay.next(3);
replay.next(4);
replay.next(5);

// Subscribe now — gets the last 3 values replayed immediately
replay.subscribe(v => console.log('Got:', v));

// Output:
// Got: 3
// Got: 4
// Got: 5
// (Values 1 and 2 are gone, but 3, 4, 5 are replayed)

Real Use Case — Message History / Notifications

@Injectable({ providedIn: 'root' })
export class NotificationService {

  // Keep the last 5 notifications so late subscribers see recent history
  private notificationsSubject = new ReplaySubject<Notification>(5);

  notifications$ = this.notificationsSubject.asObservable();

  addNotification(message: string, type: 'success' | 'error' | 'info'): void {
    this.notificationsSubject.next({
      id: Date.now(),
      message,
      type,
      timestamp: new Date()
    });
  }
}

Even if a component subscribes after 3 notifications were emitted, it still gets those 3 notifications to display.

---

📡 AsyncSubject — Only Emits the Last Value on Complete

AsyncSubject only emits the LAST value, and ONLY when .complete() is called.

import { AsyncSubject } from 'rxjs';

const async$ = new AsyncSubject<number>();

async$.subscribe(v => console.log('Got:', v));

async$.next(1);
async$.next(2);
async$.next(3);

// Nothing logged yet...

async$.complete(); // NOW it emits!

// Output: Got: 3 (only the last value!)

This is rarely used but good to know.

---

🔐 The Security Pattern — Hide Your Subject!

A best practice: never expose a Subject directly to components. Use .asObservable() to expose a read-only view.

// ❌ BAD — any component can push data into this!
userSubject = new BehaviorSubject<User>(null);

// ✅ GOOD — only this service controls the data
private userSubject = new BehaviorSubject<User | null>(null);
user$ = this.userSubject.asObservable(); // Read-only!

// Only this service's methods can change the user
setUser(user: User): void {
  this.userSubject.next(user);
}

---

🏗️ Real-World Example — Auth Service with BehaviorSubject

This pattern is used in almost every real Angular app:

auth.service.ts

import { Injectable } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { BehaviorSubject, Observable } from 'rxjs';
import { tap, map } from 'rxjs/operators';

export interface User {
  id: number;
  name: string;
  email: string;
  role: 'admin' | 'user';
}

@Injectable({ providedIn: 'root' })
export class AuthService {

  private currentUserSubject = new BehaviorSubject<User | null>(null);

  // Read-only public streams
  currentUser$ = this.currentUserSubject.asObservable();
  isLoggedIn$ = this.currentUser$.pipe(map(user => user !== null));
  isAdmin$ = this.currentUser$.pipe(map(user => user?.role === 'admin'));

  constructor(private http: HttpClient) {
    // Restore user from localStorage on app start
    const savedUser = localStorage.getItem('currentUser');
    if (savedUser) {
      this.currentUserSubject.next(JSON.parse(savedUser));
    }
  }

  login(email: string, password: string): Observable<User> {
    return this.http.post<User>('/api/login', { email, password })
      .pipe(
        tap(user => {
          localStorage.setItem('currentUser', JSON.stringify(user));
          this.currentUserSubject.next(user); // 📡 Broadcast to all listeners!
        })
      );
  }

  logout(): void {
    localStorage.removeItem('currentUser');
    this.currentUserSubject.next(null); // 📡 Broadcast logout to all!
  }

  get currentUser(): User | null {
    return this.currentUserSubject.value; // Synchronous access
  }
}

nav.component.ts

@Component({
  template: `
    <nav>
      <a routerLink="/">Home</a>
      <a routerLink="/products">Products</a>

      <!-- Only show if logged in -->
      <a *ngIf="authService.isLoggedIn$ | async" routerLink="/profile">
        Profile
      </a>

      <!-- Only show if admin -->
      <a *ngIf="authService.isAdmin$ | async" routerLink="/admin">
        Admin Panel
      </a>

      <span *ngIf="authService.currentUser$ | async as user">
        Hello, {{ user.name }}!
      </span>

      <button *ngIf="authService.isLoggedIn$ | async" (click)="logout()">
        Logout
      </button>
    </nav>
  `
})
export class NavComponent {
  constructor(public authService: AuthService) {}

  logout(): void {
    this.authService.logout();
    // Nav updates automatically because BehaviorSubject broadcast the change!
  }
}

When the user logs in, the navigation bar, user profile page, and admin section all update automatically and simultaneously — no complex event passing needed!

---

🧠 Choosing the Right Subject

Use Subject when:

• You just need to broadcast events (button clicks, action triggers)
• Late subscribers don't need previous values
• Example: Notification popup triggers

Use BehaviorSubject when:

• Shared state that always has a current value
• New subscribers should get the current state immediately
• Example: Shopping cart, auth state, theme preference, language selection

Use ReplaySubject when:

• New subscribers need to see recent history
• Example: Chat messages, audit log, recent notifications

---

🧠 Chapter 5 Summary — What You Learned

• Subject is both Observable and Observer — you can push data into it AND subscribe to it
• Subject only delivers to current subscribers — late subscribers miss old values
• BehaviorSubject always has a current value and gives it to new subscribers immediately — perfect for shared state
• ReplaySubject remembers the last N values and replays them to late subscribers
• Use .asObservable() to hide the Subject and expose a read-only stream
• BehaviorSubject is used in almost every real Angular app for auth state, cart, settings, etc.

---

📚 Coming Up in Chapter 6...

We've covered the core building blocks. Now let's tackle a very common real-world challenge:

Error Handling — What do you do when an API fails? How do you show user-friendly errors? How do you retry?

Chapter 6 covers catchError, retry, retryWhen, and best practices for bulletproof Angular apps.

See you in Chapter 6! 🚀

---

💌 RxJS Deep Dive Newsletter Series | Chapter 5 of 10

Follow me on : Github Linkedin Threads Youtube Channel