Real-world Application Patterns

// Autocomplete search
const searchInput$ = fromEvent<Event>(searchInput, 'input').pipe(
  map(e => (e.target as HTMLInputElement).value),
  debounceTime(300),           // Wait for user to stop typing
  distinctUntilChanged(),      // Only if value changed
  filter(term => term.length >= 2),  // Min 2 characters
  tap(() => showLoadingSpinner()),
  switchMap(term =>            // Cancel previous searches
    this.http.get<Result[]>(`/api/search?q=${term}`).pipe(
      catchError(err => {
        console.error('Search failed:', err);
        return of([]);         // Return empty on error
      }),
      finalize(() => hideLoadingSpinner())
    )
  ),
  shareReplay({ bufferSize: 1, refCount: true })
);

searchInput$.subscribe(results => displayResults(results));

// Save button that prevents duplicate clicks
const saveButton$ = fromEvent(saveBtn, 'click');

saveButton$.pipe(
  tap(() => saveBtn.disabled = true),
  exhaustMap(() =>               // Ignore clicks while saving
    this.http.post('/api/save', formData).pipe(
      timeout(10000),            // 10 second timeout
      retry(2),                  // Retry twice
      catchError(err => {
        showError('Save failed');
        return throwError(() => err);
      }),
      finalize(() => saveBtn.disabled = false)
    )
  )
).subscribe({
  next: response => showSuccess('Saved successfully'),
  error: err => console.error('Save error:', err)
});

// Load user details for multiple IDs with max 5 concurrent
const userIds = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];

from(userIds).pipe(
  mergeMap(
    id => this.http.get<User>(`/api/users/${id}`).pipe(
      catchError(err => {
        console.error(`Failed to load user ${id}:`, err);
        return of(null);         // Continue with others
      })
    ),
    5                            // Max 5 concurrent requests
  ),
  filter(user => user !== null), // Remove failed requests
  toArray()                      // Collect all results
).subscribe(users => {
  console.log('Loaded users:', users);
  displayUsers(users);
});

// Polling service with start/stop
class PollingService {
  private polling$ = new Subject<boolean>();
  
  startPolling(url: string, intervalMs: number = 5000): Observable<any> {
    return this.polling$.pipe(
      startWith(true),
      switchMap(shouldPoll => 
        shouldPoll 
          ? interval(intervalMs).pipe(
              startWith(0),      // Immediate first request
              switchMap(() => this.http.get(url).pipe(
                catchError(err => {
                  console.error('Poll failed:', err);
                  return of(null);
                })
              ))
            )
          : EMPTY                // Stop polling
      )
    );
  }
  
  stop() {
    this.polling$.next(false);
  }
  
  start() {
    this.polling$.next(true);
  }
}

// Usage
const poller = new PollingService();
const data$ = poller.startPolling('/api/status', 3000);

data$.subscribe(status => updateStatus(status));

// Stop polling when tab not visible
document.addEventListener('visibilitychange', () => {
  if (document.hidden) {
    poller.stop();
  } else {
    poller.start();
  }
});

// Email input with async validation
const emailInput$ = fromEvent<Event>(emailField, 'input').pipe(
  map(e => (e.target as HTMLInputElement).value),
  shareReplay({ bufferSize: 1, refCount: true })
);

// Synchronous validation
const emailValid$ = emailInput$.pipe(
  map(email => ({
    valid: /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email),
    error: 'Invalid email format'
  }))
);

// Asynchronous validation (check availability)
const emailAvailable$ = emailInput$.pipe(
  debounceTime(500),
  distinctUntilChanged(),
  filter(email => /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email)),
  switchMap(email => 
    this.http.get<{available: boolean}>(`/api/check-email?email=${email}`).pipe(
      map(result => ({
        valid: result.available,
        error: result.available ? null : 'Email already taken'
      })),
      catchError(() => of({ valid: true, error: null }))
    )
  ),
  startWith({ valid: true, error: null })
);

// Combine validations
combineLatest([emailValid$, emailAvailable$]).pipe(
  map(([format, available]) => {
    if (!format.valid) return format;
    if (!available.valid) return available;
    return { valid: true, error: null };
  })
).subscribe(validation => {
  if (validation.valid) {
    emailField.classList.remove('error');
    errorMsg.textContent = '';
  } else {
    emailField.classList.add('error');
    errorMsg.textContent = validation.error;
  }
});

// Password strength indicator
const passwordInput$ = fromEvent<Event>(passwordField, 'input').pipe(
  map(e => (e.target as HTMLInputElement).value),
  shareReplay({ bufferSize: 1, refCount: true })
);

