Flutter Riverpod 이해하기 4편

StreamProvider란?

StreamProvider는 Riverpod에서 지속적으로 변하는 데이터를 관리하는 도구입니다. 한 번 실행되고 끝나는 FutureProvider와 달리, StreamProvider는 시간이 지나면서 계속해서 새로운 값을 받을 수 있어요.

쉽게 말해서 "실시간으로 계속 업데이트되는 데이터를 위젯에서 편리하게 사용할 수 있게 해주는 도구"입니다. 새로운 데이터가 들어올 때마다 관련된 위젯들을 자동으로 다시 그려줍니다.

StreamProvider는 Stream을 구독해서 새로운 값이 나올 때마다 상태를 업데이트하고, 위젯이 사라지면 자동으로 구독을 해제해서 메모리 누수를 방지해줍니다.

언제 StreamProvider를 사용하는가?

StreamProvider는 지속적으로 변하는 데이터를 다룰 때 사용해야 합니다.

사용하기 좋은 경우:
• 실시간 채팅 메시지 (새 메시지가 계속 들어옴)
• 위치 추적 (GPS 좌표가 계속 업데이트됨)
• 주식 가격, 암호화폐 시세 (실시간으로 변함)
• Firebase 실시간 데이터베이스 (데이터가 실시간으로 동기화됨)
• WebSocket 연결 (서버에서 지속적으로 데이터 전송)
• 타이머, 스톱워치 (시간이 계속 흐름)
• 센서 데이터 (가속도계, 자이로스코프 등)

사용하면 안 되는 경우:
• 한 번만 가져오면 되는 데이터 (사용자 프로필, 설정값)
• 사용자가 직접 변경하는 상태 (토글, 카운터)
• 정적인 데이터 (앱 버전, 상수값)

한 번 요청해서 끝나는 데이터라면 FutureProvider를, 사용자가 직접 조작하는 상태라면 StateProvider를 사용해야 합니다.

기본 사용법

1. StreamProvider 만들기

Dart

// 현재 시간 스트림
final currentTimeProvider = StreamProvider<DateTime>((ref) {
  return Stream.periodic(Duration(seconds: 1), (_) => DateTime.now());
});

// 카운터 스트림  
final counterStreamProvider = StreamProvider<int>((ref) {
  return Stream.periodic(Duration(seconds: 1), (count) => count);
});

// Firebase 실시간 데이터
final messagesProvider = StreamProvider<List<Message>>((ref) {
  return FirebaseFirestore.instance
      .collection('messages')
      .orderBy('timestamp')
      .snapshots()
      .map((snapshot) => snapshot.docs
          .map((doc) => Message.fromFirestore(doc))
          .toList());
});

StreamProvider는 Stream을 반환하는 함수를 받아요. Stream.periodic으로 주기적인 데이터를 만들거나, Firebase나 WebSocket 같은 외부 소스의 Stream을 연결할 수 있습니다.

2. StreamProvider 사용하기

Dart

class CurrentTimeWidget extends ConsumerWidget {
  @override
  Widget build(BuildContext context, WidgetRef ref) {
    final timeAsync = ref.watch(currentTimeProvider);
    
    return timeAsync.when(
      loading: () => Text('시간 로딩 중...'),
      error: (error, stack) => Text('에러: $error'),
      data: (time) => Text(
        '현재 시간: ${time.hour}:${time.minute}:${time.second}',
        style: TextStyle(fontSize: 24),
      ),
    );
  }
}

StreamProvider도 FutureProvider처럼 AsyncValue를 반환해요. when() 메소드로 로딩, 에러, 데이터 상태를 처리할 수 있습니다. 새로운 데이터가 들어올 때마다 자동으로 위젯이 다시 그려집니다.

3. 매개변수가 있는 StreamProvider

Dart

// 특정 채팅방의 메시지 스트림
final chatMessagesProvider = StreamProvider.family<List<Message>, String>((ref, roomId) {
  return FirebaseFirestore.instance
      .collection('chatRooms')
      .doc(roomId)
      .collection('messages')
      .orderBy('timestamp', descending: true)
      .snapshots()
      .map((snapshot) => snapshot.docs
          .map((doc) => Message.fromFirestore(doc))
          .toList());
});

// 사용할 때
class ChatRoom extends ConsumerWidget {
  final String roomId;
  
  ChatRoom({required this.roomId});
  
