Fix chunk upload with ChunkedStreamReader to cancel underlying stream

Author: hoangdat

data/lib/src/repository/flow/flow_uploader_impl.dart

@@ -35,6 +35,7 @@ import 'dart:async';
 import 'dart:developer' as developer;
 import 'dart:io';
 
+import 'package:async/async.dart';
 import 'package:dartz/dartz.dart';
 import 'package:data/src/network/config/endpoint.dart';
 import 'package:data/src/network/dio_client.dart';
@@ -43,7 +44,6 @@ import 'package:data/src/network/model/query/query_parameter.dart';
 import 'package:data/src/repository/flow/flow_response.dart';
 import 'package:dio/dio.dart';
 import 'package:domain/domain.dart';
-import 'package:domain/src/model/async_task/async_task.dart';
 
 class FlowUploaderImpl extends FlowUploader {
   final DioClient _dioClient;
@@ -102,7 +102,7 @@ class FlowUploaderImpl extends FlowUploader {
 
   @override
   Future<Flow> uploadChunk(
-      File file,
+      ChunkedStreamReader<int> chunkedStreamReaderFile,
       int chunkNumber,
       int chunkSize,
       int currentChunkSize,
@@ -116,30 +116,51 @@ class FlowUploaderImpl extends FlowUploader {
       {String? sharedSpaceId,
       String? parentNodeId}
   ) async {
-    final _fileSize = file.lengthSync();
-    developer.log('uploadChunk(): chunk: $chunkNumber - currentChunkSize: $currentChunkSize', name: 'FlowUploaderImpl');
-
-    final formData = generateFormData(file, chunkNumber, chunkSize, currentChunkSize,
-        startByte, endByte, _fileSize, flowIdentifier, flowFile, totalChunk, uploadedByte,
-        onSendController, sharedSpaceId: sharedSpaceId, parentNodeId: parentNodeId);
-
-    final response = await _dioClient.post(
-      Endpoint.flow.generateEndpointPath(),
-      data: formData,
-      options: Options(
-        headers: _getRangeHeadersForChunkUpload(startByte, endByte, _fileSize)
-      ),
-      onSendProgress: (progress, total) {
-        onSendController.add(Right(UploadingFlowUploadState(flowFile, uploadedByte + progress, flowFile.fileInfo.fileSize)));
-      }
-    );
-    final flowResponse = FlowResponse.fromJson(response);
-    developer.log('uploadChunk(): ${flowResponse.toString()}', name: 'FlowUploaderImpl');
-    return flowResponse.toFlow();
+    try {
+      final _fileSize = flowFile.fileInfo.fileSize;
+      developer.log(
+        'uploadChunk(): chunk: $chunkNumber - currentChunkSize: $currentChunkSize',
+        name: 'FlowUploaderImpl');
+
+      final formData = generateFormData(
+        chunkedStreamReaderFile,
+        chunkNumber,
+        chunkSize,
+        currentChunkSize,
+        startByte,
+        endByte,
+        _fileSize,
+        flowIdentifier,
+        flowFile,
+        totalChunk,
+        uploadedByte,
+        onSendController,
+        sharedSpaceId: sharedSpaceId,
+        parentNodeId: parentNodeId);
+
+      final response = await _dioClient.post(
+        Endpoint.flow.generateEndpointPath(),
+        data: formData,
+        options: Options(
+          headers: _getRangeHeadersForChunkUpload(startByte, endByte, _fileSize)
+        ),
+        onSendProgress: (progress, total) {
+          onSendController.add(Right(UploadingFlowUploadState(
+            flowFile, uploadedByte + progress,
+            flowFile.fileInfo.fileSize)));
+        }
+      );
+      final flowResponse = FlowResponse.fromJson(response);
+      developer.log('uploadChunk(): ${flowResponse.toString()}', name: 'FlowUploaderImpl');
+      return flowResponse.toFlow();
+    } catch (e) {
+      developer.log('uploadChunk(): exception vkl ${e}', name: 'FlowUploaderImpl');
+      rethrow;
+    }
   }
 
   FormData generateFormData(
-    File file,
+    ChunkedStreamReader<int> chunkedStreamReaderFile,
     int chunkNumber,
     int chunkSize,
     int currentChunkSize,
@@ -163,7 +184,7 @@ class FlowUploaderImpl extends FlowUploader {
       IDENTIFIER: flowIdentifier,
       FILENAME: flowFile.fileInfo.fileName,
       RELATIVE_PATH: flowFile.fileInfo.fileName,
-      FILE: MultipartFile(file.openRead(startByte, endByte), endByte - startByte),
+      FILE: MultipartFile(chunkedStreamReaderFile.readStream(currentChunkSize), currentChunkSize),
       ASYNC_TASK: true,
     };
 

domain/lib/src/model/flow/flow_chunk.dart

@@ -32,9 +32,11 @@
  */
 
 import 'dart:async';
+import 'dart:developer' as developer;
 import 'dart:io';
 import 'dart:math';
 
+import 'package:async/async.dart';
 import 'package:dartz/dartz.dart';
 import 'package:domain/domain.dart';
 import 'package:domain/src/model/flow/flow_chunk_upload_state.dart';
@@ -58,6 +60,7 @@ class FlowChunk extends Equatable {
   FlowChunkUploadState status = FlowChunkUploadState.pending;
 
   bool error = false;
+  ChunkedStreamReader<int>? _chunkedStreamReader;
 
   final FlowUploader _flowUploader;
 
@@ -75,8 +78,9 @@ class FlowChunk extends Equatable {
 
   Future<Flow> upload(int uploadedByte, StreamController<Either<Failure, Success>> onSendController) async {
     status = FlowChunkUploadState.uploading;
+    _chunkedStreamReader = ChunkedStreamReader(file.openRead(startByte, endByte));
     return _flowUploader.uploadChunk(
-      file,
+      _chunkedStreamReader!,
       offset + 1,
       chunkSize,
       currentChunkSize,
@@ -108,11 +112,16 @@ class FlowChunk extends Equatable {
     if (isSuccess) {
       status = FlowChunkUploadState.success;
     } else {
+      developer.log('test(): update status error for offset $offset', name: 'FlowChunk');
       status = FlowChunkUploadState.error;
     }
     return isSuccess;
   }
 
+  void completed() {
+    _chunkedStreamReader?.cancel();
+  }
+
   @override
   List<Object?> get props => [offset, flowFile, identifier];
 }

domain/lib/src/model/flow/flow_file.dart

@@ -35,15 +35,12 @@ import 'dart:async';
 import 'dart:developer' as developer;
 import 'dart:io';
 
+import 'package:async/async.dart';
 import 'package:dartz/dartz.dart';
 import 'package:domain/domain.dart';
-import 'package:domain/src/model/async_task/async_task.dart';
 import 'package:domain/src/model/async_task/async_task_exception.dart';
 import 'package:domain/src/model/async_task/async_task_status.dart';
-import 'package:domain/src/model/flow/flow_chunk.dart';
 import 'package:domain/src/model/flow/flow_chunk_upload_state.dart';
-import 'package:domain/src/repository/flow/flow_uploader.dart';
-import 'package:domain/src/usecases/upload_file/flow_upload_state.dart';
 import 'package:equatable/equatable.dart';
 import 'package:retry/retry.dart';
 
@@ -110,21 +107,26 @@ class FlowFile extends Equatable {
   }
 
   void _updateEvent(Either<Failure, Success> flowUploadState) {
+    developer.log('_updateEvent() for $this = $flowUploadState', name: 'FlowFile');
     _progressStateController.add(flowUploadState);
   }
 
   void _handleCompleted() {
-    if (chunks.every((chunk) => chunk.status == FlowChunkUploadState.success)) {
-      final asyncTaskId = _getProcessingTaskId();
-      if (asyncTaskId != null) {
-        _handleProcessing(asyncTaskId);
+    try {
+      if (chunks.every((chunk) => chunk.status == FlowChunkUploadState.success)) {
+        final asyncTaskId = _getProcessingTaskId();
+        if (asyncTaskId != null) {
+          _handleProcessing(asyncTaskId);
+        } else {
+          _handleSuccess();
+        }
       } else {
-        _handleSuccess();
+        developer.log('handleCompleted(): error', name: 'FlowFile');
+        _updateEvent(Left(ErrorFlowUploadState(this)));
+        _progressStateController.close();
       }
-    } else {
-      developer.log('handleCompleted(): error', name: 'FlowFile');
-      _updateEvent(Left(ErrorFlowUploadState(this)));
-      _progressStateController.close();
+    } catch (e) {
+      developer.log('_handleCompleted(): exception $e', name: 'FlowFile');
     }
   }
 
@@ -176,6 +178,7 @@ class FlowFile extends Equatable {
       final success = await chunk.test();
       if (success) {
         _updateProgress(chunk);
+        chunk.completed();
       }
     }
   }

domain/lib/src/repository/flow/flow_uploader.dart

@@ -32,16 +32,19 @@
  */
 
 import 'dart:async';
-import 'dart:io';
 
+import 'package:async/async.dart';
 import 'package:dartz/dartz.dart';
-import 'package:domain/domain.dart';
 import 'package:domain/src/model/async_task/async_task.dart';
+import 'package:domain/src/model/file_info.dart';
 import 'package:domain/src/model/flow/flow.dart';
+import 'package:domain/src/model/flow/flow_file.dart';
+import 'package:domain/src/state/failure.dart';
+import 'package:domain/src/state/success.dart';
 
 abstract class FlowUploader {
   Future<Flow> uploadChunk(
-    File file,
+    ChunkedStreamReader<int> chunkedStreamReaderFile,
     int chunkNumber,
     int chunkSize,
     int currentChunkSize,

lib/presentation/manager/upload_and_share_file/upload_and_share_file_manager.dart

@@ -144,6 +144,7 @@ class UploadShareFileManager {
     _progressStateStreamGroup.stream.listen((flowUploadState) {
       flowUploadState.fold(
         (failure) {
+          developer.log('_handleFlowProgressState(): [Failure] $failure', name: 'UploadShareFileManager');
           if (failure is ErrorFlowUploadState) {
             _uploadingStateFiles.updateElementByUploadTaskId(
               failure.flowFile.uploadTaskId,
@@ -154,8 +155,6 @@ class UploadShareFileManager {
         },
         (success) {
           if (success is UploadingFlowUploadState) {
-            developer.log('_handleFlowProgressState(): uploading: ${success.progress}', name: 'UploadShareFileManager');
-
             _uploadingStateFiles.updateElementByUploadTaskId(
               success.flowFile.uploadTaskId,
               (currentState) => (currentState?.uploadStatus.completed ?? false)
@@ -177,7 +176,7 @@ class UploadShareFileManager {
   }
 
   void _handleUploadFileSucceedWithResource(SuccessWithResourceFlowUploadState successWithResourceFlowUploadState) {
-    developer.log('_handleUploadFileSucceed()', name: 'UploadShareFileManager');
+    developer.log('_handleUploadFileSucceedWithResource()', name: 'UploadShareFileManager');
     final fileState = _uploadingStateFiles.getElementByUploadTaskId(successWithResourceFlowUploadState.flowFile.uploadTaskId);
     if (fileState != null) {
       _fileHelper.deleteFile(fileState.file);
@@ -227,6 +226,7 @@ class UploadShareFileManager {
   }
 
   void _handleFlowUploadFileFailure(ErrorFlowUploadState errorFlowUploadState) {
+    developer.log('_handleFlowUploadFileFailure(): $errorFlowUploadState', name: 'UploadShareFileManager');
     final fileState = _uploadingStateFiles.getElementByUploadTaskId(errorFlowUploadState.flowFile.uploadTaskId);
     if (fileState != null) {
       _fileHelper.deleteFile(fileState.file);

lib/presentation/util/helper/file_helper.dart

@@ -30,19 +30,19 @@
 //  the Additional Terms applicable to LinShare software.
 //
 
+import 'dart:developer' as developer;
 import 'dart:io';
 
 import 'package:domain/domain.dart';
 
 class FileHelper {
   void deleteFile(FileInfo fileInfo) async {
-    if (fileInfo != null) {
-      final file = File(fileInfo.filePath + fileInfo.fileName);
-      try {
-        await file.delete();
-      } catch (exception) {
-        print('error when delete file: ${fileInfo.fileName} ' + exception.toString());
-      }
+    developer.log('deleteFile(): $fileInfo', name: 'FileHelper');
+    final file = File(fileInfo.filePath + fileInfo.fileName);
+    try {
+      await file.delete();
+    } catch (exception) {
+      developer.log('deleteFile(): error when delete file: ${fileInfo.fileName}' + exception.toString(), name: 'FileHelper');
     }
   }
 }