interface PasswordStrength {
  score: number;      // 0-4
  label: string;      // weak, fair, good, strong
  requirements: {
    length: boolean;
    uppercase: boolean;
    lowercase: boolean;
    number: boolean;
    special: boolean;
  };
}

const passwordStrength$ = passwordInput$.pipe(
  map(password => {
    const requirements = {
      length: password.length >= 8,
      uppercase: /[A-Z]/.test(password),
      lowercase: /[a-z]/.test(password),
      number: /[0-9]/.test(password),
      special: /[^A-Za-z0-9]/.test(password)
    };
    
    const metCount = Object.values(requirements).filter(Boolean).length;
    
    let score = 0;
    let label = 'weak';
    
    if (metCount === 5) {
      score = 4;
      label = 'strong';
    } else if (metCount === 4) {
      score = 3;
      label = 'good';
    } else if (metCount === 3) {
      score = 2;
      label = 'fair';
    } else if (metCount >= 1) {
      score = 1;
      label = 'weak';
    }
    
    return { score, label, requirements };
  })
);

passwordStrength$.subscribe(strength => {
  updateStrengthMeter(strength.score);
  strengthLabel.textContent = strength.label;
  
  // Update requirement checklist
  Object.entries(strength.requirements).forEach(([req, met]) => {
    const element = document.getElementById(`req-${req}`);
    element.classList.toggle('met', met);
  });
});

// Multi-step form manager
class FormWizard {
  private currentStep$ = new BehaviorSubject<number>(1);
  private formData$ = new BehaviorSubject<Partial<FormData>>({});
  
  public step$ = this.currentStep$.asObservable();
  public data$ = this.formData$.asObservable();
  
  // Validation status for each step
  public stepValid$ = (step: number): Observable<boolean> => {
    return this.formData$.pipe(
      map(data => this.validateStep(step, data))
    );
  };
  
  // Can navigate to next step
  public canGoNext$ = combineLatest([
    this.currentStep$,
    this.formData$
  ]).pipe(
    map(([step, data]) => this.validateStep(step, data))
  );
  
  updateData(stepData: Partial<FormData>) {
    this.formData$.next({
      ...this.formData$.value,
      ...stepData
    });
  }
  
  nextStep() {
    const current = this.currentStep$.value;
    if (this.validateStep(current, this.formData$.value)) {
      this.currentStep$.next(current + 1);
    }
  }
  
  previousStep() {
    const current = this.currentStep$.value;
    if (current > 1) {
      this.currentStep$.next(current - 1);
    }
  }
  
  submit(): Observable<any> {
    return this.http.post('/api/submit', this.formData$.value);
  }
  
  private validateStep(step: number, data: Partial<FormData>): boolean {
    // Validation logic per step
    switch(step) {
      case 1:
        return !!data.name && !!data.email;
      case 2:
        return !!data.address && !!data.city;
      case 3:
        return !!data.payment;
      default:
        return false;
    }
  }
}

// Usage
const wizard = new FormWizard();

wizard.canGoNext$.subscribe(canGoNext => {
  nextBtn.disabled = !canGoNext;
});

wizard.step$.subscribe(step => {
  showStep(step);
});

// Todo service with optimistic updates
class TodoService {
  private todos$ = new BehaviorSubject<Todo[]>([]);
  
  getTodos(): Observable<Todo[]> {
    return this.todos$.asObservable();
  }
  
  addTodo(text: string): Observable<Todo> {
    // Create optimistic todo
    const optimisticTodo: Todo = {
      id: `temp-${Date.now()}`,
      text,
      completed: false,
      synced: false
    };
    
    // Update UI immediately
    this.todos$.next([...this.todos$.value, optimisticTodo]);
    
    // Sync to server
    return this.http.post<Todo>('/api/todos', { text }).pipe(
      tap(serverTodo => {
        // Replace optimistic with server response
        const todos = this.todos$.value.map(t => 
          t.id === optimisticTodo.id 
            ? { ...serverTodo, synced: true }
            : t
        );
        this.todos$.next(todos);
      }),
      catchError(err => {
        // Rollback on error
        const todos = this.todos$.value.filter(t => t.id !== optimisticTodo.id);
        this.todos$.next(todos);
        return throwError(() => err);
      })
    );
  }
  