  @override
  Widget build(BuildContext context, WidgetRef ref) {
    final messagesAsync = ref.watch(chatMessagesProvider(roomId));
    
    return messagesAsync.when(
      loading: () => Center(child: CircularProgressIndicator()),
      error: (error, stack) => Text('메시지를 불러올 수 없습니다'),
      data: (messages) => ListView.builder(
        reverse: true,
        itemCount: messages.length,
        itemBuilder: (context, index) {
          final message = messages[index];
          return ListTile(
            title: Text(message.text),
            subtitle: Text(message.senderName),
          );
        },
      ),
    );
  }
}

family 키워드를 사용하면 매개변수를 받는 StreamProvider를 만들 수 있어요. 각 매개변수마다 독립적인 Stream이 생성됩니다.

실무 예제

실시간 채팅 시스템

Dart

// 채팅방 목록 스트림
final chatRoomsProvider = StreamProvider<List<ChatRoom>>((ref) {
  return FirebaseFirestore.instance
      .collection('chatRooms')
      .where('participants', arrayContains: getCurrentUserId())
      .snapshots()
      .map((snapshot) => snapshot.docs
          .map((doc) => ChatRoom.fromFirestore(doc))
          .toList());
});

// 특정 채팅방의 메시지 스트림
final chatMessagesProvider = StreamProvider.family<List<ChatMessage>, String>((ref, roomId) {
  return FirebaseFirestore.instance
      .collection('chatRooms')
      .doc(roomId)
      .collection('messages')
      .orderBy('timestamp', descending: true)
      .limit(50)
      .snapshots()
      .map((snapshot) => snapshot.docs
          .map((doc) => ChatMessage.fromFirestore(doc))
          .toList());
});

// 온라인 사용자 상태 스트림
final onlineUsersProvider = StreamProvider.family<List<String>, String>((ref, roomId) {
  return FirebaseFirestore.instance
      .collection('chatRooms')
      .doc(roomId)
      .collection('presence')
      .where('isOnline', isEqualTo: true)
      .snapshots()
      .map((snapshot) => snapshot.docs
          .map((doc) => doc.id)
          .toList());
});

// 채팅 화면
class ChatScreen extends ConsumerWidget {
  final String roomId;
  
  ChatScreen({required this.roomId});
  
  @override
  Widget build(BuildContext context, WidgetRef ref) {
    final messagesAsync = ref.watch(chatMessagesProvider(roomId));
    final onlineUsersAsync = ref.watch(onlineUsersProvider(roomId));
    
    return Scaffold(
      appBar: AppBar(
        title: Text('채팅방'),
        actions: [
          onlineUsersAsync.when(
            loading: () => SizedBox.shrink(),
            error: (_, __) => SizedBox.shrink(),
            data: (users) => Chip(
              label: Text('접속자 ${users.length}명'),
              backgroundColor: Colors.green.withOpacity(0.1),
            ),
          ),
        ],
      ),
      body: messagesAsync.when(
        loading: () => Center(child: CircularProgressIndicator()),
        error: (error, stack) => Center(
          child: Text('메시지를 불러올 수 없습니다: $error'),
        ),
        data: (messages) => ListView.builder(
          reverse: true,
          itemCount: messages.length,
          itemBuilder: (context, index) {
            final message = messages[index];
            return MessageBubble(message: message);
          },
        ),
      ),
    );
  }
}

실시간 채팅에서는 여러 개의 StreamProvider를 조합해서 메시지, 온라인 상태, 채팅방 정보를 각각 관리해요.

위치 추적 앱

Dart

// 현재 위치 스트림
final locationProvider = StreamProvider<Position>((ref) async* {
  // 위치 권한 확인
  LocationPermission permission = await Geolocator.checkPermission();
  if (permission == LocationPermission.denied) {
    permission = await Geolocator.requestPermission();
  }
  
  if (permission == LocationPermission.deniedForever) {
    throw Exception('위치 권한이 필요합니다');
  }
  
  // 위치 스트림 시작
  yield* Geolocator.getPositionStream(
    locationSettings: LocationSettings(
      accuracy: LocationAccuracy.high,
      distanceFilter: 10, // 10미터 이동 시에만 업데이트
    ),
  );
});

// 이동 거리 계산 스트림
final travelDistanceProvider = StreamProvider<double>((ref) {
  double totalDistance = 0.0;
  Position? lastPosition;
  
  return ref.watch(locationProvider.stream).map((position) {
    if (lastPosition != null) {
      final distance = Geolocator.distanceBetween(
        lastPosition!.latitude,
        lastPosition!.longitude,
        position.latitude,
        position.longitude,
      );
      totalDistance += distance;
    }
    lastPosition = position;
    return totalDistance;
  });
});

