import { of, from, fromEvent } from 'rxjs';

// of - emit static values
of(1, 2, 3).subscribe(val => console.log(val)); // 1, 2, 3

// from - convert array to observable
from([10, 20, 30]).subscribe(val => console.log(val)); // 10, 20, 30

// from - convert promise to observable
from(fetch('/api/data')).subscribe(response => console.log(response));

// fromEvent - DOM event to observable
const clicks$ = fromEvent(document, 'click');
clicks$.subscribe(event => console.log('Clicked at:', event.clientX, event.clientY));

import { interval, timer, range } from 'rxjs';
import { take } from 'rxjs/operators';

// interval - emit every 1000ms
interval(1000).pipe(take(3))
  .subscribe(n => console.log('Interval:', n)); // 0, 1, 2

// timer - emit after 2s, then every 1s
timer(2000, 1000).pipe(take(3))
  .subscribe(n => console.log('Timer:', n)); // 0 (after 2s), 1, 2

// timer - single emission after delay
timer(3000).subscribe(() => console.log('Delayed action'));

// range - synchronous sequence
range(1, 5).subscribe(n => console.log('Range:', n)); // 1, 2, 3, 4, 5

import { from, defer } from 'rxjs';
import { firstValueFrom } from 'rxjs';

// Convert Promise to Observable
const promise = fetch('/api/user');
from(promise).subscribe(
  response => console.log('Success:', response),
  error => console.error('Error:', error)
);

// Lazy Promise creation with defer
const lazyPromise$ = defer(() => fetch('/api/data'));
lazyPromise$.subscribe(); // Fetch happens only on subscribe

// Convert Observable to Promise for async/await
async function fetchData() {
  const obs$ = from(fetch('/api/items'));
  const response = await firstValueFrom(obs$);
  return response.json();
}

import { ajax } from 'rxjs/ajax';
import { from } from 'rxjs';
import { map, catchError } from 'rxjs/operators';

// ajax GET request
ajax.get('https://api.example.com/users')
  .pipe(
    map(response => response.response),
    catchError(error => of({ error: error.message }))
  )
  .subscribe(data => console.log(data));

// ajax POST request
ajax.post('https://api.example.com/users', 
  { name: 'John', email: 'john@example.com' },
  { 'Content-Type': 'application/json' }
).subscribe(response => console.log(response));

// Custom ajax configuration
ajax({
  url: 'https://api.example.com/data',
  method: 'PUT',
  headers: { 'Authorization': 'Bearer token123' },
  body: { value: 42 },
  timeout: 5000
}).subscribe(response => console.log(response));

// Fetch API as Observable
from(fetch('https://api.example.com/items'))
  .pipe(
    switchMap(response => from(response.json()))
  )
  .subscribe(data => console.log(data));

import { Observable } from 'rxjs';

// Custom Observable with manual emissions
const custom$ = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
  
  // Teardown function for cleanup
  return () => console.log('Unsubscribed');
});

custom$.subscribe({
  next: val => console.log('Value:', val),
  complete: () => console.log('Complete')
});

// Custom event-based Observable
const createEventObservable = (element, eventName) => {
  return new Observable(subscriber => {
    const handler = (event) => subscriber.next(event);
    element.addEventListener(eventName, handler);
    
    // Cleanup on unsubscribe
    return () => element.removeEventListener(eventName, handler);
  });
};

// WebSocket Observable
const websocket$ = new Observable(subscriber => {
  const ws = new WebSocket('wss://example.com/socket');
  
  ws.onmessage = (event) => subscriber.next(event.data);
  ws.onerror = (error) => subscriber.error(error);
  ws.onclose = () => subscriber.complete();
  
  return () => ws.close();
});

import { EMPTY, NEVER, throwError, of } from 'rxjs';
import { catchError, timeout } from 'rxjs/operators';

// EMPTY - completes immediately
EMPTY.subscribe({
  next: val => console.log('Value:', val),      // Never called
  complete: () => console.log('Complete')        // Immediately called
});

// EMPTY in conditional logic
const shouldProcess = false;
(shouldProcess ? of(1, 2, 3) : EMPTY)
  .subscribe(val => console.log(val)); // No output

// NEVER - infinite stream
NEVER.subscribe({
  next: val => console.log(val),        // Never called
  complete: () => console.log('Done')   // Never called
});

// throwError - immediate error
throwError(() => new Error('Custom error'))
  .subscribe({
    next: val => console.log(val),
    error: err => console.error('Error:', err.message)
  });

// Practical use in error handling
of(1, 2, 3).pipe(
  map(n => {
    if (n === 2) throw new Error('Invalid value');
    return n * 10;
  }),
  catchError(err => throwError(() => new Error('Transformed: ' + err.message)))
).subscribe({
  next: val => console.log(val),
  error: err => console.error(err)
});

import { of, throwError, interval } from 'rxjs';
import { take } from 'rxjs/operators';

// Full observer object pattern
of(1, 2, 3).subscribe({
  next: value => console.log('Next:', value),
  error: err => console.error('Error:', err),
  complete: () => console.log('Complete!')
});

// Shorthand next only
of(10, 20, 30).subscribe(val => console.log(val));

