1. Observables — основа всего

📦 Observables — Сердце RxJS

Привет снова, Яша! 👋 Сегодня разбираем самую важную концепцию RxJS — Observable. Без понимания этого всё остальное будет туманом 🌫️

---

Что такое Observable? 🤔

Observable — это объект, который представляет поток данных во времени.

Аналогия: представь подписку на YouTube-канал 📺:

• Канал = Observable (источник данных)
• Ты = Observer (получатель)
• Нажать “Подписаться” = subscribe()
• Нажать “Отписаться” = unsubscribe()
• Новые видео = значения (next)
• Канал закрылся = complete
• Ошибка загрузки = error

import { Observable } from 'rxjs';

// Создаём Observable вручную
const myObservable$ = new Observable(subscriber => {
  // Это функция-исполнитель (producer)
  subscriber.next('Первое значение');   // отправляем значение
  subscriber.next('Второе значение');
  subscriber.next('Третье значение');
  subscriber.complete();               // сообщаем что всё!
});

// Пока не подписались — НИЧЕГО не происходит! 😴
// Observable ленивый!

// Подписываемся и начинаем получать данные
myObservable$.subscribe({
  next: value => console.log('Получили:', value),
  error: err => console.error('Ошибка:', err),
  complete: () => console.log('Поток завершён!')
});

// Вывод:
// Получили: Первое значение
// Получили: Второе значение
// Получили: Третье значение
// Поток завершён!

---

Observable vs Promise — главное отличие 🥊

// PROMISE — одно значение, запускается СРАЗУ
const promise = new Promise(resolve => {
  console.log('Promise запустился!'); // выполнится СРАЗУ
  setTimeout(() => resolve('Готово'), 1000);
});
// ↑ Уже запущен, даже без .then()!

// OBSERVABLE — ленивый, запускается только при подписке
const observable$ = new Observable(subscriber => {
  console.log('Observable запустился!'); // выполнится только при subscribe()
  setTimeout(() => {
    subscriber.next('Готово');
    subscriber.complete();
  }, 1000);
});
// ↑ Ещё ничего не произошло!

observable$.subscribe(value => console.log(value));
// Только теперь: 'Observable запустился!'
// Через секунду: 'Готово'

Ключевые отличия

// Promise: ОДНО значение
const p = Promise.resolve(42);
p.then(v => console.log(v)); // 42 — и всё

// Observable: МНОГО значений
import { interval } from 'rxjs';
const timer$ = interval(1000); // каждую секунду
timer$.subscribe(n => console.log(n)); // 0, 1, 2, 3... до бесконечности!

---

Жизненный цикл Observable 🔄

Создание → Подписка → Эмиссия значений → Завершение/Ошибка
   ↓           ↓              ↓                    ↓
new Observable  .subscribe()   .next(value)    .complete() / .error()

import { Observable } from 'rxjs';

const lifecycle$ = new Observable(subscriber => {
  console.log('1️⃣ Подписка создана, начинаем работу');

  subscriber.next('🟢 Значение 1');
  subscriber.next('🟢 Значение 2');

  // Асинхронное значение
  setTimeout(() => {
    subscriber.next('🟢 Значение 3 (async)');
    subscriber.complete(); // 🏁 Завершаем поток
    // После complete() — subscriber.next() игнорируется!
    subscriber.next('Это уже не придёт...');
  }, 1000);

  // Функция очистки — вызывается при unsubscribe()
  return () => {
    console.log('🧹 Очистка! Подписка отменена');
  };
});

const sub = lifecycle$.subscribe({
  next: v => console.log('Получили:', v),
  error: e => console.log('Ошибка:', e),
  complete: () => console.log('✅ Завершено!')
});

// Если хотим отписаться раньше:
// sub.unsubscribe(); // вызовет функцию очистки

---

Создание Observable: вспомогательные функции 🛠️

В реальном коде ты редко пишешь new Observable() вручную. Для этого есть creation operators:

of() — из готовых значений

import { of } from 'rxjs';

// Создаём Observable из любых значений
const fromValues$ = of(1, 2, 3);
fromValues$.subscribe(v => console.log(v)); // 1, 2, 3

const fromMixed$ = of('привет', 42, true, { name: 'Яша' });
fromMixed$.subscribe(v => console.log(v));
// 'привет', 42, true, { name: 'Яша' }
// Всё синхронно!

from() — из промисов, массивов, итерируемых

import { from } from 'rxjs';

// Из массива
const fromArray$ = from([10, 20, 30, 40]);
fromArray$.subscribe(v => console.log(v)); // 10, 20, 30, 40

// Из Promise!
const fetchUser = fetch('/api/user').then(r => r.json());
const fromPromise$ = from(fetchUser);
fromPromise$.subscribe({
  next: user => console.log('Пользователь:', user),
  error: err => console.log('Ошибка:', err)
});

// Из строки (итерируемой)
const fromString$ = from('RxJS');
fromString$.subscribe(v => console.log(v)); // R, x, J, S

interval() — таймер, каждые N миллисекунд

import { interval } from 'rxjs';
import { take } from 'rxjs/operators';

// Каждую секунду эмитирует: 0, 1, 2, 3...
const every1sec$ = interval(1000);

// Берём только первые 5 значений
every1sec$.pipe(
  take(5)
).subscribe({
  next: n => console.log(`Тик ${n}`),
  complete: () => console.log('5 тиков — стоп!')
});
// Тик 0 (через 1 сек)
// Тик 1 (через 2 сек)
// ... до Тик 4

timer() — задержка или отложенный интервал

import { timer } from 'rxjs';

// Одно значение через 2 секунды
const delayed$ = timer(2000);
delayed$.subscribe(v => console.log('После 2 сек:', v)); // 0

// Первое значение через 1 сек, потом каждые 500мс
const delayedInterval$ = timer(1000, 500);
delayedInterval$.pipe(take(4)).subscribe(v => console.log(v));
// (через 1000ms) 0
// (через 1500ms) 1
// (через 2000ms) 2
// (через 2500ms) 3

fromEvent() — из DOM событий

import { fromEvent } from 'rxjs';

// Клики по кнопке
const clicks$ = fromEvent(document.getElementById('btn'), 'click');
clicks$.subscribe(event => console.log('Клик!', event));

// Ввод в поле
const input$ = fromEvent(document.getElementById('search'), 'input');
input$.subscribe(e => console.log('Ввод:', e.target.value));

---

Marble Diagrams — визуализация потоков 🎨

Marble diagrams — это способ рисовать Observable на временной шкале:

Время:    ─────────────────────────────────→
of(1,2,3): (1 2 3|)   ← все сразу, | это complete

interval(1000): ──0──1──2──3──4──...→   ← каждую секунду

timer(2000):   ──────0|   ← через 2 сек, одно значение и complete

---

Важно: каждая подписка — отдельный поток! 🔑

import { Observable } from 'rxjs';

let count = 0;
const counter$ = new Observable(sub => {
  count++;
  console.log(`Создан поток #${count}`);
  sub.next(count);
  sub.complete();
});

// Каждая подписка запускает функцию-исполнитель заново!
counter$.subscribe(v => console.log('Подписчик 1:', v)); // Поток #1, значение 1
counter$.subscribe(v => console.log('Подписчик 2:', v)); // Поток #2, значение 2
counter$.subscribe(v => console.log('Подписчик 3:', v)); // Поток #3, значение 3

🧠 Observable — холодный по умолчанию. Каждый подписчик получает свой независимый поток. О горячих Observable поговорим в теме “Subjects”!

---

Практика

Попробуйте примеры в интерактивном редакторе: