diff --git a/app/lib/isolate/isolate.dart b/app/lib/isolate/isolate.dart index 233d1a5f..e5a49973 100644 --- a/app/lib/isolate/isolate.dart +++ b/app/lib/isolate/isolate.dart @@ -1,5 +1,13 @@ export 'package:localsend_app/isolate/src/isolate/child/sync_provider.dart'; -export 'package:localsend_app/isolate/src/isolate/child/upload_isolate.dart' show UriContentStreamResolver; +export 'package:localsend_app/isolate/src/isolate/child/upload_isolate.dart' + show + HttpUploadEvent, + HttpUploadFile, + HttpUploadFileFailedEvent, + HttpUploadFileFinishedEvent, + HttpUploadFileProgressEvent, + HttpUploadFileStartedEvent, + UriContentStreamResolver; export 'package:localsend_app/isolate/src/isolate/parent/actions.dart'; export 'package:localsend_app/isolate/src/isolate/parent/actions_sync.dart'; export 'package:localsend_app/isolate/src/isolate/parent/parent_isolate_provider.dart'; diff --git a/app/lib/isolate/src/isolate/child/upload_isolate.dart b/app/lib/isolate/src/isolate/child/upload_isolate.dart index d4e6f67f..e6316914 100644 --- a/app/lib/isolate/src/isolate/child/upload_isolate.dart +++ b/app/lib/isolate/src/isolate/child/upload_isolate.dart @@ -21,26 +21,6 @@ class HttpUploadSetContentStreamResolverTask implements BaseHttpUploadTask { }); } -class HttpUploadTask implements BaseHttpUploadTask { - final String? remoteSessionId; - final String remoteFileToken; - final String fileId; - final String? filePath; - final List? fileBytes; - final int fileSize; - final Device device; - - HttpUploadTask({ - required this.remoteSessionId, - required this.remoteFileToken, - required this.fileId, - required this.filePath, - required this.fileBytes, - required this.fileSize, - required this.device, - }); -} - class HttpUploadFile { final String remoteFileToken; final String fileId; @@ -80,6 +60,44 @@ class HttpUploadCancelTask implements BaseHttpUploadTask { HttpUploadCancelTask({required this.taskId}); } +/// A message sent from the upload isolate to the main isolate +/// reporting the state of a single file of a [HttpUploadFilesTask]. +sealed class HttpUploadEvent { + final String fileId; + + HttpUploadEvent({required this.fileId}); +} + +/// The upload of the file has started. +class HttpUploadFileStartedEvent extends HttpUploadEvent { + HttpUploadFileStartedEvent({required super.fileId}); +} + +/// The upload progress of the file in the range [0, 1]. +class HttpUploadFileProgressEvent extends HttpUploadEvent { + final double progress; + + HttpUploadFileProgressEvent({ + required super.fileId, + required this.progress, + }); +} + +/// The file has been uploaded successfully. +class HttpUploadFileFinishedEvent extends HttpUploadEvent { + HttpUploadFileFinishedEvent({required super.fileId}); +} + +/// The upload of the file has failed. The next file is still uploaded. +class HttpUploadFileFailedEvent extends HttpUploadEvent { + final String error; + + HttpUploadFileFailedEvent({ + required super.fileId, + required this.error, + }); +} + /// Map of cancel tokens for each task. /// Task ID -> CancelToken final _cancelTokenProvider = Provider((ref) => {}); @@ -95,7 +113,7 @@ abstract class UriContentStreamResolver { Future setupHttpUploadIsolate( Stream>> receiveFromMain, - void Function(IsolateTaskStreamResult) sendToMain, + void Function(IsolateTaskStreamResult) sendToMain, InitialData initialData, ) async { await setupChildIsolateHelper( @@ -104,9 +122,7 @@ Future setupHttpUploadIsolate( sendToMain: sendToMain, initialData: initialData, handler: (ref, task) async { - final String? remoteSessionId; - final List files; - final Device device; + final HttpUploadFilesTask uploadTask; switch (task.data) { case HttpUploadSetContentStreamResolverTask task: final rootIsolateToken = ref.read(syncProvider).rootIsolateToken; @@ -114,23 +130,8 @@ Future setupHttpUploadIsolate( rootIsolateToken: rootIsolateToken, ); return; - case HttpUploadTask task: - remoteSessionId = task.remoteSessionId; - files = [ - HttpUploadFile( - remoteFileToken: task.remoteFileToken, - fileId: task.fileId, - filePath: task.filePath, - fileBytes: task.fileBytes, - fileSize: task.fileSize, - ), - ]; - device = task.device; - break; case HttpUploadFilesTask task: - remoteSessionId = task.remoteSessionId; - files = task.files; - device = task.device; + uploadTask = task; break; case HttpUploadCancelTask task: final cancelToken = ref.read(_cancelTokenProvider)[task.taskId]; @@ -139,41 +140,69 @@ Future setupHttpUploadIsolate( return; } + final cancelToken = createCancellationToken(); + ref.read(_cancelTokenProvider).putIfAbsent(task.id, () => cancelToken); try { - final cancelToken = createCancellationToken(); - ref.read(_cancelTokenProvider).putIfAbsent(task.id, () => cancelToken); - final totalSize = files.fold(0, (sum, file) => sum + file.fileSize); - var completedSize = 0; + for (final file in uploadTask.files) { + sendToMain( + IsolateTaskStreamResult.event( + id: task.id, + data: HttpUploadFileStartedEvent(fileId: file.fileId), + ), + ); - for (final file in files) { - final filePath = file.filePath; - final isContentUri = filePath?.startsWith('content://') ?? false; - final fileDescriptor = isContentUri ? await getFileDescriptorAndroid(uri: filePath!) : null; + try { + final filePath = file.filePath; + final isContentUri = filePath?.startsWith('content://') ?? false; + final fileDescriptor = isContentUri ? await getFileDescriptorAndroid(uri: filePath!) : null; - await ref - .read(httpUploadProvider) - .upload( - stream: filePath == null && file.fileBytes != null ? Stream.value(file.fileBytes!) : null, - path: !isContentUri ? filePath : null, - fileDescriptor: fileDescriptor, - contentLength: file.fileSize, - target: device, - remoteSessionId: remoteSessionId, - fileId: file.fileId, - token: file.remoteFileToken, - onSendProgress: (progress) { - final totalProgress = totalSize == 0 ? 1.0 : (completedSize + progress * file.fileSize) / totalSize; - sendToMain( - IsolateTaskStreamResult.event( - id: task.id, - data: totalProgress, - ), - ); - }, - cancelToken: cancelToken, - ); + await ref + .read(httpUploadProvider) + .upload( + stream: filePath == null && file.fileBytes != null ? Stream.value(file.fileBytes!) : null, + path: !isContentUri ? filePath : null, + fileDescriptor: fileDescriptor, + contentLength: file.fileSize, + target: uploadTask.device, + remoteSessionId: uploadTask.remoteSessionId, + fileId: file.fileId, + token: file.remoteFileToken, + onSendProgress: (progress) { + sendToMain( + IsolateTaskStreamResult.event( + id: task.id, + data: HttpUploadFileProgressEvent( + fileId: file.fileId, + progress: progress, + ), + ), + ); + }, + cancelToken: cancelToken, + ); - completedSize += file.fileSize; + sendToMain( + IsolateTaskStreamResult.event( + id: task.id, + data: HttpUploadFileFinishedEvent(fileId: file.fileId), + ), + ); + } catch (e) { + sendToMain( + IsolateTaskStreamResult.event( + id: task.id, + data: HttpUploadFileFailedEvent( + fileId: file.fileId, + error: e.humanErrorMessage, + ), + ), + ); + } + + if (!ref.read(_cancelTokenProvider).containsKey(task.id)) { + // the task was canceled, do not upload the remaining files + break; + } } sendToMain( @@ -181,13 +210,6 @@ Future setupHttpUploadIsolate( id: task.id, ), ); - } catch (e) { - sendToMain( - IsolateTaskStreamResult.error( - id: task.id, - error: e.humanErrorMessage, - ), - ); } finally { ref.read(_cancelTokenProvider).remove(task.id); } diff --git a/app/lib/isolate/src/isolate/parent/actions.dart b/app/lib/isolate/src/isolate/parent/actions.dart index b01323a7..e559ea52 100644 --- a/app/lib/isolate/src/isolate/parent/actions.dart +++ b/app/lib/isolate/src/isolate/parent/actions.dart @@ -113,66 +113,14 @@ class IsolateSendMulticastRestartListenerAction extends ReduxAction progress; + final Stream events; IsolateHttpUploadActionResult({ required this.taskId, - required this.progress, + required this.events, }); } -class IsolateHttpUploadAction extends ReduxActionWithResult { - final String? remoteSessionId; - final String remoteFileToken; - final String fileId; - final String? filePath; - final List? fileBytes; - final int fileSize; - final Device device; - - IsolateHttpUploadAction({ - required this.remoteSessionId, - required this.remoteFileToken, - required this.fileId, - required this.filePath, - required this.fileBytes, - required this.fileSize, - required this.device, - }); - - @override - (ParentIsolateState, IsolateHttpUploadActionResult) reduce() { - final connection = state.httpUpload; - if (connection == null) { - throw StateError('httpUpload is not initialized'); - } - - final task = HttpUploadTask( - remoteSessionId: remoteSessionId, - remoteFileToken: remoteFileToken, - fileId: fileId, - filePath: filePath, - fileBytes: fileBytes, - fileSize: fileSize, - device: device, - ); - - final taskId = IdProvider.instance.getNextId(); - final progress = connection.sendWrappedTaskAndListenStream( - task: task, - taskId: taskId, - ); - - return ( - state, - IsolateHttpUploadActionResult( - taskId: taskId, - progress: progress, - ), - ); - } -} - class IsolateHttpUploadFilesAction extends ReduxActionWithResult { final String? remoteSessionId; final List files; @@ -191,7 +139,7 @@ class IsolateHttpUploadFilesAction extends ReduxActionWithResult, SendToIsolateData>>? httpScanDiscovery; final IsolateConnector>? multicastDiscovery; - final IsolateConnector, SendToIsolateData>>? httpUpload; + final IsolateConnector, SendToIsolateData>>? httpUpload; ParentIsolateState({ required this.syncState, @@ -87,7 +87,7 @@ class IsolateSetupAction extends AsyncReduxAction, SendToIsolateData>, InitialData>( + await TypedIsolates.startIsolate, SendToIsolateData>, InitialData>( task: setupHttpUploadIsolate, param: InitialData( syncState: state.syncState, diff --git a/app/lib/isolate/src/isolate/parent/parent_isolate_provider.mapper.dart b/app/lib/isolate/src/isolate/parent/parent_isolate_provider.mapper.dart index 94b3dc11..f00c95a5 100644 --- a/app/lib/isolate/src/isolate/parent/parent_isolate_provider.mapper.dart +++ b/app/lib/isolate/src/isolate/parent/parent_isolate_provider.mapper.dart @@ -50,14 +50,14 @@ class ParentIsolateStateMapper extends ClassMapperBase { > _f$multicastDiscovery = Field('multicastDiscovery', _$multicastDiscovery); static IsolateConnector< - IsolateTaskStreamResult, + IsolateTaskStreamResult, SendToIsolateData> >? _$httpUpload(ParentIsolateState v) => v.httpUpload; static const Field< ParentIsolateState, IsolateConnector< - IsolateTaskStreamResult, + IsolateTaskStreamResult, SendToIsolateData> > > @@ -162,7 +162,7 @@ abstract class ParentIsolateStateCopyWith< IsolateConnector>? multicastDiscovery, IsolateConnector< - IsolateTaskStreamResult, + IsolateTaskStreamResult, SendToIsolateData> >? httpUpload, diff --git a/app/lib/provider/network/send_provider.dart b/app/lib/provider/network/send_provider.dart index e7a9d72c..7fc37cb5 100644 --- a/app/lib/provider/network/send_provider.dart +++ b/app/lib/provider/network/send_provider.dart @@ -306,22 +306,19 @@ class SendNotifier extends Notifier> { ), ); - await _sendLoop(ref, sessionId, target, sendingFiles); + await _sendLoop(sessionId, sendingFiles); } - Future _sendLoop(Ref ref, String sessionId, Device target, Map files) async { + Future _sendLoop(String sessionId, Map files) async { state = state.updateSession( sessionId: sessionId, state: (s) => s?.copyWith(startTime: DateTime.now().millisecondsSinceEpoch), ); - for (final file in files.values) { - await sendFile( - sessionId: sessionId, - file: file, - isRetry: false, - ); - } + await _sendFiles( + sessionId: sessionId, + files: files.values.toList(), + ); _finish(sessionId: sessionId); } @@ -361,27 +358,22 @@ class SendNotifier extends Notifier> { final uriContent = UriContent(); - /// Sends a file. - /// Returns true, if the next file should be sent. - Future sendFile({ + /// Sends a single file. Currently only used to retry a failed file. + Future sendFile({ required String sessionId, required SendingFile file, required bool isRetry, }) async { - final token = file.token; - if (token == null) { - return true; + if (file.token == null) { + return; } final status = state[sessionId]?.status; const allowedStates = {SessionStatus.sending, SessionStatus.finishedWithErrors}; if (status == null || !allowedStates.contains(status)) { - return false; + return; } - final remoteSessionId = state[sessionId]!.remoteSessionId; - final target = state[sessionId]!.target; - if (isRetry) { _logger.info('Retrying ${file.file.fileName}'); @@ -397,64 +389,126 @@ class SendNotifier extends Notifier> { }), ), ); - } else { - _logger.info('Sending ${file.file.fileName}'); } - state = state.updateSession( + await _sendFiles( sessionId: sessionId, - state: (s) => s?.withFileStatus(file.file.id, FileStatus.sending, null), + files: [file], ); - final taskResult = ref - .redux(parentIsolateProvider) - .dispatchTakeResult( - IsolateHttpUploadAction( - remoteSessionId: remoteSessionId, - remoteFileToken: token, + if (isRetry) { + final state = this.state[sessionId]; + if (state != null && state.files.values.map((e) => e.status).isFinishedOrError) { + _finish(sessionId: sessionId); + } + } + } + + /// Sends the given [files] as one isolate task. + /// The isolate iterates through the list and reports the state of each file + /// via [HttpUploadEvent]s. + /// Files without a token (i.e. not selected by the receiver) are skipped. + Future _sendFiles({ + required String sessionId, + required List files, + }) async { + final sessionState = state[sessionId]; + if (sessionState == null) { + return; + } + + final uploadFiles = [ + for (final file in files) + if (file.token != null) + HttpUploadFile( + remoteFileToken: file.token!, fileId: file.file.id, filePath: file.path, fileBytes: file.bytes, fileSize: file.file.size, - device: target, + ), + ]; + + if (uploadFiles.isEmpty) { + return; + } + + final taskResult = ref + .redux(parentIsolateProvider) + .dispatchTakeResult( + IsolateHttpUploadFilesAction( + remoteSessionId: sessionState.remoteSessionId, + files: uploadFiles, + device: sessionState.target, ), ); - String? fileError; + state = state.updateSession( + sessionId: sessionId, + state: (s) => s?.copyWith( + sendingTasks: [ + ...?s.sendingTasks, + SendingTask( + taskId: taskResult.taskId, + ), + ], + ), + ); + try { + await for (final event in taskResult.events) { + switch (event) { + case HttpUploadFileStartedEvent(): + _logger.info('Sending ${state[sessionId]?.files[event.fileId]?.file.fileName}'); + state = state.updateSession( + sessionId: sessionId, + state: (s) => s?.withFileStatus(event.fileId, FileStatus.sending, null), + ); + case HttpUploadFileProgressEvent(): + ref + .notifier(progressProvider) + .setProgress( + sessionId: sessionId, + fileId: event.fileId, + progress: event.progress, + ); + case HttpUploadFileFinishedEvent(): + // set progress to 100% when successfully finished + ref + .notifier(progressProvider) + .setProgress( + sessionId: sessionId, + fileId: event.fileId, + progress: 1, + ); + state = state.updateSession( + sessionId: sessionId, + state: (s) => s?.withFileStatus(event.fileId, FileStatus.finished, null), + ); + case HttpUploadFileFailedEvent(): + _logger.warning('Error while sending file ${state[sessionId]?.files[event.fileId]?.file.fileName}: ${event.error}'); + state = state.updateSession( + sessionId: sessionId, + state: (s) => s?.withFileStatus(event.fileId, FileStatus.failed, event.error), + ); + } + } + } catch (e, st) { + // the whole task failed, mark all files of this task that did not finish as failed + _logger.warning('Error while sending files', e, st); + final error = e.humanErrorMessage; + final fileIds = uploadFiles.map((file) => file.fileId).toSet(); state = state.updateSession( sessionId: sessionId, state: (s) => s?.copyWith( - sendingTasks: [ - ...?s.sendingTasks, - SendingTask( - taskId: taskResult.taskId, - ), - ], + files: s.files.map((key, value) { + if (fileIds.contains(key) && (value.status == FileStatus.queue || value.status == FileStatus.sending)) { + return MapEntry(key, value.copyWith(status: FileStatus.failed, errorMessage: error)); + } + return MapEntry(key, value); + }), ), ); - - await for (final progress in taskResult.progress) { - ref - .notifier(progressProvider) - .setProgress( - sessionId: sessionId, - fileId: file.file.id, - progress: progress, - ); - } - - // set progress to 100% when successfully finished - ref - .notifier(progressProvider) - .setProgress( - sessionId: sessionId, - fileId: file.file.id, - progress: 1, - ); - } catch (e, st) { - fileError = e.humanErrorMessage; - _logger.warning('Error while sending file ${file.file.fileName}', e, st); } finally { state = state.updateSession( sessionId: sessionId, @@ -463,21 +517,6 @@ class SendNotifier extends Notifier> { ), ); } - - state = state.updateSession( - sessionId: sessionId, - state: (s) => s?.withFileStatus(file.file.id, fileError != null ? FileStatus.failed : FileStatus.finished, fileError), - ); - - if (isRetry) { - final state = this.state[sessionId]; - if (state != null && state.files.values.map((e) => e.status).isFinishedOrError) { - _finish(sessionId: sessionId); - return false; - } - } - - return true; } /// Closes the send-session and sends a cancel event to the receiver.