Что такое Subject?

Subject: Издатель и подписчик в одном

Subject — это один из ключевых объектов в RxDart (Reactive Extensions для Dart), который комбинирует возможности Sink (потребитель) и Stream (издатель). Subject — это мост между кодом, который генерирует события, и кодом, который их потребляет.

Основное понимание

В обычном потоке данных в Dart:

┌──────────┐      ┌─────────────┐      ┌───────────┐
│ Publisher│ --→ │   Stream    │ --→ │ Subscriber│
└──────────┘      └─────────────┘      └───────────┘

С Subject'ом:

┌──────────┐                          ┌───────────┐
│Publisher │ --→                      │Subscriber │
└──────────┘  \                        ↑ /
              \ Subject (Sink + Stream)  /
               \                        /
                ←──────────────────────
     
 Результат: Publisher отправляет события → Subject слушает →
            Subject отправляет их подписчикам

Subject активно передаёт события (push model), в отличие от обычного Stream'а, который создаётся с функцией-генератором.

Зачем нужен Subject

// ❌ БЕЗ Subject — нужно вручную управлять StreamController
final controller = StreamController<int>();

controller.add(1);
controller.add(2);
controller.add(3);

controller.stream.listen(print);  // Выведет: 1, 2, 3
controller.close();

// ✅ С Subject — проще и понятнее
final subject = PublishSubject<int>();

subject.add(1);
subject.add(2);
subject.add(3);

subject.stream.listen(print);  // Тоже выведет: 1, 2, 3
subject.close();

Subject лучше для:

• State management
• Event management
• Real-time приложений
• Кэширования значений

Типы Subject'ов

1. PublishSubject — самый базовый

Не хранит значения, просто транслирует события подписчикам.

final subject = PublishSubject<int>();

subject.add(1);
subject.add(2);

// Подписка ДО этого момента НЕ получит 1 и 2
subject.listen((value) {
  print('Listener 1: $value');
});

subject.add(3);
// Выведет: "Listener 1: 3"

// Вторая подписка также только получит новые события
subject.listen((value) {
  print('Listener 2: $value');
});

subject.add(4);
// Выведет: "Listener 1: 4" и "Listener 2: 4"

Практический пример: User tap события

class ButtonEventService {
  final _buttonTaps = PublishSubject<void>();
  
  Stream<void> get onButtonTap => _buttonTaps.stream;
  
  void notifyButtonTap() {
    _buttonTaps.add(null);
  }
  
  void dispose() {
    _buttonTaps.close();
  }
}

// Использование
final service = ButtonEventService();

service.onButtonTap.listen((_) {
  print('Button tapped!');
});

service.notifyButtonTap();  // Выведет: "Button tapped!"

2. BehaviorSubject — хранит последнее значение

Мы уже разбирали это подробно, но вкратце:

final subject = BehaviorSubject<int>(seededValue: 0);

subject.add(1);
subject.add(2);

// Новый подписчик сразу получит текущее значение (2)
subject.listen((value) {
  print('Value: $value');
});
// Выведет: "Value: 2" (текущее значение)

subject.add(3);
// Выведет: "Value: 3"

3. ReplaySubject — хранит последние N значений

Отправляет несколько последних значений новым подписчикам.

final subject = ReplaySubject<int>(maxSize: 3);

subject.add(1);
subject.add(2);
subject.add(3);
subject.add(4);
subject.add(5);

// Новый подписчик получит последние 3 значения (3, 4, 5)
subject.listen((value) {
  print('Value: $value');
});
// Выведет: "Value: 3", "Value: 4", "Value: 5"

subject.add(6);
// Выведет: "Value: 6"

Практический пример: История поиска

class SearchHistoryService {
  final _searchHistory = ReplaySubject<String>(maxSize: 5);
  
  Stream<List<String>> get history => _searchHistory.stream
      .scan<List<String>>([], (list, query) => [...list, query])
      .map((list) => list.toSet().toList());  // Unique
  
