Приведи пример использования StreamBuilder

StreamBuilder в Flutter: работа с асинхронными потоками данных

StreamBuilder — это виджет, позволяющий строить UI на основе Stream данных. Это одна из самых важных техник реактивного программирования в Flutter.

Что такое StreamBuilder

StreamBuilder — это виджет, который "слушает" Stream и перестраивается каждый раз, когда в Stream поступляют новые данные. Это идеально подходит для реактивного обновления UI.

// Базовая структура
StreamBuilder<T>(
  stream: myStream, // Stream данных
  initialData: initialValue, // начальное значение (опционально)
  builder: (context, snapshot) {
    // snapshot содержит текущее состояние Stream
    if (snapshot.connectionState == ConnectionState.waiting) {
      return CircularProgressIndicator(); // загрузка
    } else if (snapshot.hasError) {
      return Text('Error: ${snapshot.error}'); // ошибка
    } else if (snapshot.hasData) {
      return Text('Data: ${snapshot.data}'); // данные готовы
    } else {
      return Text('No data'); // нет данных
    }
  },
)

Как работает Stream

Stream — это асинхронная последовательность событий (data events или error events). Думайте о нём как о потоке данных со временем.

// Stream можно создать несколькими способами

// 1. Из List через Stream.fromIterable
var myStream = Stream.fromIterable([1, 2, 3, 4, 5]);

// 2. С задержкой через Stream.periodic
var tickStream = Stream.periodic(Duration(seconds: 1), (count) => count);

// 3. Вручную через StreamController
var controller = StreamController<int>();
controller.add(1);
controller.add(2);
controller.close();

var myStream = controller.stream;

Практический пример: загрузка данных пользователя

class UserRepository {
  // Возвращаем Stream вместо Future для continuous обновлений
  Stream<User> getUser(String userId) async* {
    try {
      // Первый emit — загружаем базовые данные
      final response = await http.get(
        Uri.parse('https://api.example.com/users/$userId'),
      );
      
      if (response.statusCode == 200) {
        final user = User.fromJson(jsonDecode(response.body));
        yield user; // первое значение в Stream
      } else {
        throw Exception('Failed to load user');
      }
    } catch (e) {
      // Если ошибка, Stream автоматически её обработает
      rethrow;
    }
  }
}

// Использование StreamBuilder
class UserProfileScreen extends StatelessWidget {
  final String userId;
  final userRepository = UserRepository();

  UserProfileScreen({required this.userId});

  @override
  Widget build(BuildContext context) {
    return StreamBuilder<User>(
      stream: userRepository.getUser(userId),
      builder: (context, snapshot) {
        if (snapshot.connectionState == ConnectionState.waiting) {
          return Scaffold(
            appBar: AppBar(title: const Text('Profile')),
            body: const Center(child: CircularProgressIndicator()),
          );
        } else if (snapshot.hasError) {
          return Scaffold(
            appBar: AppBar(title: const Text('Profile')),
            body: Center(
              child: Text('Error: ${snapshot.error}'),
            ),
          );
        } else if (snapshot.hasData) {
          final user = snapshot.data!;
          return Scaffold(
            appBar: AppBar(title: Text(user.name)),
            body: Center(
              child: Column(
                mainAxisAlignment: MainAxisAlignment.center,
                children: [
                  Text('Name: ${user.name}'),
                  Text('Email: ${user.email}'),
                  Text('Age: ${user.age}'),
                ],
              ),
            ),
          );
        } else {
          return Scaffold(
            appBar: AppBar(title: const Text('Profile')),
            body: const Center(child: Text('No data')),
          );
        }
      },
    );
  }
}

Пример с real-time обновлениями: счётчик

class CounterService {
  final _counterController = StreamController<int>();

  // Expose Stream для слушания
  Stream<int> get counterStream => _counterController.stream;

  int _count = 0;

  void increment() {
    _count++;
    _counterController.add(_count); // emit новое значение
  }

  void decrement() {
    _count--;
    _counterController.add(_count);
  }

  void reset() {
    _count = 0;
    _counterController.add(_count);
  }

  void dispose() {
    _counterController.close();
  }
}

class CounterApp extends StatefulWidget {
  @override
  State<CounterApp> createState() => _CounterAppState();
}

class _CounterAppState extends State<CounterApp> {
  late CounterService _counterService;

  @override
  void initState() {
    super.initState();
    _counterService = CounterService();
  }

  @override
  void dispose() {
    _counterService.dispose();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('Counter')),
      body: Center(
        child: StreamBuilder<int>(
          stream: _counterService.counterStream,
          initialData: 0, // начальное значение
          builder: (context, snapshot) {
            return Column(
              mainAxisAlignment: MainAxisAlignment.center,
              children: [
                Text(
                  'Count: ${snapshot.data}',
                  style: Theme.of(context).textTheme.headlineMedium,
                ),
                const SizedBox(height: 20),
                Row(
                  mainAxisAlignment: MainAxisAlignment.center,
                  children: [
                    ElevatedButton(
                      onPressed: _counterService.decrement,
                      child: const Text('Decrement'),
                    ),
                    const SizedBox(width: 20),
                    ElevatedButton(
                      onPressed: _counterService.reset,
                      child: const Text('Reset'),
                    ),
                    const SizedBox(width: 20),
                    ElevatedButton(
                      onPressed: _counterService.increment,
                      child: const Text('Increment'),
                    ),
                  ],
                ),
              ],
            );
          },
        ),
      ),
    );
  }
}

Поиск в реальном времени (с debounce)

class SearchService {
  final _searchController = StreamController<String>();
  final _resultsController = StreamController<List<String>>();

  Stream<List<String>> get resultsStream => _resultsController.stream;

