opensource/localsend
1
0
Fork 0
mirror of https://github.com/localsend/localsend.git synced 2026-08-07 07:14:52 +00:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

feat: iterate through file list within isolate

Browse Source
This commit is contained in:
Tien Do Nam
2026-07-16 04:05:50 +02:00
parent ba7ee985ad
commit 911c1956e1
6 changed files with 231 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/lib/isolate/isolate.dart
+9 -1
View File
@@ -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';
app/lib/isolate/src/isolate/child/upload_isolate.dart
+100 -78
View File
@@ -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<int>? 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) => <int, RsCancellationToken>{});
@@ -95,7 +113,7 @@ abstract class UriContentStreamResolver {
Future<void> setupHttpUploadIsolate(
Stream<SendToIsolateData<IsolateTask<BaseHttpUploadTask>>> receiveFromMain,
void Function(IsolateTaskStreamResult<double>) sendToMain,
void Function(IsolateTaskStreamResult<HttpUploadEvent>) sendToMain,
InitialData initialData,
) async {
await setupChildIsolateHelper(
@@ -104,9 +122,7 @@ Future<void> setupHttpUploadIsolate(
sendToMain: sendToMain,
initialData: initialData,
handler: (ref, task) async {
final String? remoteSessionId;
final List<HttpUploadFile> files;
final Device device;
final HttpUploadFilesTask uploadTask;
switch (task.data) {
case HttpUploadSetContentStreamResolverTask task:
final rootIsolateToken = ref.read(syncProvider).rootIsolateToken;
@@ -114,23 +130,8 @@ Future<void> 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<void> 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<int>(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<void> setupHttpUploadIsolate(
id: task.id,
),
);
} catch (e) {
sendToMain(
IsolateTaskStreamResult.error(
id: task.id,
error: e.humanErrorMessage,
),
);
} finally {
ref.read(_cancelTokenProvider).remove(task.id);
}
app/lib/isolate/src/isolate/parent/actions.dart
+4 -56
View File
@@ -113,66 +113,14 @@ class IsolateSendMulticastRestartListenerAction extends ReduxAction<IsolateContr
class IsolateHttpUploadActionResult {
final int taskId;
final Stream<double> progress;
final Stream<HttpUploadEvent> events;
IsolateHttpUploadActionResult({
required this.taskId,
required this.progress,
required this.events,
});
}
class IsolateHttpUploadAction extends ReduxActionWithResult<IsolateController, ParentIsolateState, IsolateHttpUploadActionResult> {
final String? remoteSessionId;
final String remoteFileToken;
final String fileId;
final String? filePath;
final List<int>? 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<IsolateController, ParentIsolateState, IsolateHttpUploadActionResult> {
final String? remoteSessionId;
final List<HttpUploadFile> files;
@@ -191,7 +139,7 @@ class IsolateHttpUploadFilesAction extends ReduxActionWithResult<IsolateControll
throw StateError('httpUpload is not initialized');
}
final taskId = IdProvider.instance.getNextId();
final progress = connection.sendWrappedTaskAndListenStream(
final events = connection.sendWrappedTaskAndListenStream(
task: HttpUploadFilesTask(
remoteSessionId: remoteSessionId,
files: files,
@@ -204,7 +152,7 @@ class IsolateHttpUploadFilesAction extends ReduxActionWithResult<IsolateControll
state,
IsolateHttpUploadActionResult(
taskId: taskId,
progress: progress,
events: events,
),
);
}
app/lib/isolate/src/isolate/parent/parent_isolate_provider.dart
+2 -2
View File
@@ -20,7 +20,7 @@ class ParentIsolateState with ParentIsolateStateMappable {
final SyncState syncState;
final IsolateConnector<IsolateTaskStreamResult<Device>, SendToIsolateData<IsolateTask<HttpScanTask>>>? httpScanDiscovery;
final IsolateConnector<Device, SendToIsolateData<MulticastTask>>? multicastDiscovery;
final IsolateConnector<IsolateTaskStreamResult<double>, SendToIsolateData<IsolateTask<BaseHttpUploadTask>>>? httpUpload;
final IsolateConnector<IsolateTaskStreamResult<HttpUploadEvent>, SendToIsolateData<IsolateTask<BaseHttpUploadTask>>>? httpUpload;
ParentIsolateState({
required this.syncState,
@@ -87,7 +87,7 @@ class IsolateSetupAction extends AsyncReduxAction<IsolateController, ParentIsola
);
final httpUpload =
await TypedIsolates.startIsolate<IsolateTaskStreamResult<double>, SendToIsolateData<IsolateTask<BaseHttpUploadTask>>, InitialData>(
await TypedIsolates.startIsolate<IsolateTaskStreamResult<HttpUploadEvent>, SendToIsolateData<IsolateTask<BaseHttpUploadTask>>, InitialData>(
task: setupHttpUploadIsolate,
param: InitialData(
syncState: state.syncState,
app/lib/isolate/src/isolate/parent/parent_isolate_provider.mapper.dart
+3 -3
View File
@@ -50,14 +50,14 @@ class ParentIsolateStateMapper extends ClassMapperBase<ParentIsolateState> {
>
_f$multicastDiscovery = Field('multicastDiscovery', _$multicastDiscovery);
static IsolateConnector<
IsolateTaskStreamResult<double>,
IsolateTaskStreamResult<HttpUploadEvent>,
SendToIsolateData<IsolateTask<BaseHttpUploadTask>>
>?
_$httpUpload(ParentIsolateState v) => v.httpUpload;
static const Field<
ParentIsolateState,
IsolateConnector<
IsolateTaskStreamResult<double>,
IsolateTaskStreamResult<HttpUploadEvent>,
SendToIsolateData<IsolateTask<BaseHttpUploadTask>>
>
>
@@ -162,7 +162,7 @@ abstract class ParentIsolateStateCopyWith<
IsolateConnector<Device, SendToIsolateData<MulticastTask>>?
multicastDiscovery,
IsolateConnector<
IsolateTaskStreamResult<double>,
IsolateTaskStreamResult<HttpUploadEvent>,
SendToIsolateData<IsolateTask<BaseHttpUploadTask>>
>?
httpUpload,
app/lib/provider/network/send_provider.dart
+113 -74
View File
@@ -306,22 +306,19 @@ class SendNotifier extends Notifier<Map<String, SendSessionState>> {
),
);
await _sendLoop(ref, sessionId, target, sendingFiles);
await _sendLoop(sessionId, sendingFiles);
}
Future<void> _sendLoop(Ref ref, String sessionId, Device target, Map<String, SendingFile> files) async {
Future<void> _sendLoop(String sessionId, Map<String, SendingFile> 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<Map<String, SendSessionState>> {
final uriContent = UriContent();
/// Sends a file.
/// Returns true, if the next file should be sent.
Future<bool> sendFile({
/// Sends a single file. Currently only used to retry a failed file.
Future<void> 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<Map<String, SendSessionState>> {
}),
),
);
} 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<void> _sendFiles({
required String sessionId,
required List<SendingFile> 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<Map<String, SendSessionState>> {
),
);
}
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.