// Next and error callbacks
throwError(() => new Error('Fail')).subscribe(
  val => console.log(val),
  err => console.error('Caught:', err.message)
);

// All three callbacks
interval(1000).pipe(take(3)).subscribe(
  val => console.log('Value:', val),
  err => console.error('Error:', err),
  () => console.log('Stream completed')
);

import { interval } from 'rxjs';

// Create and store subscription
const subscription = interval(1000).subscribe(
  val => console.log('Tick:', val)
);

// Check subscription state
console.log('Closed?', subscription.closed); // false

// Unsubscribe after 5 seconds
setTimeout(() => {
  subscription.unsubscribe();
  console.log('Closed?', subscription.closed); // true
}, 5000);

// Component lifecycle example
class MyComponent {
  private subscription: Subscription;
  
  onInit() {
    this.subscription = interval(1000).subscribe(
      val => this.updateUI(val)
    );
  }
  
  onDestroy() {
    // Always unsubscribe to prevent memory leaks
    this.subscription.unsubscribe();
  }
}

import { interval, fromEvent } from 'rxjs';

// Parent subscription
const parentSub = interval(1000).subscribe(
  val => console.log('Parent:', val)
);

// Add child subscriptions
const childSub1 = interval(500).subscribe(val => console.log('Child1:', val));
const childSub2 = fromEvent(document, 'click').subscribe(() => console.log('Click'));

parentSub.add(childSub1);
parentSub.add(childSub2);

// Add custom teardown logic
parentSub.add(() => console.log('Custom cleanup executed'));

// Unsubscribing parent unsubscribes all children
setTimeout(() => {
  parentSub.unsubscribe(); // Cleans up parent + all children
}, 5000);

// Practical component example
class DataService {
  private subscriptions = new Subscription();
  
  startStreams() {
    this.subscriptions.add(
      interval(1000).subscribe(val => this.process(val))
    );
    this.subscriptions.add(
      fromEvent(window, 'resize').subscribe(() => this.handleResize())
    );
  }
  
  stopStreams() {
    this.subscriptions.unsubscribe(); // Stops all streams
    this.subscriptions = new Subscription(); // Reset for reuse
  }
}