  updateTodo(id: string, updates: Partial<Todo>): Observable<Todo> {
    // Optimistic update
    const todos = this.todos$.value.map(t => 
      t.id === id ? { ...t, ...updates, synced: false } : t
    );
    this.todos$.next(todos);
    
    // Sync to server
    return this.http.put<Todo>(`/api/todos/${id}`, updates).pipe(
      tap(serverTodo => {
        const updated = this.todos$.value.map(t => 
          t.id === id ? { ...serverTodo, synced: true } : t
        );
        this.todos$.next(updated);
      }),
      catchError(err => {
        // Revert on error
        this.loadTodos().subscribe();
        return throwError(() => err);
      })
    );
  }
  
  private loadTodos(): Observable<Todo[]> {
    return this.http.get<Todo[]>('/api/todos').pipe(
      tap(todos => this.todos$.next(todos.map(t => ({ ...t, synced: true }))))
    );
  }
}

// Offline sync manager
class OfflineSyncService {
  private syncQueue$ = new BehaviorSubject<SyncOperation[]>([]);
  private online$ = merge(
    fromEvent(window, 'online').pipe(mapTo(true)),
    fromEvent(window, 'offline').pipe(mapTo(false))
  ).pipe(
    startWith(navigator.onLine),
    shareReplay({ bufferSize: 1, refCount: true })
  );
  
  // Sync when online
  constructor() {
    this.online$.pipe(
      filter(online => online),
      switchMap(() => this.processSyncQueue())
    ).subscribe();
  }
  
  queueOperation(operation: SyncOperation) {
    const queue = [...this.syncQueue$.value, operation];
    this.syncQueue$.next(queue);
    
    // Save to localStorage for persistence
    localStorage.setItem('syncQueue', JSON.stringify(queue));
    
    // Try to sync if online
    if (navigator.onLine) {
      this.processSyncQueue().subscribe();
    }
  }
  
  private processSyncQueue(): Observable<void> {
    const queue = this.syncQueue$.value;
    
    if (queue.length === 0) {
      return of(undefined);
    }
    
    return from(queue).pipe(
      concatMap(operation => 
        this.executeOperation(operation).pipe(
          tap(() => this.removeFromQueue(operation)),
          catchError(err => {
            console.error('Sync failed:', err);
            // Keep in queue, try again later
            return of(undefined);
          })
        )
      ),
      toArray(),
      map(() => undefined)
    );
  }
  
  private executeOperation(op: SyncOperation): Observable<any> {
    switch (op.type) {
      case 'CREATE':
        return this.http.post(op.url, op.data);
      case 'UPDATE':
        return this.http.put(op.url, op.data);
      case 'DELETE':
        return this.http.delete(op.url);
      default:
        return throwError(() => new Error('Unknown operation'));
    }
  }
  
  private removeFromQueue(operation: SyncOperation) {
    const queue = this.syncQueue$.value.filter(op => op.id !== operation.id);
    this.syncQueue$.next(queue);
    localStorage.setItem('syncQueue', JSON.stringify(queue));
  }
  
  getQueueStatus(): Observable<{ pending: number; online: boolean }> {
    return combineLatest([
      this.syncQueue$,
      this.online$
    ]).pipe(
      map(([queue, online]) => ({
        pending: queue.length,
        online
      }))
    );
  }
}

// Usage
const syncService = new OfflineSyncService();

// Queue operation
syncService.queueOperation({
  id: Date.now().toString(),
  type: 'CREATE',
  url: '/api/todos',
  data: { text: 'New todo' }
});

// Monitor sync status
syncService.getQueueStatus().subscribe(status => {
  if (status.pending > 0) {
    showSyncIndicator(`${status.pending} items pending`);
  }
});

// Advanced WebSocket message router
class WebSocketRouter {
  private ws$: WebSocketSubject<any>;
  private messageRoutes = new Map<string, Subject<any>>();
  
  constructor(url: string) {
    this.ws$ = webSocket({
      url,
      openObserver: { next: () => console.log('WS Connected') },
      closeObserver: { next: () => console.log('WS Disconnected') }
    });
    
    // Route incoming messages
    this.ws$.pipe(
      retryWhen(errors => errors.pipe(delay(5000)))
    ).subscribe(message => {
      const route = this.messageRoutes.get(message.type);
      if (route) {
        route.next(message.data);
      } else {
        console.warn('Unhandled message type:', message.type);
      }
    });
  }
  
  // Subscribe to specific message type
  on<T>(messageType: string): Observable<T> {
    if (!this.messageRoutes.has(messageType)) {
      this.messageRoutes.set(messageType, new Subject<T>());
    }
    return this.messageRoutes.get(messageType)!.asObservable();
  }
  
  // Send message
  send(type: string, data: any) {
    this.ws$.next({ type, data, timestamp: Date.now() });
  }
  
