Pub/Sub (2) - Angular + RxJS

• EventBus 서비스 기반 상태 공유
• debounceTime, distinctUntilChanged, takeUntil로 실수 방지
• OnPush 전략을 고려한 RxJS 사용
• 여러 컴포넌트가 하나의 이벤트 흐름을 공유하는 구조

 

 

예제 시나리오

상황: 검색창에서 키워드를 입력하면, 다른 컴포넌트가 해당 키워드를 받아 API 요청을 보내고 결과를 표시
단, 빠른 타이핑은 debounce 처리하고, 중복 검색은 distinctUntilChanged로 무시

 

 

 

EventBusService

// keyword-event-bus.service.ts
import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable } from 'rxjs';
import { debounceTime, distinctUntilChanged, filter } from 'rxjs/operators';

@Injectable({ providedIn: 'root' })
export class KeywordEventBusService {
  private keyword$ = new BehaviorSubject<string>(''); // 초기값 설정 가능

  // 외부에서 입력
  public publish(keyword: string): void {
    this.keyword$.next(keyword.trim());
  }

  // 외부에서 구독 (debounce, 중복제거 포함)
  public getKeywordStream(): Observable<string> {
    return this.keyword$.asObservable().pipe(
      map(kw => kw.trim()),
      filter(kw => kw.length > 0),               // 공백 제외
      debounceTime(300),                         // 300ms 딜레이
      distinctUntilChanged()                     // 이전과 같으면 무시
    );
  }
}

• @Injectable({ providedIn: 'root' })
  • Angular의 의존성 주입 시스템(DI) 에 의해 애플리케이션 전체에서 싱글턴으로 사용됨.
  • 컴포넌트들끼리 new 없이 동일한 인스턴스를 공유하게 되어 이벤트 버스 역할에 적합.
• private keyword$ = new BehaviorSubject<string>(''); 
  • BehaviorSubject는 마지막으로 발행된 값을 기억하고 있다가, 새로운 Subscriber에게 즉시 전달합니다.
  • 즉, 나중에 구독하더라도 마지막 발행된 키워드를 받을 수 있습니다.
  • 초기값을 ''로 설정하여 null 안전하고, 빈 상태에서도 .getValue() 사용 가능.
• publish(keyword: string)
  • 외부 컴포넌트(예: 입력창)에서 호출하는 이벤트 발행(Publisher) 메서드입니다.
  • 입력값은 .trim()을 적용하여 앞뒤 공백 제거 후 발행
• getKeywordStream(
  • 이 메서드는 외부 컴포넌트가 구독자(Subscriber) 로서 키워드 이벤트 스트림을 받아보는 역할입니다.
  • RxJS 연산자를 조합하여 불필요한 발행을 줄이고, UX 성능을 최적화합니다.
  •  .asObservable()
    • 내부에서 직접 .next()를 못하게 막고, 읽기 전용 스트림으로 노출
  • .map(kw => kw.trim())
    • 한 번 더 안전하게 공백 제거 (혹시 publish 쪽에서 안 한 경우 대비)
  • .filter(kw => kw.length > 0)
    • 공백 키워드 제외 → API 낭비 방지
  • .debounceTime(300)
    • 사용자가 입력 중일 때는 기다렸다가, 300ms 동안 입력이 멈추면 발행
  • .distinctUntilChanged()
    • 이전에 입력한 키워드와 같다면 무시

---

 

Publisher Component – 입력 이벤트 전파

// search-input.component.ts
import { Component } from '@angular/core';
import { KeywordEventBusService } from '../services/keyword-event-bus.service';

@Component({
  selector: 'app-search-input',
  template: `<input type="text" (input)="onInput($event)" placeholder="검색어 입력" />`
})
export class SearchInputComponent {
  constructor(private keywordBus: KeywordEventBusService) {}

  onInput(event: Event) {
    const input = event.target as HTMLInputElement;
    this.keywordBus.publish(input.value);
  }
}

 

 

• (input) 이벤트에서 전달된 DOM event 객체를 사용하여 입력값을 가져옴
  • event.target as HTMLInputElement: DOM 타입 명시
  • input.value: 현재 입력된 텍스트 값
• this.keywordBus.publish(...)로 검색어를 발행(Publish)
  • 이 검색어는 BehaviorSubject를 통해 스트림으로 전달됨
  • 구독 중인 다른 컴포넌트들은 이 값을 받아 반응함

 

---

 

Subscriber Component – 이벤트 구독 + API 호출

// search-result.component.ts
import { Component, OnDestroy } from '@angular/core';
import { Subject, switchMap, takeUntil, of } from 'rxjs';
import { KeywordEventBusService } from '../services/keyword-event-bus.service';

@Component({
  selector: 'app-search-result',
  template: `
    <p *ngIf="result">검색 결과: {{ result }}</p>
  `
})
export class SearchResultComponent implements OnDestroy {
  private destroy$ = new Subject<void>();
  public result = '';

  constructor(private keywordBus: KeywordEventBusService) {
    this.keywordBus.getKeywordStream()
      .pipe(
        takeUntil(this.destroy$),
        switchMap(keyword => this.fakeSearch(keyword)) // API 호출처럼
      )
      .subscribe(result => {
        this.result = result;
      });
  }

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

  // 실제로는 HttpClient 등을 사용
  private fakeSearch(keyword: string) {
    return of(`"${keyword}"에 대한 결과입니다.`); // 1초 대기 넣고 싶으면 delay(1000)
  }
}

 

• KeywordEventBusService에서 발행된 키워드 값을 구독
• 새로운 키워드가 들어오면, 이를 기반으로 검색 결과를 생성 (fakeSearch)
• API 요청처럼 처리되며, 메모리 누수를 막기 위해 takeUntil과 ngOnDestroy()로 구독 종료
• this.keywordBus.getKeywordStream()
• 이벤트 버스로부터 검색어 Observable 스트림 구독
• ✅ this.keywordBus.getKeywordStream()
  
  • 이벤트 버스로부터 검색어 Observable 스트림 구독
• takeUntil(this.destroy$)
  
  • 컴포넌트가 파괴되면 자동으로 구독 해제\
  • 메모리 누수를 방지하는 RxJS 실무 필수 패턴
    
    
• switchMap(...)
  
  • keyword가 새로 들어올 때마다 fakeSearch()를 호출
  • 이전 요청은 자동으로 취소됨 (중첩 요청 방지)
    
    • mergeMap: 모든 요청 유지 (동시 요청)
    • exhaustMap: 첫 요청만 허용 (중복 무시)
    • switchMap: 가장 최신 요청만 유지 (검색에 가장 적합)
• .subscribe(result => this.result = result)
  
  • 검색 결과를 받아서 result에 할당 → 화면에 즉시 렌더링

 

 

---

BehaviorSubject vs Subject  요약

항목	BehaviorSubject	Subject
초기값	필요	필요 없음
최근 값 보관	있음	없음
새 Subscriber에 값 전송	최신 값 즉시 전송	발행 시점 이후만 수신
주 용도	상태 저장 및 전파	순수 이벤트 발행 용도

 

RxJS 연산자 설명 요약 

연산자	설명
debounceTime(300)	타이핑 300ms 이후에만 발행
distinctUntilChanged()	같은 값 중복 발행 방지
filter()	조건에 맞는 값만 통과
takeUntil()	컴포넌트 파괴 시 자동 구독 종료
switchMap()	이전 요청을 취소하고 최신 요청만 실행