// 위치 추적 화면
class LocationTracker extends ConsumerWidget {
  @override
  Widget build(BuildContext context, WidgetRef ref) {
    final locationAsync = ref.watch(locationProvider);
    final distanceAsync = ref.watch(travelDistanceProvider);
    
    return Scaffold(
      appBar: AppBar(title: Text('위치 추적')),
      body: Column(
        children: [
          Card(
            child: Padding(
              padding: EdgeInsets.all(16),
              child: Column(
                children: [
                  locationAsync.when(
                    loading: () => Text('위치 찾는 중...'),
                    error: (error, stack) => Text('위치 오류: $error'),
                    data: (position) => Column(
                      children: [
                        Text('위도: ${position.latitude.toStringAsFixed(6)}'),
                        Text('경도: ${position.longitude.toStringAsFixed(6)}'),
                        Text('정확도: ${position.accuracy.toStringAsFixed(1)}m'),
                      ],
                    ),
                  ),
                  SizedBox(height: 16),
                  distanceAsync.when(
                    loading: () => Text('거리 계산 중...'),
                    error: (error, stack) => Text('거리 오류: $error'),
                    data: (distance) => Text(
                      '총 이동거리: ${(distance / 1000).toStringAsFixed(2)}km',
                      style: TextStyle(fontSize: 18, fontWeight: FontWeight.bold),
                    ),
                  ),
                ],
              ),
            ),
          ),
        ],
      ),
    );
  }
}

위치 추적에서는 GPS 데이터가 계속 업데이트되므로 StreamProvider가 적합해요. 여러 StreamProvider를 조합해서 위치와 이동거리를 동시에 관리할 수 있습니다.

실시간 주식 시세

Dart

// 주식 시세 스트림 (WebSocket 사용)
final stockPriceProvider = StreamProvider.family<StockPrice, String>((ref, symbol) async* {
  final channel = WebSocketChannel.connect(
    Uri.parse('wss://api.example.com/stocks/$symbol'),
  );
  
  try {
    await for (final message in channel.stream) {
      final data = jsonDecode(message);
      yield StockPrice.fromJson(data);
    }
  } finally {
    await channel.sink.close();
  }
});

// 여러 종목 감시
final watchlistProvider = StreamProvider<Map<String, StockPrice>>((ref) async* {
  final symbols = ['AAPL', 'GOOGL', 'MSFT', 'TSLA'];
  final Map<String, StockPrice> prices = {};
  
  // 각 종목의 스트림을 구독
  final subscriptions = symbols.map((symbol) {
    return ref.watch(stockPriceProvider(symbol).stream).listen((price) {
      prices[symbol] = price;
    });
  }).toList();
  
  // 주기적으로 현재 가격 맵 방출
  yield* Stream.periodic(Duration(milliseconds: 500), (_) => Map.from(prices));
});

// 주식 시세 화면
class StockPriceScreen extends ConsumerWidget {
  @override
  Widget build(BuildContext context, WidgetRef ref) {
    final watchlistAsync = ref.watch(watchlistProvider);
    
    return Scaffold(
      appBar: AppBar(title: Text('실시간 주식 시세')),
      body: watchlistAsync.when(
        loading: () => Center(child: CircularProgressIndicator()),
        error: (error, stack) => Center(child: Text('연결 오류: $error')),
        data: (prices) => ListView.builder(
          itemCount: prices.length,
          itemBuilder: (context, index) {
            final symbol = prices.keys.elementAt(index);
            final stockPrice = prices[symbol]!;
            
            return ListTile(
              title: Text(symbol),
              subtitle: Text(
                '달러 ${stockPrice.currentPrice.toStringAsFixed(2)}',
                style: TextStyle(
                  color: stockPrice.change > 0 ? Colors.green : Colors.red,
                  fontWeight: FontWeight.bold,
                ),
              ),
              trailing: Text(
                '${stockPrice.change > 0 ? '+' : ''}${stockPrice.change.toStringAsFixed(2)}',
                style: TextStyle(
                  color: stockPrice.change > 0 ? Colors.green : Colors.red,
                ),
              ),
            );
          },
        ),
      ),
    );
  }
}

WebSocket을 통한 실시간 데이터 수신도 StreamProvider로 쉽게 처리할 수 있어요. 여러 종목을 동시에 모니터링하는 것도 가능합니다.

StreamProvider의 특별한 기능들

Stream 변환과 필터링

Dart

// 기본 카운터 스트림
final counterStreamProvider = StreamProvider<int>((ref) {
  return Stream.periodic(Duration(seconds: 1), (count) => count);
});