  // Request-response pattern
  request<T>(type: string, data: any, timeout = 5000): Observable<T> {
    const correlationId = `req-${Date.now()}-${Math.random()}`;
    
    // Wait for response with matching correlationId
    const response$ = this.ws$.pipe(
      filter((msg: any) => 
        msg.type === `${type}-response` && 
        msg.correlationId === correlationId
      ),
      map((msg: any) => msg.data as T),
      take(1),
      timeout(timeout)
    );
    
    // Send request
    this.ws$.next({ 
      type, 
      data, 
      correlationId,
      timestamp: Date.now() 
    });
    
    return response$;
  }
}

// Usage
const wsRouter = new WebSocketRouter('ws://localhost:8080');

// Subscribe to chat messages
wsRouter.on<ChatMessage>('chat').subscribe(
  msg => displayChatMessage(msg)
);

// Subscribe to notifications
wsRouter.on<Notification>('notification').subscribe(
  notif => showNotification(notif)
);

// Request-response
wsRouter.request<UserProfile>('get-profile', { userId: 123 })
  .subscribe({
    next: profile => console.log('Profile:', profile),
    error: err => console.error('Request timeout or error:', err)
  });

// Smooth scroll to element
function smoothScrollTo(element: HTMLElement, duration = 1000): Observable<number> {
  const start = window.pageYOffset;
  const target = element.offsetTop;
  const distance = target - start;
  const startTime = performance.now();
  
  return interval(0, animationFrameScheduler).pipe(
    map(() => (performance.now() - startTime) / duration),
    takeWhile(progress => progress < 1),
    map(progress => {
      // Easing function (ease-in-out)
      const eased = progress < 0.5
        ? 2 * progress * progress
        : -1 + (4 - 2 * progress) * progress;
      return start + (distance * eased);
    }),
    tap(position => window.scrollTo(0, position)),
    endWith(target)
  );
}

// Usage
const scrollBtn = document.getElementById('scroll-btn');
const targetSection = document.getElementById('target');

fromEvent(scrollBtn, 'click').pipe(
  switchMap(() => smoothScrollTo(targetSection, 800))
).subscribe({
  complete: () => console.log('Scroll complete')
});

// Draggable element
function makeDraggable(element: HTMLElement): Observable<{x: number; y: number}> {
  const mouseDown$ = fromEvent<MouseEvent>(element, 'mousedown');
  const mouseMove$ = fromEvent<MouseEvent>(document, 'mousemove');
  const mouseUp$ = fromEvent<MouseEvent>(document, 'mouseup');
  
  return mouseDown$.pipe(
    switchMap(start => {
      const startX = start.clientX - element.offsetLeft;
      const startY = start.clientY - element.offsetTop;
      
      return mouseMove$.pipe(
        map(move => ({
          x: move.clientX - startX,
          y: move.clientY - startY
        })),
        takeUntil(mouseUp$)
      );
    })
  );
}

// Usage
const draggable = document.getElementById('draggable');

makeDraggable(draggable).subscribe(pos => {
  draggable.style.left = `${pos.x}px`;
  draggable.style.top = `${pos.y}px`;
});

// Modal state machine
type ModalState = 'closed' | 'opening' | 'open' | 'closing';
type ModalAction = 'OPEN' | 'CLOSE' | 'TRANSITION_COMPLETE';

class ModalStateMachine {
  private actions$ = new Subject<ModalAction>();
  
  public state$ = this.actions$.pipe(
    scan((state: ModalState, action: ModalAction) => {
      switch (state) {
        case 'closed':
          return action === 'OPEN' ? 'opening' : state;
        case 'opening':
          return action === 'TRANSITION_COMPLETE' ? 'open' : state;
        case 'open':
          return action === 'CLOSE' ? 'closing' : state;
        case 'closing':
          return action === 'TRANSITION_COMPLETE' ? 'closed' : state;
        default:
          return state;
      }
    }, 'closed' as ModalState),
    startWith('closed' as ModalState),
    shareReplay({ bufferSize: 1, refCount: true })
  );
  
  open() {
    this.actions$.next('OPEN');
    
    // Auto-complete transition after animation
    timer(300).subscribe(() => {
      this.actions$.next('TRANSITION_COMPLETE');
    });
  }
  
  close() {
    this.actions$.next('CLOSE');
    
    timer(300).subscribe(() => {
      this.actions$.next('TRANSITION_COMPLETE');
    });
  }
}

// Usage
const modal = new ModalStateMachine();