import { interval, Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

// takeUntil pattern (most common)
class Component {
  private destroy$ = new Subject<void>();
  
  ngOnInit() {
    interval(1000).pipe(
      takeUntil(this.destroy$)
    ).subscribe(val => console.log(val));
    
    fromEvent(window, 'resize').pipe(
      takeUntil(this.destroy$)
    ).subscribe(() => this.handleResize());
  }
  
  ngOnDestroy() {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

// Angular async pipe (template)
// component.ts
export class MyComponent {
  data$ = this.http.get('/api/data');
}
// template.html
// <div>{{ data$ | async }}</div>

// Angular v16+ takeUntilDestroyed
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

export class ModernComponent {
  constructor() {
    interval(1000).pipe(
      takeUntilDestroyed() // Automatically tied to component lifecycle
    ).subscribe(val => console.log(val));
  }
}

// Custom decorator pattern
function AutoUnsubscribe() {
  return function(constructor: Function) {
    const original = constructor.prototype.ngOnDestroy;
    constructor.prototype.ngOnDestroy = function() {
      for (let prop in this) {
        const property = this[prop];
        if (property && typeof property.unsubscribe === 'function') {
          property.unsubscribe();
        }
      }
      original?.apply(this);
    };
  };
}

import { interval, fromEvent, Subject } from 'rxjs';
import { takeUntil, take } from 'rxjs/operators';

// BAD - Memory leak
class LeakyComponent {
  ngOnInit() {
    interval(1000).subscribe(val => console.log(val)); // Never stops!
    fromEvent(window, 'scroll').subscribe(() => {}); // Listener never removed!
  }
}

// GOOD - Proper cleanup
class SafeComponent {
  private destroy$ = new Subject<void>();
  private dataSubject$ = new Subject<any>();
  
  ngOnInit() {
    // Auto-cleanup with takeUntil
    interval(1000).pipe(
      takeUntil(this.destroy$)
    ).subscribe(val => console.log(val));
    
    // Event listener cleanup
    fromEvent(window, 'scroll').pipe(
      takeUntil(this.destroy$)
    ).subscribe(() => this.handleScroll());
    
    // HTTP request - auto-completes but explicit is better
    this.http.get('/api/data').pipe(
      take(1) // Ensure single emission
    ).subscribe(data => this.processData(data));
  }
  
  ngOnDestroy() {
    // Cleanup pattern
    this.destroy$.next();
    this.destroy$.complete();
    
    // Complete subjects to release observers
    this.dataSubject$.complete();
  }
}

// Subscription tracking for debugging
class DebugComponent {
  private subs: Subscription[] = [];
  
  ngOnInit() {
    this.subs.push(interval(1000).subscribe());
    this.subs.push(fromEvent(window, 'click').subscribe());
    console.log('Active subscriptions:', this.subs.length);
  }
  
  ngOnDestroy() {
    this.subs.forEach(sub => sub.unsubscribe());
    console.log('Cleaned up', this.subs.length, 'subscriptions');
  }
}

import { Subscription, interval, fromEvent } from 'rxjs';

// Subscription bag pattern (recommended)
class SubscriptionBagComponent {
  private subscriptions = new Subscription();
  
  ngOnInit() {
    this.subscriptions.add(
      interval(1000).subscribe(val => this.tick(val))
    );
    this.subscriptions.add(
      fromEvent(window, 'resize').subscribe(() => this.resize())
    );
  }
  
  ngOnDestroy() {
    this.subscriptions.unsubscribe();
  }
}

// Array container pattern
class ArrayContainerComponent {
  private subscriptions: Subscription[] = [];
  
  ngOnInit() {
    this.subscriptions.push(interval(1000).subscribe());
    this.subscriptions.push(fromEvent(document, 'click').subscribe());
  }
  
  ngOnDestroy() {
    this.subscriptions.forEach(sub => sub.unsubscribe());
    this.subscriptions = [];
  }
}

// Map container for selective cleanup
class MapContainerComponent {
  private subscriptions = new Map<string, Subscription>();
  
  ngOnInit() {
    this.subscriptions.set('timer', 
      interval(1000).subscribe(val => console.log(val))
    );
    this.subscriptions.set('clicks', 
      fromEvent(document, 'click').subscribe(() => console.log('click'))
    );
  }
  
  stopTimer() {
    this.subscriptions.get('timer')?.unsubscribe();
    this.subscriptions.delete('timer');
  }
  
  ngOnDestroy() {
    this.subscriptions.forEach(sub => sub.unsubscribe());
    this.subscriptions.clear();
  }
}

// Reusable subscription manager
class SubscriptionManager {
  private subs = new Subscription();
  
  add(subscription: Subscription): void {
    this.subs.add(subscription);
  }
  
  unsubscribeAll(): void {
    this.subs.unsubscribe();
    this.subs = new Subscription();
  }
  
  get count(): number {
    // Track active subscriptions (custom implementation)
    return Object.keys(this.subs).length;
  }
}

import { of, fromEvent } from 'rxjs';
import { map, mapTo } from 'rxjs/operators';

// map - transform values
of(1, 2, 3).pipe(
  map(x => x * 10)
).subscribe(val => console.log(val)); // 10, 20, 30

// map - extract object properties
of(
  { name: 'Alice', age: 25 },
  { name: 'Bob', age: 30 }
).pipe(
  map(user => user.name)
).subscribe(name => console.log(name)); // 'Alice', 'Bob'

// map - complex transformation
of(1, 2, 3, 4, 5).pipe(
  map(n => ({ value: n, squared: n * n, isEven: n % 2 === 0 }))
).subscribe(obj => console.log(obj));

// mapTo - convert all values to constant
fromEvent(document, 'click').pipe(
  mapTo('Clicked!')
).subscribe(msg => console.log(msg)); // Always 'Clicked!'

// mapTo - signal transformation
of('a', 'b', 'c').pipe(
  mapTo(1)
).subscribe(val => console.log(val)); // 1, 1, 1

import { of } from 'rxjs';
import { map } from 'rxjs/operators';

const users = of(
  { name: 'Alice', address: { city: 'NYC', zip: '10001' } },
  { name: 'Bob', address: { city: 'LA', zip: '90001' } }
);

// OLD - pluck (deprecated)
// users.pipe(pluck('address', 'city'))

// NEW - use map instead (recommended)
users.pipe(
  map(user => user.address.city)
).subscribe(city => console.log(city)); // 'NYC', 'LA'

// Multiple property extraction with map
users.pipe(
  map(user => ({
    name: user.name,
    city: user.address.city
  }))
).subscribe(data => console.log(data));

// Safe property access with optional chaining
of(
  { name: 'Alice', address: { city: 'NYC' } },
  { name: 'Bob' } // Missing address
).pipe(
  map(user => user.address?.city ?? 'Unknown')
).subscribe(city => console.log(city)); // 'NYC', 'Unknown'

import { of } from 'rxjs';
import { scan, reduce } from 'rxjs/operators';

// scan - emits intermediate accumulations
of(1, 2, 3, 4, 5).pipe(
  scan((acc, val) => acc + val, 0)
).subscribe(total => console.log('Scan:', total));
// Output: 1, 3, 6, 10, 15 (running total)

// reduce - emits only final result
of(1, 2, 3, 4, 5).pipe(
  reduce((acc, val) => acc + val, 0)
).subscribe(total => console.log('Reduce:', total));
// Output: 15 (only final total)

// scan - track state over time
of(1, -2, 3, -4, 5).pipe(
  scan((acc, val) => ({
    sum: acc.sum + val,
    count: acc.count + 1,
    min: Math.min(acc.min, val),
    max: Math.max(acc.max, val)
  }), { sum: 0, count: 0, min: Infinity, max: -Infinity })
).subscribe(stats => console.log(stats));

// Practical: running balance
const transactions = of(100, -50, 200, -75, 50);
transactions.pipe(
  scan((balance, amount) => balance + amount, 0)
).subscribe(balance => console.log('Balance:', balance));
// 100, 50, 250, 175, 225

// Practical: collect items into array
of('a', 'b', 'c').pipe(
  scan((arr, item) => [...arr, item], [])
).subscribe(arr => console.log('Array:', arr));
// ['a'], ['a','b'], ['a','b','c']

import { interval, fromEvent } from 'rxjs';
import { buffer, bufferTime, bufferCount, take } from 'rxjs/operators';

// buffer - collect until notifier emits
const clicks$ = fromEvent(document, 'click');
interval(1000).pipe(
  buffer(clicks$),
  take(3)
).subscribe(buffered => console.log('Buffered:', buffered));
// Emits array of interval values when user clicks

// bufferTime - time-based buffering
interval(100).pipe(
  bufferTime(1000),
  take(3)
).subscribe(arr => console.log('Buffer:', arr));
// Emits ~10 values every 1 second: [0,1,2...9], [10,11...19], etc.

// bufferCount - count-based buffering
interval(100).pipe(
  bufferCount(5),
  take(3)
).subscribe(arr => console.log('Count buffer:', arr));
// [0,1,2,3,4], [5,6,7,8,9], [10,11,12,13,14]

// bufferCount with overlap
interval(100).pipe(
  bufferCount(3, 1), // size 3, start every 1
  take(5)
).subscribe(arr => console.log('Sliding:', arr));
// [0,1,2], [1,2,3], [2,3,4], [3,4,5], [4,5,6]

// Practical: batch API calls
const userActions$ = fromEvent(button, 'click').pipe(
  map(() => ({ action: 'click', timestamp: Date.now() }))
);
userActions$.pipe(
  bufferTime(5000), // Collect 5 seconds of actions
  filter(actions => actions.length > 0)
).subscribe(batch => sendToAnalytics(batch));

import { of, interval } from 'rxjs';
import { concatMap, mergeMap, switchMap, take, delay } from 'rxjs/operators';

// concatMap - sequential execution (waits for completion)
of(1, 2, 3).pipe(
  concatMap(n => 
    of(n * 10).pipe(delay(1000)) // Each waits for previous
  )
).subscribe(val => console.log('Concat:', val));
// Output at: 1s: 10, 2s: 20, 3s: 30

// mergeMap - concurrent execution (all run in parallel)
of(1, 2, 3).pipe(
  mergeMap(n => 
    of(n * 10).pipe(delay(1000)) // All start immediately
  )
).subscribe(val => console.log('Merge:', val));
// Output at: 1s: 10, 20, 30 (all at once)

// switchMap - cancel previous (only latest completes)
interval(1000).pipe(
  take(3),
  switchMap(n => 
    interval(700).pipe(
      take(3),
      map(i => `${n}-${i}`)
    )
  )
).subscribe(val => console.log('Switch:', val));
// Cancels previous inner observables when new value arrives

// Practical: Search typeahead with switchMap
const searchInput$ = fromEvent(inputElement, 'input');
searchInput$.pipe(
  debounceTime(300),
  map(e => e.target.value),
  switchMap(query => 
    this.http.get(`/api/search?q=${query}`) // Cancels previous request
  )
).subscribe(results => displayResults(results));

// Practical: Sequential file uploads with concatMap
const files$ = from([file1, file2, file3]);
files$.pipe(
  concatMap(file => uploadFile(file)) // Upload one at a time
).subscribe(response => console.log('Uploaded:', response));

// Practical: Parallel API calls with mergeMap
const userIds$ = of(1, 2, 3, 4, 5);
userIds$.pipe(
  mergeMap(id => 
    this.http.get(`/api/users/${id}`), // All requests in parallel
    3 // Limit to 3 concurrent requests
  )
).subscribe(user => console.log(user));

import { fromEvent, of } from 'rxjs';
import { exhaustMap, delay, tap } from 'rxjs/operators';

// exhaustMap - ignore clicks while request is pending
const loginButton = document.getElementById('login-btn');
fromEvent(loginButton, 'click').pipe(
  exhaustMap(() => 
    this.http.post('/api/login', credentials).pipe(
      delay(2000) // Simulates API call
    )
  )
).subscribe(response => console.log('Login response:', response));
// Ignores additional clicks until first request completes

// Compare with other operators
const clicks$ = fromEvent(button, 'click');

// switchMap - cancels previous request
clicks$.pipe(
  switchMap(() => apiCall()) // Latest click cancels previous
);

// mergeMap - allows concurrent requests
clicks$.pipe(
  mergeMap(() => apiCall()) // Every click creates new request
);

// exhaustMap - ignores new clicks while busy
clicks$.pipe(
  exhaustMap(() => apiCall()) // Ignores clicks during request
);

// Practical: Form submission protection
const submitButton$ = fromEvent(submitBtn, 'click');
submitButton$.pipe(
  exhaustMap(() => 
    this.http.post('/api/submit', formData).pipe(
      tap(() => console.log('Submitting...')),
      catchError(err => {
        console.error('Submission failed');
        return EMPTY;
      })
    )
  )
).subscribe(response => {
  console.log('Success:', response);
  resetForm();
});

// Practical: Refresh button with cooldown
fromEvent(refreshBtn, 'click').pipe(
  exhaustMap(() => 
    concat(
      this.http.get('/api/data'),
      timer(5000) // 5s cooldown before accepting new clicks
    )
  )
).subscribe(data => updateUI(data));

// Practical: Game actions (prevent spam)
const attackButton$ = fromEvent(gameButton, 'click');
attackButton$.pipe(
  exhaustMap(() => 
    of('Attack!').pipe(
      tap(() => playAnimation()),
      delay(1000) // Animation duration
    )
  )
).subscribe(action => executeAction(action));

import { of, fromEvent } from 'rxjs';
import { filter, map } from 'rxjs/operators';

// Basic filter - even numbers only
of(1, 2, 3, 4, 5, 6).pipe(
  filter(n => n % 2 === 0)
).subscribe(val => console.log(val)); // 2, 4, 6

// Filter objects by property
of(
  { name: 'Alice', age: 25 },
  { name: 'Bob', age: 17 },
  { name: 'Charlie', age: 30 }
).pipe(
  filter(user => user.age >= 18)
).subscribe(user => console.log(user.name)); // Alice, Charlie

// Filter with index
of(10, 20, 30, 40, 50).pipe(
  filter((val, idx) => idx % 2 === 0) // Every other item
).subscribe(val => console.log(val)); // 10, 30, 50

// Filter null/undefined values
of(1, null, 2, undefined, 3, null).pipe(
  filter(val => val != null) // != catches both null and undefined
).subscribe(val => console.log(val)); // 1, 2, 3

// Practical: filter valid keyboard inputs
fromEvent(input, 'keyup').pipe(
  map(e => e.target.value),
  filter(text => text.length >= 3), // Min 3 characters
  filter(text => /^[a-zA-Z0-9]+$/.test(text)) // Alphanumeric only
).subscribe(validInput => search(validInput));

// Practical: filter successful HTTP responses
this.http.get('/api/data').pipe(
  filter(response => response.status === 200),
  map(response => response.body)
).subscribe(data => processData(data));

import { interval, fromEvent, Subject } from 'rxjs';
import { take, takeUntil, takeWhile, takeLast } from 'rxjs/operators';

// take - first N values
interval(1000).pipe(
  take(5)
).subscribe(val => console.log(val)); // 0, 1, 2, 3, 4, then completes

// takeUntil - common cleanup pattern
const destroy$ = new Subject<void>();
interval(500).pipe(
  takeUntil(destroy$)
).subscribe(val => console.log(val));
// Later: destroy$.next() stops the interval

// takeWhile - conditional stopping
interval(500).pipe(
  takeWhile(n => n < 5)
).subscribe(val => console.log(val)); // 0, 1, 2, 3, 4

// takeWhile with inclusive flag
interval(500).pipe(
  takeWhile(n => n < 5, true) // Includes the failing value
).subscribe(val => console.log(val)); // 0, 1, 2, 3, 4, 5

// takeLast - buffer last N
of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).pipe(
  takeLast(3)
).subscribe(val => console.log(val)); // 8, 9, 10

// Practical: stop on button click
const stop$ = fromEvent(stopButton, 'click');
interval(1000).pipe(
  takeUntil(stop$)
).subscribe(tick => updateTimer(tick));

// Practical: take until condition met
const temps$ = interval(1000).pipe(
  map(() => readTemperature())
);
temps$.pipe(
  takeWhile(temp => temp < 100, true) // Include threshold value
).subscribe(temp => console.log('Temp:', temp));

// Practical: Component lifecycle
class Component {
  private destroy$ = new Subject<void>();
  
  ngOnInit() {
    this.dataService.updates$.pipe(
      takeUntil(this.destroy$)
    ).subscribe(data => this.render(data));
  }
  
  ngOnDestroy() {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

import { of, EMPTY } from 'rxjs';
import { first, last, single } from 'rxjs/operators';

// first - get first emission
of(1, 2, 3, 4, 5).pipe(
  first()
).subscribe(val => console.log(val)); // 1

// first with predicate
of(1, 2, 3, 4, 5).pipe(
  first(n => n > 3)
).subscribe(val => console.log(val)); // 4

// first with default (prevents error on empty)
EMPTY.pipe(
  first(undefined, 'default')
).subscribe(val => console.log(val)); // 'default'

// last - get last emission
of(1, 2, 3, 4, 5).pipe(
  last()
).subscribe(val => console.log(val)); // 5

// last with predicate
of(1, 2, 3, 4, 5).pipe(
  last(n => n < 4)
).subscribe(val => console.log(val)); // 3

// single - must have exactly one emission
of(42).pipe(
  single()
).subscribe(val => console.log(val)); // 42

// single - errors on multiple values
of(1, 2, 3).pipe(
  single()
).subscribe(
  val => console.log(val),
  err => console.error('Error: Multiple values') // Triggered
);

// single with predicate
of(1, 2, 3, 4, 5).pipe(
  single(n => n === 3)
).subscribe(val => console.log(val)); // 3

// Practical: find first matching user
this.users$.pipe(
  first(user => user.id === targetId)
).subscribe(user => displayUser(user));

// Practical: get final calculation result
calculations$.pipe(
  last()
).subscribe(finalResult => saveResult(finalResult));

// Practical: ensure unique result
this.http.get('/api/config').pipe(
  single() // Validates exactly one response
).subscribe(config => applyConfig(config));

import { interval, fromEvent, timer } from 'rxjs';
import { skip, skipUntil, skipWhile, skipLast, take } from 'rxjs/operators';

// skip - ignore first N
of(1, 2, 3, 4, 5).pipe(
  skip(2)
).subscribe(val => console.log(val)); // 3, 4, 5

// skipUntil - start emitting after trigger
const trigger$ = timer(3000); // Emit after 3 seconds
interval(500).pipe(
  skipUntil(trigger$),
  take(5)
).subscribe(val => console.log(val)); // Starts after 3s

// skipWhile - skip until condition false
of(1, 2, 3, 4, 5, 1, 2).pipe(
  skipWhile(n => n < 4)
).subscribe(val => console.log(val)); // 4, 5, 1, 2

// skipLast - omit last N
of(1, 2, 3, 4, 5, 6, 7, 8).pipe(
  skipLast(3)
).subscribe(val => console.log(val)); // 1, 2, 3, 4, 5

// Practical: skip initial loading state
this.dataService.state$.pipe(
  skip(1) // Skip initial/default state
).subscribe(state => updateUI(state));

// Practical: wait for user authentication
const authComplete$ = this.authService.authenticated$;
this.appData$.pipe(
  skipUntil(authComplete$)
).subscribe(data => loadUserData(data));

// Practical: skip warmup values
sensorReadings$.pipe(
  skipWhile(reading => reading.temperature < 20) // Skip until warmed up
).subscribe(reading => processReading(reading));

// Practical: pagination - skip previous pages
const pageSize = 10;
const page = 3;
allItems$.pipe(
  skip(pageSize * (page - 1)),
  take(pageSize)
).subscribe(items => displayPage(items));

import { of } from 'rxjs';
import { distinct, distinctUntilChanged, distinctUntilKeyChanged } from 'rxjs/operators';

// distinct - all unique values in entire stream
of(1, 2, 2, 3, 1, 4, 3, 5).pipe(
  distinct()
).subscribe(val => console.log(val)); // 1, 2, 3, 4, 5

// distinct with key selector
of(
  { id: 1, name: 'Alice' },
  { id: 2, name: 'Bob' },
  { id: 1, name: 'Alice Again' } // Duplicate id
).pipe(
  distinct(user => user.id)
).subscribe(user => console.log(user.name)); // Alice, Bob

// distinctUntilChanged - consecutive duplicates only
of(1, 1, 2, 2, 2, 3, 1, 1, 4).pipe(
  distinctUntilChanged()
).subscribe(val => console.log(val)); // 1, 2, 3, 1, 4

// distinctUntilChanged with custom comparator
of(
  { value: 1 },
  { value: 1 }, // Same value
  { value: 2 }
).pipe(
  distinctUntilChanged((prev, curr) => prev.value === curr.value)
).subscribe(obj => console.log(obj)); // First and third only

// distinctUntilKeyChanged - compare object property
of(
  { status: 'loading' },
  { status: 'loading' }, // Duplicate
  { status: 'success' },
  { status: 'success' }  // Duplicate
).pipe(
  distinctUntilKeyChanged('status')
).subscribe(state => console.log(state.status)); // loading, success

// Practical: search input deduplication
fromEvent(searchInput, 'input').pipe(
  map(e => e.target.value),
  distinctUntilChanged(), // Only when value actually changes
  debounceTime(300)
).subscribe(query => search(query));

// Practical: state change detection
this.store.select('user').pipe(
  distinctUntilKeyChanged('userId'), // Only when userId changes
  switchMap(user => this.loadUserData(user.userId))
).subscribe(data => updateUI(data));

// Practical: unique IDs from stream
eventStream$.pipe(
  map(event => event.id),
  distinct() // Collect all unique IDs
).subscribe(uniqueId => processUniqueId(uniqueId));

// Performance note: distinct with flush
const flush$ = interval(10000); // Clear cache every 10s
dataStream$.pipe(
  distinct(item => item.id, flush$) // Periodically clear stored values
).subscribe(item => processItem(item));

import { fromEvent, interval } from 'rxjs';
import { debounceTime, throttleTime, auditTime, sampleTime, map } from 'rxjs/operators';

// debounceTime - wait for pause in emissions
fromEvent(searchInput, 'input').pipe(
  map(e => e.target.value),
  debounceTime(500) // Wait 500ms after user stops typing
).subscribe(query => searchAPI(query));

// throttleTime - first value, then silence
fromEvent(button, 'click').pipe(
  throttleTime(2000) // First click, ignore for 2s, repeat
).subscribe(() => console.log('Throttled click'));

// throttleTime with trailing edge
fromEvent(window, 'scroll').pipe(
  throttleTime(1000, { leading: true, trailing: true })
).subscribe(() => checkScrollPosition());

// auditTime - most recent value after period
fromEvent(document, 'mousemove').pipe(
  auditTime(1000) // Emit latest mouse position every 1s
).subscribe(event => updateTooltip(event));

// sampleTime - periodic sampling
const temperature$ = interval(100).pipe(
  map(() => Math.random() * 100)
);
temperature$.pipe(
  sampleTime(1000) // Sample every 1 second
).subscribe(temp => displayTemperature(temp));

// Practical comparison - resize handling
const resize$ = fromEvent(window, 'resize');

// debounceTime - wait for resize to finish
resize$.pipe(
  debounceTime(300)
).subscribe(() => recalculateLayout()); // After resize complete

// throttleTime - periodic updates during resize
resize$.pipe(
  throttleTime(100)
).subscribe(() => updatePreview()); // Every 100ms while resizing

// Practical: autocomplete with debounce
fromEvent(input, 'input').pipe(
  map(e => e.target.value),
  filter(text => text.length >= 3),
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(query => this.http.get(`/api/search?q=${query}`))
).subscribe(results => displaySuggestions(results));

// Practical: save button with throttle
fromEvent(saveBtn, 'click').pipe(
  throttleTime(3000) // Prevent rapid saves
).subscribe(() => saveDocument());

// Practical: scroll position tracking
fromEvent(window, 'scroll').pipe(
  auditTime(100), // Latest position every 100ms
  map(() => window.scrollY)
).subscribe(scrollY => updateScrollIndicator(scrollY));

import { merge, interval, fromEvent, of } from 'rxjs';
import { map, mergeAll, take } from 'rxjs/operators';

// merge - combine multiple sources
const fast$ = interval(500).pipe(map(x => `Fast: ${x}`), take(5));
const slow$ = interval(1000).pipe(map(x => `Slow: ${x}`), take(3));
merge(fast$, slow$).subscribe(val => console.log(val));
// Output: Fast:0, Slow:0, Fast:1, Fast:2, Slow:1, Fast:3, Fast:4, Slow:2

// Pipeable merge
of(1, 2, 3).pipe(
  merge(of(4, 5, 6), of(7, 8, 9))
).subscribe(val => console.log(val)); // 1,2,3,4,5,6,7,8,9

// mergeAll - flatten higher-order observable
const clicks$ = fromEvent(document, 'click');
clicks$.pipe(
  map(() => interval(1000).pipe(take(3))),
  mergeAll() // Subscribe to all inner intervals concurrently
).subscribe(val => console.log(val));

// mergeAll with concurrency limit
const urls = ['url1', 'url2', 'url3', 'url4', 'url5'];
from(urls).pipe(
  map(url => ajax.get(url)),
  mergeAll(2) // Max 2 concurrent HTTP requests
).subscribe(response => console.log(response));

// Practical: combine multiple data sources
const user$ = this.http.get('/api/user');
const settings$ = this.http.get('/api/settings');
const notifications$ = this.http.get('/api/notifications');

merge(user$, settings$, notifications$).subscribe(
  data => console.log('Data arrived:', data)
);

// Practical: multiple event sources
const keyPress$ = fromEvent(document, 'keypress').pipe(mapTo('key'));
const mouseClick$ = fromEvent(document, 'click').pipe(mapTo('click'));
const touchTap$ = fromEvent(document, 'touchstart').pipe(mapTo('touch'));

merge(keyPress$, mouseClick$, touchTap$).pipe(
  scan((count) => count + 1, 0)
).subscribe(count => console.log('Total interactions:', count));

import { concat, of, interval } from 'rxjs';
import { concatAll, take, delay } from 'rxjs/operators';

// concat - sequential execution
const first$ = of(1, 2, 3);
const second$ = of(4, 5, 6);
const third$ = of(7, 8, 9);
concat(first$, second$, third$).subscribe(val => console.log(val));
// Output: 1,2,3,4,5,6,7,8,9 (in order)

// concat waits for completion
const slow$ = interval(1000).pipe(take(3));
const fast$ = of('a', 'b', 'c');
concat(slow$, fast$).subscribe(val => console.log(val));
// Output: 0,1,2 (over 3 seconds), then immediately a,b,c

// Pipeable concat
of(1, 2, 3).pipe(
  concat(of(4, 5, 6))
).subscribe(val => console.log(val));

// concatAll - sequential flattening
of(
  of(1, 2, 3),
  of(4, 5, 6),
  of(7, 8, 9)
).pipe(
  concatAll()
).subscribe(val => console.log(val)); // 1,2,3,4,5,6,7,8,9

// Practical: sequential API calls
const deleteUser = (id) => this.http.delete(`/api/users/${id}`);
const userIds = [1, 2, 3];
from(userIds).pipe(
  map(id => deleteUser(id)),
  concatAll() // Delete one at a time, in order
).subscribe(
  result => console.log('Deleted:', result),
  err => console.error('Error:', err)
);

// Practical: animation sequence
const animations = [
  animate('fadeIn', 500),
  animate('slideUp', 300),
  animate('bounce', 400)
];
concat(...animations).subscribe(
  animation => console.log('Playing:', animation),
  null,
  () => console.log('Animation sequence complete')
);

// Practical: onboarding flow
const step1$ = showWelcome().pipe(delay(2000));
const step2$ = showFeatures().pipe(delay(3000));
const step3$ = showTutorial().pipe(delay(2000));
concat(step1$, step2$, step3$).subscribe(
  step => displayStep(step),
  null,
  () => completeOnboarding()
);

import { combineLatest, of, interval } from 'rxjs';
import { combineLatestWith, take, map } from 'rxjs/operators';

// combineLatest - array form
const age$ = of(25, 26, 27);
const name$ = of('Alice', 'Bob');
combineLatest([age$, name$]).subscribe(
  ([age, name]) => console.log(`${name} is ${age}`)
);
// Bob is 25, Bob is 26, Bob is 27

// combineLatest - object form (recommended)
const firstName$ = of('John');
const lastName$ = of('Doe', 'Smith');
const age2$ = of(30, 31);
combineLatest({
  firstName: firstName$,
  lastName: lastName$,
  age: age2$
}).subscribe(user => console.log(user));
// {firstName: 'John', lastName: 'Doe', age: 30}
// {firstName: 'John', lastName: 'Doe', age: 31}
// {firstName: 'John', lastName: 'Smith', age: 31}

// combineLatestWith - pipeable
interval(1000).pipe(
  take(3),
  combineLatestWith(interval(1500).pipe(take(2)))
).subscribe(([a, b]) => console.log(`a: ${a}, b: ${b}`));

// Practical: form validation
const username$ = fromEvent(usernameInput, 'input').pipe(
  map(e => e.target.value)
);
const password$ = fromEvent(passwordInput, 'input').pipe(
  map(e => e.target.value)
);
const terms$ = fromEvent(termsCheckbox, 'change').pipe(
  map(e => e.target.checked)
);

combineLatest({
  username: username$,
  password: password$,
  terms: terms$
}).pipe(
  map(form => ({
    ...form,
    valid: form.username.length > 3 && 
           form.password.length > 8 && 
           form.terms
  }))
).subscribe(form => {
  submitButton.disabled = !form.valid;
});

// Practical: derived data from multiple sources
const temperature$ = sensorData$.pipe(map(d => d.temp));
const humidity$ = sensorData$.pipe(map(d => d.humidity));
const pressure$ = sensorData$.pipe(map(d => d.pressure));

combineLatest({
  temp: temperature$,
  humidity: humidity$,
  pressure: pressure$
}).pipe(
  map(readings => calculateComfortIndex(readings))
).subscribe(index => displayComfortLevel(index));

// Practical: multi-filter search
const nameFilter$ = new BehaviorSubject('');
const categoryFilter$ = new BehaviorSubject('all');
const priceRange$ = new BehaviorSubject({ min: 0, max: 1000 });

combineLatest([nameFilter$, categoryFilter$, priceRange$]).pipe(
  debounceTime(300),
  switchMap(([name, category, price]) => 
    this.http.get('/api/products', { 
      params: { name, category, minPrice: price.min, maxPrice: price.max }
    })
  )
).subscribe(products => displayProducts(products));

import { fromEvent, interval, BehaviorSubject } from 'rxjs';
import { withLatestFrom, map, take } from 'rxjs/operators';

// withLatestFrom - basic usage
const clicks$ = fromEvent(button, 'click');
const timer$ = interval(1000);
clicks$.pipe(
  withLatestFrom(timer$)
).subscribe(([click, time]) => console.log(`Click at time: ${time}`));
// Only emits when button clicked, includes latest timer value

// Compare with combineLatest
const source$ = interval(1000).pipe(take(3));
const other$ = interval(500).pipe(take(5));

// combineLatest - emits on ANY change
combineLatest([source$, other$]).subscribe(
  ([s, o]) => console.log(`combineLatest: ${s}, ${o}`)
);

// withLatestFrom - emits only on source change
source$.pipe(
  withLatestFrom(other$)
).subscribe(
  ([s, o]) => console.log(`withLatestFrom: ${s}, ${o}`)
);

// withLatestFrom multiple sources
const userAction$ = fromEvent(actionButton, 'click');
const userState$ = new BehaviorSubject({ id: 1, role: 'user' });
const appConfig$ = new BehaviorSubject({ theme: 'dark', locale: 'en' });

userAction$.pipe(
  withLatestFrom(userState$, appConfig$)
).subscribe(([event, user, config]) => {
  console.log('Action by:', user.id);
  console.log('Theme:', config.theme);
  performAction(event, user, config);
});

// Practical: save with current state
const saveButton$ = fromEvent(saveBtn, 'click');
const formState$ = new BehaviorSubject({});
const userId$ = new BehaviorSubject(null);

saveButton$.pipe(
  withLatestFrom(formState$, userId$),
  switchMap(([_, formData, userId]) => 
    this.http.post('/api/save', { userId, data: formData })
  )
).subscribe(result => showSuccessMessage(result));

// Practical: enriched analytics
const clickEvent$ = fromEvent(document, 'click');
const currentPage$ = new BehaviorSubject('/home');
const sessionData$ = new BehaviorSubject({ sessionId: 'abc123' });

clickEvent$.pipe(
  withLatestFrom(currentPage$, sessionData$),
  map(([event, page, session]) => ({
    type: 'click',
    element: event.target.tagName,
    page: page,
    sessionId: session.sessionId,
    timestamp: Date.now()
  }))
).subscribe(analyticsData => sendToAnalytics(analyticsData));

// Practical: game input with state
const keyPress$ = fromEvent(document, 'keydown');
const playerState$ = new BehaviorSubject({ 
  x: 0, y: 0, health: 100, ammo: 50 
});

keyPress$.pipe(
  withLatestFrom(playerState$),
  map(([event, state]) => processInput(event.key, state))
).subscribe(newState => playerState$.next(newState));