// 짝수만 필터링
final evenNumbersProvider = StreamProvider<int>((ref) {
  return ref.watch(counterStreamProvider.stream)
      .where((number) => number % 2 == 0);
});

// 10으로 나눈 나머지
final moduloProvider = StreamProvider<int>((ref) {
  return ref.watch(counterStreamProvider.stream)
      .map((number) => number % 10);
});

// 최근 5개 값의 평균
final averageProvider = StreamProvider<double>((ref) {
  final List<int> recentValues = [];
  
  return ref.watch(counterStreamProvider.stream).map((value) {
    recentValues.add(value);
    if (recentValues.length > 5) {
      recentValues.removeAt(0);
    }
    
    return recentValues.reduce((a, b) => a + b) / recentValues.length;
  });
});

Stream의 map, where, take 같은 메소드로 데이터를 변환하고 필터링할 수 있어요. 하나의 기본 Stream에서 여러 개의 파생된 StreamProvider를 만들 수 있습니다.

에러 처리와 재연결

Dart

// 자동 재연결이 있는 WebSocket 스트림
final resilientWebSocketProvider = StreamProvider<String>((ref) async* {
  int retryCount = 0;
  const maxRetries = 5;
  const baseDelay = Duration(seconds: 2);
  
  while (retryCount < maxRetries) {
    try {
      final channel = WebSocketChannel.connect(
        Uri.parse('wss://api.example.com/data'),
      );
      
      await for (final message in channel.stream) {
        retryCount = 0; // 연결 성공 시 재시도 카운트 리셋
        yield message;
      }
    } catch (e) {
      retryCount++;
      if (retryCount >= maxRetries) {
        throw Exception('최대 재시도 횟수 초과: $e');
      }
      
      // 지수적 백오프로 재연결 대기
      final delay = Duration(
        milliseconds: baseDelay.inMilliseconds * (1 << retryCount),
      );
      await Future.delayed(delay);
    }
  }
});

// 에러 처리가 있는 위젯
class ResilientDataWidget extends ConsumerWidget {
  @override
  Widget build(BuildContext context, WidgetRef ref) {
    final dataAsync = ref.watch(resilientWebSocketProvider);
    
    return dataAsync.when(
      loading: () => Column(
        mainAxisAlignment: MainAxisAlignment.center,
        children: [
          CircularProgressIndicator(),
          SizedBox(height: 16),
          Text('서버에 연결 중...'),
        ],
      ),
      error: (error, stack) => Column(
        mainAxisAlignment: MainAxisAlignment.center,
        children: [
          Icon(Icons.error, color: Colors.red, size: 48),
          SizedBox(height: 16),
          Text('연결 실패: $error'),
          SizedBox(height: 16),
          ElevatedButton(
            onPressed: () => ref.refresh(resilientWebSocketProvider),
            child: Text('다시 연결'),
          ),
        ],
      ),
      data: (data) => Column(
        children: [
          Icon(Icons.wifi, color: Colors.green),
          Text('데이터: $data'),
        ],
      ),
    );
  }
}

네트워크 연결이 불안정할 수 있으므로 자동 재연결과 에러 복구 로직을 포함하는 것이 좋아요.

주의사항

메모리 누수 방지

Dart

// ❌ 이렇게 하지 마세요 - 수동으로 구독 관리
class BadWidget extends StatefulWidget {
  @override
  _BadWidgetState createState() => _BadWidgetState();
}

class _BadWidgetState extends State<BadWidget> {
  late StreamSubscription subscription;
  
  @override
  void initState() {
    super.initState();
    // 직접 스트림 구독 - 해제를 잊기 쉬움
    subscription = someStream.listen((data) {
      setState(() {
        // 상태 업데이트
      });
    });
  }
  
  // dispose를 잊으면 메모리 누수!
  @override
  void dispose() {
    subscription.cancel(); // 이걸 잊기 쉬움
    super.dispose();
  }
}

// ✅ 이렇게 하세요 - StreamProvider 사용
final dataStreamProvider = StreamProvider<String>((ref) {
  return someStream; // StreamProvider가 자동으로 구독 해제
});

class GoodWidget extends ConsumerWidget {
  @override
  Widget build(BuildContext context, WidgetRef ref) {
    final dataAsync = ref.watch(dataStreamProvider);
    
    // 위젯이 사라지면 자동으로 구독 해제됨
    return dataAsync.when(
      loading: () => CircularProgressIndicator(),
      error: (error, stack) => Text('에러: $error'),
      data: (data) => Text(data),
    );
  }
}