  void search(String query) {
    _searchHistory.add(query);
  }
  
  void dispose() {
    _searchHistory.close();
  }
}

4. StreamController vs Subject

// StreamController — низкоуровневый
final controller = StreamController<int>();
controller.add(1);
controller.stream.listen(print);

// Subject — высокоуровневый обертка
final subject = PublishSubject<int>();
subject.add(1);
subject.stream.listen(print);

// Difference: Subject автоматически manage'ит broadcast,
// обработку ошибок, и имеет удобный API

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

Пример 1: Простой event bus

class EventBus {
  // Subject для разных типов событий
  final _navigationEvents = PublishSubject<NavigationEvent>();
  final _userEvents = PublishSubject<UserEvent>();
  
  Stream<NavigationEvent> get navigation => _navigationEvents.stream;
  Stream<UserEvent> get user => _userEvents.stream;
  
  void emitNavigation(NavigationEvent event) {
    _navigationEvents.add(event);
  }
  
  void emitUser(UserEvent event) {
    _userEvents.add(event);
  }
  
  void dispose() {
    _navigationEvents.close();
    _userEvents.close();
  }
}

abstract class NavigationEvent {}
class GoToHome extends NavigationEvent {}
class GoToProfile extends NavigationEvent {}

abstract class UserEvent {}
class UserLoggedIn extends UserEvent {
  final String userId;
  UserLoggedIn(this.userId);
}

// Использование
final eventBus = EventBus();

eventBus.navigation.listen((event) {
  if (event is GoToHome) {
    Navigator.pushNamed(context, '/home');
  } else if (event is GoToProfile) {
    Navigator.pushNamed(context, '/profile');
  }
});

eventBus.emitNavigation(GoToHome());

Пример 2: Notification service

class NotificationService {
  final _notifications = PublishSubject<Notification>();
  
  Stream<Notification> get notifications => _notifications.stream;
  
  void showNotification(String title, String message) {
    _notifications.add(
      Notification(title: title, message: message)
    );
  }
  
  void dispose() {
    _notifications.close();
  }
}

class Notification {
  final String title;
  final String message;
  Notification({required this.title, required this.message});
}

// Widget использует это
class NotificationOverlay extends StatelessWidget {
  final NotificationService notificationService;
  
  const NotificationOverlay({required this.notificationService});
  
  @override
  Widget build(BuildContext context) {
    return StreamBuilder<Notification>(
      stream: notificationService.notifications,
      builder: (context, snapshot) {
        if (snapshot.hasData) {
          final notification = snapshot.data!;
          return Positioned(
            top: 50,
            left: 16,
            right: 16,
            child: Material(
              child: Container(
                padding: const EdgeInsets.all(16),
                decoration: BoxDecoration(
                  color: Colors.black87,
                  borderRadius: BorderRadius.circular(8),
                ),
                child: Column(
                  crossAxisAlignment: CrossAxisAlignment.start,
                  children: [
                    Text(
                      notification.title,
                      style: const TextStyle(
                        color: Colors.white,
                        fontWeight: FontWeight.bold,
                      ),
                    ),
                    const SizedBox(height: 4),
                    Text(
                      notification.message,
                      style: const TextStyle(color: Colors.white70),
                    ),
                  ],
                ),
              ),
            ),
          );
        }
        return const SizedBox.shrink();
      },
    );
  }
}

Пример 3: Комбинирование Subject'ов

class LoginService {
  final _username = BehaviorSubject<String>(seededValue: '');
  final _password = BehaviorSubject<String>(seededValue: '');
  final _isLoading = BehaviorSubject<bool>(seededValue: false);
  
  Stream<String> get username => _username.stream;
  Stream<String> get password => _password.stream;
  Stream<bool> get isLoading => _isLoading.stream;
  
  // Комбинированный stream для кнопки "Логин"
  late Stream<bool> canLogin = Rx.combineLatest2(
    username.map((u) => u.isNotEmpty),
    password.map((p) => p.length >= 6),
    (uValid, pValid) => uValid && pValid,
  );
  