modal.state$.subscribe(state => {
  const element = document.getElementById('modal');
  
  element.className = `modal modal-${state}`;
  
  if (state === 'open') {
    element.setAttribute('aria-hidden', 'false');
  } else if (state === 'closed') {
    element.setAttribute('aria-hidden', 'true');
  }
});

openBtn.addEventListener('click', () => modal.open());
closeBtn.addEventListener('click', () => modal.close());

// Complex checkout workflow
class CheckoutWorkflow {
  executeCheckout(order: Order): Observable<CheckoutResult> {
    return of(order).pipe(
      // Step 1: Validate cart
      concatMap(order => this.validateCart(order).pipe(
        tap(() => this.updateProgress('Validating cart...'))
      )),
      
      // Step 2: Check inventory
      concatMap(order => this.checkInventory(order).pipe(
        tap(() => this.updateProgress('Checking inventory...'))
      )),
      
      // Step 3: Calculate totals
      concatMap(order => this.calculateTotals(order).pipe(
        tap(() => this.updateProgress('Calculating totals...'))
      )),
      
      // Step 4: Process payment
      concatMap(order => this.processPayment(order).pipe(
        tap(() => this.updateProgress('Processing payment...'))
      )),
      
      // Step 5: Reserve inventory
      concatMap(order => this.reserveInventory(order).pipe(
        tap(() => this.updateProgress('Reserving items...'))
      )),
      
      // Step 6: Create order
      concatMap(order => this.createOrder(order).pipe(
        tap(() => this.updateProgress('Creating order...'))
      )),
      
      // Step 7: Send confirmation
      concatMap(result => this.sendConfirmation(result).pipe(
        map(() => result),
        tap(() => this.updateProgress('Complete!'))
      )),
      
      // Error handling with rollback
      catchError(err => {
        console.error('Checkout failed:', err);
        return this.rollbackCheckout(order).pipe(
          concatMap(() => throwError(() => err))
        );
      })
    );
  }
  
  private rollbackCheckout(order: Order): Observable<void> {
    // Compensating transactions
    return forkJoin([
      this.releaseInventory(order).pipe(catchError(() => of(null))),
      this.refundPayment(order).pipe(catchError(() => of(null))),
      this.cancelOrder(order).pipe(catchError(() => of(null)))
    ]).pipe(
      map(() => undefined),
      tap(() => console.log('Rollback complete'))
    );
  }
}

// Usage
const checkout = new CheckoutWorkflow();

checkout.executeCheckout(order).subscribe({
  next: result => {
    showSuccess(`Order #${result.orderId} confirmed!`);
    redirectToConfirmation(result.orderId);
  },
  error: err => {
    showError('Checkout failed. Please try again.');
    console.error(err);
  }
});

// Circuit breaker for resilient API calls
class CircuitBreaker {
  private failureCount = 0;
  private state: 'CLOSED' | 'OPEN' | 'HALF_OPEN' = 'CLOSED';
  private failureThreshold = 5;
  private resetTimeout = 60000; // 1 minute
  
  execute<T>(request: Observable<T>): Observable<T> {
    if (this.state === 'OPEN') {
      return throwError(() => new Error('Circuit breaker is OPEN'));
    }
    
    return request.pipe(
      tap(() => {
        // Success - reset failure count
        this.failureCount = 0;
        if (this.state === 'HALF_OPEN') {
          this.state = 'CLOSED';
          console.log('Circuit breaker CLOSED');
        }
      }),
      catchError(err => {
        this.failureCount++;
        
        if (this.failureCount >= this.failureThreshold) {
          this.state = 'OPEN';
          console.log('Circuit breaker OPEN');
          
          // Auto-retry after timeout
          timer(this.resetTimeout).subscribe(() => {
            this.state = 'HALF_OPEN';
            this.failureCount = 0;
            console.log('Circuit breaker HALF_OPEN');
          });
        }
        
        return throwError(() => err);
      })
    );
  }
  
  getState(): string {
    return this.state;
  }
}

// Usage
const breaker = new CircuitBreaker();

function makeApiCall(): Observable<any> {
  return breaker.execute(
    this.http.get('/api/unstable-endpoint').pipe(
      timeout(5000)
    )
  ).pipe(
    catchError(err => {
      if (err.message === 'Circuit breaker is OPEN') {
        // Use cached data or show user-friendly message
        return of({ cached: true, data: getCachedData() });
      }
      return throwError(() => err);
    })
  );
}

// Monitor circuit breaker state
interval(1000).subscribe(() => {
  const state = breaker.getState();
  updateStatusIndicator(state);
});