В чем разница между RxDart Merge и Combine?

Разница между RxDart Merge и Combine

Merge и Combine — это два фундаментальных оператора в RxDart для работы с несколькими потоками (Stream). Хотя они похожи, их поведение существенно отличается.

Merge

Merge объединяет несколько потоков, испуская значения по мере их поступления.

Характеристики:

• Излучает значение СРАЗУ, как только оно поступит из любого потока
• Не ждет значений от других потоков
• Порядок значений зависит от порядка поступления
• Использует логику ИЛИ ("или это, или то")

import 'package:rxdart/rxdart.dart';

final stream1 = Stream.periodic(
  Duration(seconds: 1),
  (count) => 'Stream1: $count',
);

final stream2 = Stream.periodic(
  Duration(milliseconds: 600),
  (count) => 'Stream2: $count',
);

MergeStream([stream1, stream2]).listen((value) {
  print(value);
});

// Вывод (примерно):
// Stream2: 0
// Stream1: 0
// Stream2: 1
// Stream2: 2
// Stream1: 1
// Stream2: 3
// Stream2: 4
// Stream1: 2

Когда использовать Merge:

• Обработка нескольких независимых источников событий
• Кнопки разных компонентов (все отправляют в один поток)
• Множественные источники ошибок
• Логирование из разных сервисов

class UserPreferences {
  final settingsUpdatedStream = StreamController<void>();
  final cacheUpdatedStream = StreamController<void>();
  final networkUpdatedStream = StreamController<void>();

  void setupListeners() {
    Rx.merge([
      settingsUpdatedStream.stream,
      cacheUpdatedStream.stream,
      networkUpdatedStream.stream,
    ]).listen((_) {
      // Любой из источников вызвал обновление
      _refreshUI();
    });
  }
}

Combine (CombineLatest)

Combine объединяет несколько потоков, НО излучает новое значение ТОЛЬКО когда ВСЕ потоки выпустили хотя бы по одному значению.

Характеристики:

• Излучает новое значение ТОЛЬКО когда обновляется ЛЮБОЙ поток (при условии что все выпустили значение)
• Использует ПОСЛЕДНЕЕ значение от каждого потока
• Ждет, пока все потоки выпустят хотя бы один элемент
• Использует логику И ("это И то")

final age = Stream.periodic(
  Duration(seconds: 2),
  (count) => 20 + count,
);

final name = Stream.periodic(
  Duration(seconds: 1),
  (count) => 'User$count',
);

CombineLatestStream.combine2(age, name, (a, n) => {'age': a, 'name': n})
  .listen((value) {
    print(value);
  });

// Вывод:
// {} - ничего не выпущено (age еще не выпустил)
// {age: 20, name: User0} - оба выпустили!
// {age: 20, name: User1} - name обновился
// {age: 20, name: User2} - name обновился
// {age: 21, name: User2} - age обновился, используется последний name

Сравнительная таблица

Параметр         | Merge          | Combine
─────────────────┼────────────────┼─────────────────
Услов. для выпуска | Любой поток | Все потоки
Данные для выпуска| Одно значение  | Кортеж всех
Порядок          | По времени      | Синхронизированный
Ждание первого   | Нет            | Да (кроме первого)
Использование    | События        | Состояние

Практические примеры в Flutter

Merge для событий (независимые действия):

class NotificationCenter {
  final _clicks = StreamController<String>();
  final _errors = StreamController<String>();
  final _systemEvents = StreamController<String>();

  Stream<String> get allNotifications {
    return Rx.merge([
      _clicks.stream,
      _errors.stream,
      _systemEvents.stream,
    ]);
  }

  void init() {
    allNotifications.listen((notification) {
      print('Уведомление: $notification');
    });
  }
}

Combine для формы (зависимые поля):

class LoginFormBloc {
  final _emailController = StreamController<String>();
  final _passwordController = StreamController<String>();

  Stream<String> get email => _emailController.stream;
  Stream<String> get password => _passwordController.stream;

  Stream<bool> get isFormValid {
    return Rx.combineLatest2(
      email,
      password,
      (e, p) => _isValidEmail(e) && _isValidPassword(p),
    );
  }

  bool _isValidEmail(String email) => email.contains('@');
  bool _isValidPassword(String password) => password.length >= 6;

  void dispose() {
    _emailController.close();
    _passwordController.close();
  }
}

// Использование
final bloc = LoginFormBloc();
bloc.isFormValid.listen((isValid) {
  print('Форма валидна: $isValid');
});

bloc._emailController.add('user@example.com'); // Нет выпуска (password не выпустил)
bloc._passwordController.add('password123');   // Выпуск: {true}
bloc._emailController.add('invalid');           // Выпуск: {false}

Combine для фильтра и сортировки:

class ProductsBloc {
  final _searchController = StreamController<String>();
  final _categoryController = StreamController<String>();
  final _sortController = StreamController<String>();

  Stream<List<Product>> get products {
    return Rx.combineLatest3(
      _searchController.stream,
      _categoryController.stream,
      _sortController.stream,
      (search, category, sort) {
        return _filterAndSort(
          search: search,
          category: category,
          sort: sort,
        );
      },
    );
  }

  List<Product> _filterAndSort({
    required String search,
    required String category,
    required String sort,
  }) {
    // Реализация фильтрации
    var result = allProducts
        .where((p) => p.name.contains(search))
        .where((p) => p.category == category);

    if (sort == 'price_asc') {
      result = result.toList()..sort((a, b) => a.price.compareTo(b.price));
    } else if (sort == 'price_desc') {
      result = result.toList()..sort((a, b) => b.price.compareTo(a.price));
    }

    return result.toList();
  }
}

Выбор между Merge и Combine

Используй Merge когда:

• События независимы
• Кнопки, клики, жесты
• Множественные источники ошибок
• Вам не нужна синхронизация

Используй Combine когда:

• Значения зависят друг от друга
• Форма с несколькими полями
• Фильтрация по нескольким критериям
• Нужна синхронизация состояния

RxDart в современном Flutter

В последнее время многие разработчики предпочитают Provider, Riverpod или Bloc Pattern вместо Raw RxDart. Однако понимание Merge и Combine остается критичным для:

• Работы с несколькими асинхронными источниками
• Сложной логики реактивного программирования
• Миграции старых кодовых баз
• Работы с WebSocket и Real-time данными

Вывод: Merge для параллельных событий, Combine для синхронизированного состояния.