  void setUsername(String value) => _username.add(value);
  void setPassword(String value) => _password.add(value);
  
  Future<void> login() async {
    _isLoading.add(true);
    try {
      await api.login(
        _username.value,
        _password.value,
      );
    } finally {
      _isLoading.add(false);
    }
  }
  
  void dispose() {
    _username.close();
    _password.close();
    _isLoading.close();
  }
}

// Widget
class LoginForm extends StatefulWidget {
  @override
  State<LoginForm> createState() => _LoginFormState();
}

class _LoginFormState extends State<LoginForm> {
  late LoginService loginService;
  
  @override
  void initState() {
    super.initState();
    loginService = LoginService();
  }
  
  @override
  Widget build(BuildContext context) {
    return Column(
      children: [
        StreamBuilder<String>(
          stream: loginService.username,
          builder: (context, snapshot) {
            return TextField(
              onChanged: loginService.setUsername,
              decoration: const InputDecoration(hintText: 'Username'),
            );
          },
        ),
        StreamBuilder<String>(
          stream: loginService.password,
          builder: (context, snapshot) {
            return TextField(
              onChanged: loginService.setPassword,
              obscureText: true,
              decoration: const InputDecoration(hintText: 'Password'),
            );
          },
        ),
        StreamBuilder<bool>(
          stream: loginService.canLogin,
          builder: (context, snapshot) {
            final canLogin = snapshot.data ?? false;
            return StreamBuilder<bool>(
              stream: loginService.isLoading,
              builder: (context, loadingSnapshot) {
                final isLoading = loadingSnapshot.data ?? false;
                return ElevatedButton(
                  onPressed: canLogin && !isLoading
                      ? loginService.login
                      : null,
                  child: isLoading
                      ? const CircularProgressIndicator()
                      : const Text('Login'),
                );
              },
            );
          },
        ),
      ],
    );
  }
  
  @override
  void dispose() {
    loginService.dispose();
    super.dispose();
  }
}

Subject vs Stream

// STREAM — создаётся с функцией-генератором
final stream = Stream.fromIterable([1, 2, 3]);
stream.listen(print);  // Выведет: 1, 2, 3
stream.listen(print);  // Выведет: 1, 2, 3 (создаёт новый поток)

// SUBJECT — явно управляемый издатель
final subject = PublishSubject<int>();
subject.add(1);
subject.add(2);
subject.add(3);
subject.listen(print);  // Пропустил события 1, 2, 3!
subject.add(4);
subject.listen(print);  // Пропустит события до подписки

// С BehaviorSubject
final behaviorSubject = BehaviorSubject<int>(seededValue: 0);
behaviorSubject.add(1);
behaviorSubject.add(2);
behaviorSubject.add(3);
behaviorSubject.listen(print);  // Сразу выведет: 3 (последнее значение)

Best Practices

✅ ИСПОЛЬЗУЙ Subject для:

1. Event management и user interactions
2. State management (особенно BehaviorSubject)
3. Real-time обновлений
4. Notification систем
5. Команд между компонентами

❌ НЕ ИСПОЛЬЗУЙ Subject для:

1. Simple async operations (используй Future)
2. Когда работаешь с нативными Stream API
3. Если нужна только одна подписка (используй Future)

Обязательно dispose!

class MyService {
  final _subject = BehaviorSubject<int>();
  
  void dispose() {
    _subject.close();  // ОБЯЗАТЕЛЬНО!
  }
}

Итог

Subject — это:

• Комбинация Sink (вход) и Stream (выход)
• Способ явно управлять потоком событий
• Essential для event-driven архитектуры
• Бывает несколько типов: PublishSubject, BehaviorSubject, ReplaySubject
• Требует careful management памяти (dispose)

Subject'ы делают Flutter приложение реактивным и гибким, позволяя легко синхронизировать различные части приложения.