  SearchService() {
    // Слушаем поисковые запросы с debounce
    _searchController.stream
        .debounceTime(Duration(milliseconds: 300)) // ждём 300ms без ввода
        .distinct() // игнорируем дублирующиеся запросы
        .asyncMap((query) => _performSearch(query))
        .listen(
          (results) => _resultsController.add(results),
          onError: (error) => _resultsController.addError(error),
        );
  }

  void search(String query) {
    _searchController.add(query);
  }

  Future<List<String>> _performSearch(String query) async {
    if (query.isEmpty) return [];
    
    // Имитируем API запрос
    await Future.delayed(Duration(milliseconds: 500));
    
    final allItems = [
      'Flutter',
      'Dart',
      'Firebase',
      'Provider',
      'Riverpod',
    ];
    
    return allItems
        .where((item) => item.toLowerCase().contains(query.toLowerCase()))
        .toList();
  }

  void dispose() {
    _searchController.close();
    _resultsController.close();
  }
}

class SearchScreen extends StatefulWidget {
  @override
  State<SearchScreen> createState() => _SearchScreenState();
}

class _SearchScreenState extends State<SearchScreen> {
  late SearchService _searchService;
  late TextEditingController _textController;

  @override
  void initState() {
    super.initState();
    _searchService = SearchService();
    _textController = TextEditingController();
    _textController.addListener(() {
      _searchService.search(_textController.text);
    });
  }

  @override
  void dispose() {
    _searchService.dispose();
    _textController.dispose();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('Search')),
      body: Padding(
        padding: const EdgeInsets.all(16.0),
        child: Column(
          children: [
            TextField(
              controller: _textController,
              decoration: InputDecoration(
                hintText: 'Search...',
                prefixIcon: const Icon(Icons.search),
                border: OutlineInputBorder(
                  borderRadius: BorderRadius.circular(8),
                ),
              ),
            ),
            const SizedBox(height: 16),
            Expanded(
              child: StreamBuilder<List<String>>(
                stream: _searchService.resultsStream,
                builder: (context, snapshot) {
                  if (snapshot.connectionState == ConnectionState.waiting) {
                    return const Center(child: CircularProgressIndicator());
                  } else if (snapshot.hasError) {
                    return Center(
                      child: Text('Error: ${snapshot.error}'),
                    );
                  } else if (snapshot.hasData && snapshot.data!.isNotEmpty) {
                    return ListView.builder(
                      itemCount: snapshot.data!.length,
                      itemBuilder: (context, index) {
                        return ListTile(
                          title: Text(snapshot.data![index]),
                        );
                      },
                    );
                  } else {
                    return const Center(
                      child: Text('No results found'),
                    );
                  }
                },
              ),
            ),
          ],
        ),
      ),
    );
  }
}

Мониторинг WebSocket соединения

class WebSocketService {
  late WebSocket _webSocket;
  final _messageController = StreamController<String>.broadcast();

  Stream<String> get messageStream => _messageController.stream;

  Future<void> connect(String url) async {
    try {
      _webSocket = await WebSocket.connect(url);
      _webSocket.listen(
        (data) => _messageController.add(data), // новое сообщение
        onError: (error) => _messageController.addError(error),
        onDone: () => _messageController.close(),
      );
    } catch (e) {
      _messageController.addError(e);
    }
  }

  void send(String message) {
    _webSocket.add(message);
  }

  void dispose() {
    _webSocket.close();
    _messageController.close();
  }
}

class ChatScreen extends StatefulWidget {
  @override
  State<ChatScreen> createState() => _ChatScreenState();
}

class _ChatScreenState extends State<ChatScreen> {
  late WebSocketService _wsService;

  @override
  void initState() {
    super.initState();
    _wsService = WebSocketService();
    _wsService.connect('ws://echo.websocket.org');
  }

  @override
  void dispose() {
    _wsService.dispose();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('Chat')),
      body: StreamBuilder<String>(
        stream: _wsService.messageStream,
        builder: (context, snapshot) {
          if (snapshot.connectionState == ConnectionState.waiting) {
            return const Center(child: CircularProgressIndicator());
          } else if (snapshot.hasError) {
            return Center(
              child: Text('Connection error: ${snapshot.error}'),
            );
          } else if (snapshot.hasData) {
            return Center(
              child: Text('Received: ${snapshot.data}'),
            );
          } else {
            return const Center(child: Text('No messages'));
          }
        },
      ),
    );
  }
}

Stream vs StreamBuilder vs FutureBuilder

// FutureBuilder — для одноразовых асинхронных операций
FutureBuilder<List<User>>(
  future: userRepository.getUsers(), // один раз
  builder: (context, snapshot) { ... },
)

// StreamBuilder — для continuous потока данных
StreamBuilder<List<User>>(
  stream: userRepository.getUsersStream(), // постоянно обновляется
  builder: (context, snapshot) { ... },
)

Best Practices

Всегда закрывайте StreamController:

@override
void dispose() {
  _controller.close(); // обязательно!
  super.dispose();
}

Используйте .broadcast() для множественных слушателей:

// ❌ Плохо — не может быть двух слушателей
final stream = Stream.periodic(Duration(seconds: 1));

// ✅ Хорошо — может быть множество слушателей
final stream = Stream.periodic(Duration(seconds: 1)).asBroadcastStream();
// или
final controller = StreamController<int>.broadcast();

Используйте RxDart для удобных операций:

import 'package:rxdart/rxdart.dart';

// debounce, throttle, switchMap и другие операторы
searchStream
    .debounceTime(Duration(milliseconds: 300))
    .switchMap((query) => _performSearch(query))
    .listen((results) => updateUI(results));

Выводы

StreamBuilder идеален для:

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

Это один из самых мощных инструментов для создания реактивных и отзывчивых приложений.