RxDart - Handling Multiple Responses for the Same Event in a Single Bloc

요구 상황 :

• 3개의 문서를 업로드하는 동일한 이벤트를 실행합니다.

• 모든 업로드가 성공하면 승인(Approval) 이벤트를 실행합니다.

문제 접근:

• 각기 다른 Bloc에서 각자의 Stream을 사용했다면, combineLatest로 쉽게 처리할 수 있습니다.

• 그러나, 현재 상황에서는 하나의 Bloc(documentBloc)에서 동일한 이벤트(UploadDocumen)를 여러 번 처리해야 합니다.

• 따라서 각 업로드의 성공/실패 상태를 수집하고, 모든 업로드가 끝났을 때만 후속 처리를 실행해야 합니다.

해결 방법:

• RxDart의 **forkJoin**을 활용해 각 업로드의 결과를 수집합니다.

• 모든 업로드가 성공했는지 확인한 후, 승인 이벤트를 실행합니다.

StreamSubscription<List<DocumentState>>? _uploadSubscription;

late final documentBloc = context.read<DocumentBloc>();

void _submitDocuments() {
  // 1. 업로드가 필요한 문서를 가져옴
  final uploadingDocuments = requiredDocumentsCubit.state.documents.uploadingTypeDocuments();

  // 업로드할 문서가 없으면 종료
  if (uploadingDocuments.isEmpty) return;

  // 2. 각 문서에 대한 업로드 이벤트 전송 및 상태 스트림 수집
  final uploadStateStreams = uploadingDocuments.map((document) {
    // 각 문서별 업로드 이벤트 전송 (documentBloc에 이벤트 추가)
    documentBloc.add(
      UpdateDocument(
        documentType: document.documentType,
        photoData: document.photoData,
      ),
    );

    // 각 문서의 업로드 결과(성공/실패) 중 첫 번째 방출 값을 Stream으로 변환, 이를 리스트로 수집
    return documentBloc.stream
        .where((state) => state is UploadDocumentSuccess || state is UploadDocumentFailure)
        .first
        .asStream();
  }).toList();

  // 3. 모든 업로드가 끝난 후 로직 수행
  _uploadSubscription = Rx.forkJoin(uploadStateStreams, (List<DocumentState> states) => states).listen(
    (states) {
      if (!mounted) return;

      // 모든 업로드가 성공했는지 확인
      final isAllUploadSuccess = states.every((state) => state is UploadDocumentSuccess);

      // 최종 상태를 가져옴 (실패 시 실패 상태, 성공 시 첫 성공 상태)
      final finalState = isAllUploadSuccess
          ? states.firstWhere((state) => state is UploadDocumentSuccess)
          : states.firstWhere((state) => state is UploadDocumentFailure);

      // UI 업데이트 및 후속 처리
      _onDocumentBlocListener(context: context, state: finalState);

      // 모든 업로드 성공 시 승인 이벤트 전송
      if (isAllUploadSuccess) {
        approvalRequiredDocumentsBloc.add(const DoApprovalModifyRequiredDocuments());
      }
    },
    onDone: () => _uploadSubscription?.cancel(),
    onError: (error) {},
  );
}

firstWhere 대신 where().first 를 사용한 이유:

firstWhere()는 전체 Stream을 순회해야 하므로 비효율적입니다.
where().first는 첫 번째로 만족하는 값이 나오면 바로 반환하므로 성능상 더 유리합니다.

return documentBloc.stream
        .where((state) => state is UploadDocumentSuccess || state is UploadDocumentFailure)
        .first
        .asStream();
  }).toList();

forkJoin과 combineLatest의 차이 (https://rxmarbles.com)

• forkJoin

  • 모든 스트림이 완료된 후, 각 스트림의 마지막 값을 결합합니다.

  • 모든 이벤트의 결과를 한 번에 처리해야 할 때 적합합니다.

• combineLatest

  • 각 스트림에서 새로운 값이 방출될 때마다, 가장 최신 값을 결합합니다.

  • 이벤트가 발생할 때마다 중간 결과를 실시간으로 처리해야 할 때 유용합니다.

이 예제에서 forkJoin을 선택한 이유

• 3개의 업로드 이벤트가 모두 완료된 후 승인 이벤트를 실행해야 합니다.

• combineLatest는 업로드 도중 불필요한 중간 상태 (UploadDocumentInProcess)도 반응하므로 적합하지 않음.

• 동일한 Bloc에서 여러 번의 비동기 작업을 기다릴 때는 forkJoin이 정확한 시점 제어 가능