StreamProvider를 사용하면 위젯이 사라질 때 자동으로 Stream 구독이 해제되어 메모리 누수를 방지할 수 있어요.

너무 많은 이벤트 방지

Dart

// ❌ 이렇게 하지 마세요 - 너무 자주 업데이트
final tooFrequentProvider = StreamProvider<DateTime>((ref) {
  // 매 10밀리초마다 업데이트 - 과도함
  return Stream.periodic(Duration(milliseconds: 10), (_) => DateTime.now());
});

// ✅ 이렇게 하세요 - 적절한 간격과 필터링
final optimizedTimeProvider = StreamProvider<DateTime>((ref) {
  return Stream.periodic(Duration(seconds: 1), (_) => DateTime.now())
      .distinct((prev, curr) => prev.second == curr.second); // 초가 바뀔 때만
});

// 또는 debounce 사용
final debouncedSearchProvider = StreamProvider.family<List<String>, String>((ref, query) {
  return Stream.value(query)
      .debounceTime(Duration(milliseconds: 300)) // 300ms 후에 검색
      .distinct() // 같은 쿼리 제거
      .asyncMap((q) => searchAPI(q));
});

너무 자주 업데이트되면 성능 문제가 생길 수 있어요. 적절한 간격을 두거나 distinct, debounce 같은 필터를 사용하세요.

다른 Provider와 비교

vs FutureProvider
• FutureProvider: 한 번 실행되고 끝나는 비동기 작업 (API 호출, 파일 읽기)
• StreamProvider: 지속적으로 데이터가 들어오는 작업 (실시간 채팅, 위치 추적)

vs StateProvider
• StateProvider: 사용자가 직접 변경하는 상태 (버튼 클릭, 입력)
• StreamProvider: 외부에서 자동으로 들어오는 데이터 (센서, 네트워크)

vs StateNotifierProvider
• StateNotifierProvider: 복잡한 상태를 관리하고 사용자가 조작
• StreamProvider: 외부 데이터 소스를 구독하고 자동 업데이트

언제 어떤 것을 쓸까?
• 한 번만 가져오면 됨 → FutureProvider
• 사용자가 직접 바꿈 → StateProvider
• 복잡한 상태 관리 → StateNotifierProvider
• 실시간으로 계속 바뀜 → StreamProvider

정리

StreamProvider는 Riverpod에서 실시간으로 변하는 데이터를 관리하는 핵심 도구입니다. 채팅, 위치, 주식 시세처럼 지속적으로 업데이트되는 데이터를 다룰 때 사용하세요.

핵심 포인트:
• 지속적으로 변하는 데이터의 실시간 관리
• 자동 구독/해제로 메모리 누수 방지
• when() 메소드로 로딩/에러/데이터 상태 처리
• Stream 변환과 필터링으로 데이터 가공

실무에서 자주 사용하는 패턴:
• Firebase 실시간 데이터베이스 연동
• WebSocket을 통한 실시간 통신
• GPS 위치 추적과 센서 데이터
• 타이머와 주기적 업데이트

주의할 점:
• 메모리 누수 방지 (StreamProvider가 자동 처리)
• 적절한 업데이트 간격 설정
• 네트워크 에러에 대한 복구 로직
• 너무 빈번한 업데이트 방지

성능 최적화 팁:
• distinct()로 중복 이벤트 제거
• debounce()로 과도한 요청 방지
• 적절한 필터링과 변환 적용

StreamProvider를 잘 활용하면 실시간 기능이 필요한 앱을 쉽게 만들 수 있습니다. 채팅 앱, 위치 기반 서비스, 라이브 데이터 대시보드 등 다양한 곳에서 활용해보세요.

다음 편 예고: 다음 글에서는 복잡한 상태를 안전하게 관리하는 StateNotifierProvider에 대해 알아보겠습니다. 불변성과 상태 관리의 고급 패턴을 다룰 예정이에요.

Flutter Riverpod 이해하기 4편 - StreamProvider

StreamProvider란?

언제 StreamProvider를 사용하는가?

기본 사용법

1. StreamProvider 만들기

2. StreamProvider 사용하기

3. 매개변수가 있는 StreamProvider

실무 예제

실시간 채팅 시스템

위치 추적 앱

실시간 주식 시세

StreamProvider의 특별한 기능들

Stream 변환과 필터링

에러 처리와 재연결

주의사항

메모리 누수 방지

너무 많은 이벤트 방지

다른 Provider와 비교

정리

관련 글