diff --git a/packages/localsend_isolates/.gitignore b/packages/localsend_isolates/.gitignore index 68fcca3d..b864c9eb 100644 --- a/packages/localsend_isolates/.gitignore +++ b/packages/localsend_isolates/.gitignore @@ -3,3 +3,4 @@ .flutter-plugins .flutter-plugins-dependencies /build/ +/pubspec.lock diff --git a/packages/localsend_isolates/lib/rust/api/crypto.dart b/packages/localsend_isolates/lib/rust/api/crypto.dart index d5e7ad7d..77f301bf 100644 --- a/packages/localsend_isolates/lib/rust/api/crypto.dart +++ b/packages/localsend_isolates/lib/rust/api/crypto.dart @@ -28,6 +28,11 @@ Future generateSecurityContext() => RustLib.instance.api.crateA /// Cancelling [cancel_token] aborts the read, so hashing a large file does not /// have to be waited out. /// +/// Failures (including cancellation) are emitted as errors on the stream: +/// flutter_rust_bridge discards the returned `Result` of functions taking a +/// [StreamSink], so a returned error would become an uncaught async error +/// killing the calling isolate. +/// /// Exactly one content source must be provided: /// a [path] to a regular file, a [file_descriptor] (Android only), or [bytes] /// for a file that only lives in memory. diff --git a/packages/localsend_isolates/lib/rust/api/http.dart b/packages/localsend_isolates/lib/rust/api/http.dart index 0445f0ba..ef91da4c 100644 --- a/packages/localsend_isolates/lib/rust/api/http.dart +++ b/packages/localsend_isolates/lib/rust/api/http.dart @@ -13,7 +13,7 @@ import 'package:localsend_isolates/rust/frb_generated.dart'; part 'http.freezed.dart'; // These functions are ignored because they are not marked as `pub`: `error_chain`, `resolve_file_content` -// These function are ignored because they are on traits that is not defined in current crate (put an empty `#[frb]` on it to unignore): `from` +// These function are ignored because they are on traits that is not defined in current crate (put an empty `#[frb]` on it to unignore): `clone`, `clone`, `from` /// Creates an HTTP client. /// @@ -56,7 +56,13 @@ abstract class RsHttpClient implements RustOpaqueInterface { required RegisterDto payload, }); - Stream upload({ + /// Uploads a single file, emitting [RsUploadEvent]s on [sink]. + /// + /// Failures are emitted as [RsUploadEvent::Failed] instead of being + /// returned: flutter_rust_bridge discards the returned `Result` of + /// functions taking a [StreamSink], so a returned error would become an + /// uncaught async error killing the calling isolate. + Stream upload({ required ProtocolType protocol, required String ip, required int port, @@ -134,3 +140,18 @@ sealed class RsHttpClientError with _$RsHttpClientError implements FrbException String field0, ) = RsHttpClientError_Other; } + +@freezed +sealed class RsUploadEvent with _$RsUploadEvent { + const RsUploadEvent._(); + + /// The upload progress as a fraction (0.0 to 1.0). Throttled. + const factory RsUploadEvent.progress({ + required double progress, + }) = RsUploadEvent_Progress; + + /// The upload failed. Always the last event of the stream. + const factory RsUploadEvent.failed({ + required RsHttpClientError error, + }) = RsUploadEvent_Failed; +} diff --git a/packages/localsend_isolates/lib/rust/api/http.freezed.dart b/packages/localsend_isolates/lib/rust/api/http.freezed.dart index b1da4f1f..463f09e4 100644 --- a/packages/localsend_isolates/lib/rust/api/http.freezed.dart +++ b/packages/localsend_isolates/lib/rust/api/http.freezed.dart @@ -521,4 +521,305 @@ as String, } +/// @nodoc +mixin _$RsUploadEvent { + + + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is RsUploadEvent); +} + + +@override +int get hashCode => runtimeType.hashCode; + +@override +String toString() { + return 'RsUploadEvent()'; +} + + +} + +/// @nodoc +class $RsUploadEventCopyWith<$Res> { +$RsUploadEventCopyWith(RsUploadEvent _, $Res Function(RsUploadEvent) __); +} + + +/// Adds pattern-matching-related methods to [RsUploadEvent]. +extension RsUploadEventPatterns on RsUploadEvent { +/// A variant of `map` that fallback to returning `orElse`. +/// +/// It is equivalent to doing: +/// ```dart +/// switch (sealedClass) { +/// case final Subclass value: +/// return ...; +/// case _: +/// return orElse(); +/// } +/// ``` + +@optionalTypeArgs TResult maybeMap({TResult Function( RsUploadEvent_Progress value)? progress,TResult Function( RsUploadEvent_Failed value)? failed,required TResult orElse(),}){ +final _that = this; +switch (_that) { +case RsUploadEvent_Progress() when progress != null: +return progress(_that);case RsUploadEvent_Failed() when failed != null: +return failed(_that);case _: + return orElse(); + +} +} +/// A `switch`-like method, using callbacks. +/// +/// Callbacks receives the raw object, upcasted. +/// It is equivalent to doing: +/// ```dart +/// switch (sealedClass) { +/// case final Subclass value: +/// return ...; +/// case final Subclass2 value: +/// return ...; +/// } +/// ``` + +@optionalTypeArgs TResult map({required TResult Function( RsUploadEvent_Progress value) progress,required TResult Function( RsUploadEvent_Failed value) failed,}){ +final _that = this; +switch (_that) { +case RsUploadEvent_Progress(): +return progress(_that);case RsUploadEvent_Failed(): +return failed(_that);} +} +/// A variant of `map` that fallback to returning `null`. +/// +/// It is equivalent to doing: +/// ```dart +/// switch (sealedClass) { +/// case final Subclass value: +/// return ...; +/// case _: +/// return null; +/// } +/// ``` + +@optionalTypeArgs TResult? mapOrNull({TResult? Function( RsUploadEvent_Progress value)? progress,TResult? Function( RsUploadEvent_Failed value)? failed,}){ +final _that = this; +switch (_that) { +case RsUploadEvent_Progress() when progress != null: +return progress(_that);case RsUploadEvent_Failed() when failed != null: +return failed(_that);case _: + return null; + +} +} +/// A variant of `when` that fallback to an `orElse` callback. +/// +/// It is equivalent to doing: +/// ```dart +/// switch (sealedClass) { +/// case Subclass(:final field): +/// return ...; +/// case _: +/// return orElse(); +/// } +/// ``` + +@optionalTypeArgs TResult maybeWhen({TResult Function( double progress)? progress,TResult Function( RsHttpClientError error)? failed,required TResult orElse(),}) {final _that = this; +switch (_that) { +case RsUploadEvent_Progress() when progress != null: +return progress(_that.progress);case RsUploadEvent_Failed() when failed != null: +return failed(_that.error);case _: + return orElse(); + +} +} +/// A `switch`-like method, using callbacks. +/// +/// As opposed to `map`, this offers destructuring. +/// It is equivalent to doing: +/// ```dart +/// switch (sealedClass) { +/// case Subclass(:final field): +/// return ...; +/// case Subclass2(:final field2): +/// return ...; +/// } +/// ``` + +@optionalTypeArgs TResult when({required TResult Function( double progress) progress,required TResult Function( RsHttpClientError error) failed,}) {final _that = this; +switch (_that) { +case RsUploadEvent_Progress(): +return progress(_that.progress);case RsUploadEvent_Failed(): +return failed(_that.error);} +} +/// A variant of `when` that fallback to returning `null` +/// +/// It is equivalent to doing: +/// ```dart +/// switch (sealedClass) { +/// case Subclass(:final field): +/// return ...; +/// case _: +/// return null; +/// } +/// ``` + +@optionalTypeArgs TResult? whenOrNull({TResult? Function( double progress)? progress,TResult? Function( RsHttpClientError error)? failed,}) {final _that = this; +switch (_that) { +case RsUploadEvent_Progress() when progress != null: +return progress(_that.progress);case RsUploadEvent_Failed() when failed != null: +return failed(_that.error);case _: + return null; + +} +} + +} + +/// @nodoc + + +class RsUploadEvent_Progress extends RsUploadEvent { + const RsUploadEvent_Progress({required this.progress}): super._(); + + + final double progress; + +/// Create a copy of RsUploadEvent +/// with the given fields replaced by the non-null parameter values. +@JsonKey(includeFromJson: false, includeToJson: false) +@pragma('vm:prefer-inline') +$RsUploadEvent_ProgressCopyWith get copyWith => _$RsUploadEvent_ProgressCopyWithImpl(this, _$identity); + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is RsUploadEvent_Progress&&(identical(other.progress, progress) || other.progress == progress)); +} + + +@override +int get hashCode => Object.hash(runtimeType,progress); + +@override +String toString() { + return 'RsUploadEvent.progress(progress: $progress)'; +} + + +} + +/// @nodoc +abstract mixin class $RsUploadEvent_ProgressCopyWith<$Res> implements $RsUploadEventCopyWith<$Res> { + factory $RsUploadEvent_ProgressCopyWith(RsUploadEvent_Progress value, $Res Function(RsUploadEvent_Progress) _then) = _$RsUploadEvent_ProgressCopyWithImpl; +@useResult +$Res call({ + double progress +}); + + + + +} +/// @nodoc +class _$RsUploadEvent_ProgressCopyWithImpl<$Res> + implements $RsUploadEvent_ProgressCopyWith<$Res> { + _$RsUploadEvent_ProgressCopyWithImpl(this._self, this._then); + + final RsUploadEvent_Progress _self; + final $Res Function(RsUploadEvent_Progress) _then; + +/// Create a copy of RsUploadEvent +/// with the given fields replaced by the non-null parameter values. +@pragma('vm:prefer-inline') $Res call({Object? progress = null,}) { + return _then(RsUploadEvent_Progress( +progress: null == progress ? _self.progress : progress // ignore: cast_nullable_to_non_nullable +as double, + )); +} + + +} + +/// @nodoc + + +class RsUploadEvent_Failed extends RsUploadEvent { + const RsUploadEvent_Failed({required this.error}): super._(); + + + final RsHttpClientError error; + +/// Create a copy of RsUploadEvent +/// with the given fields replaced by the non-null parameter values. +@JsonKey(includeFromJson: false, includeToJson: false) +@pragma('vm:prefer-inline') +$RsUploadEvent_FailedCopyWith get copyWith => _$RsUploadEvent_FailedCopyWithImpl(this, _$identity); + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is RsUploadEvent_Failed&&(identical(other.error, error) || other.error == error)); +} + + +@override +int get hashCode => Object.hash(runtimeType,error); + +@override +String toString() { + return 'RsUploadEvent.failed(error: $error)'; +} + + +} + +/// @nodoc +abstract mixin class $RsUploadEvent_FailedCopyWith<$Res> implements $RsUploadEventCopyWith<$Res> { + factory $RsUploadEvent_FailedCopyWith(RsUploadEvent_Failed value, $Res Function(RsUploadEvent_Failed) _then) = _$RsUploadEvent_FailedCopyWithImpl; +@useResult +$Res call({ + RsHttpClientError error +}); + + +$RsHttpClientErrorCopyWith<$Res> get error; + +} +/// @nodoc +class _$RsUploadEvent_FailedCopyWithImpl<$Res> + implements $RsUploadEvent_FailedCopyWith<$Res> { + _$RsUploadEvent_FailedCopyWithImpl(this._self, this._then); + + final RsUploadEvent_Failed _self; + final $Res Function(RsUploadEvent_Failed) _then; + +/// Create a copy of RsUploadEvent +/// with the given fields replaced by the non-null parameter values. +@pragma('vm:prefer-inline') $Res call({Object? error = null,}) { + return _then(RsUploadEvent_Failed( +error: null == error ? _self.error : error // ignore: cast_nullable_to_non_nullable +as RsHttpClientError, + )); +} + +/// Create a copy of RsUploadEvent +/// with the given fields replaced by the non-null parameter values. +@override +@pragma('vm:prefer-inline') +$RsHttpClientErrorCopyWith<$Res> get error { + + return $RsHttpClientErrorCopyWith<$Res>(_self.error, (value) { + return _then(_self.copyWith(error: value)); + }); +} +} + // dart format on diff --git a/packages/localsend_isolates/lib/rust/api/server.dart b/packages/localsend_isolates/lib/rust/api/server.dart index 61a8d935..388444bc 100644 --- a/packages/localsend_isolates/lib/rust/api/server.dart +++ b/packages/localsend_isolates/lib/rust/api/server.dart @@ -97,7 +97,10 @@ abstract class RsHttpServer implements RustOpaqueInterface { /// and waits until the file has been received completely. /// /// The progress (fraction of [file_size]) is emitted on [sink] - /// while the file is being received. + /// while the file is being received. Failures are emitted on [sink] as + /// well: flutter_rust_bridge discards the returned `Result` of functions + /// taking a [StreamSink], so a returned error would become an uncaught + /// async error killing the calling isolate. /// /// Timestamps provided in the sender's file metadata are applied to the /// written file by the server. diff --git a/packages/localsend_isolates/lib/rust/frb_generated.dart b/packages/localsend_isolates/lib/rust/frb_generated.dart index ee877322..1021e789 100644 --- a/packages/localsend_isolates/lib/rust/frb_generated.dart +++ b/packages/localsend_isolates/lib/rust/frb_generated.dart @@ -166,7 +166,7 @@ abstract class RustLibApi extends BaseApi { required RegisterDto payload, }); - Stream crateApiHttpRsHttpClientUpload({ + Stream crateApiHttpRsHttpClientUpload({ required RsHttpClient that, required ProtocolType protocol, required String ip, @@ -897,7 +897,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { ); @override - Stream crateApiHttpRsHttpClientUpload({ + Stream crateApiHttpRsHttpClientUpload({ required RsHttpClient that, required ProtocolType protocol, required String ip, @@ -912,14 +912,14 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { required BigInt contentLength, required RsCancellationToken cancelToken, }) { - final sink = RustStreamSink(); + final sink = RustStreamSink(); unawaited( handler.executeNormal( NormalTask( callFfi: (port_) { final serializer = SseSerializer(generalizedFrbRustBinding); sse_encode_Auto_Ref_RustOpaque_flutter_rust_bridgefor_generatedRustAutoOpaqueInnerRsHttpClient(that, serializer); - sse_encode_StreamSink_f_64_Sse(sink, serializer); + sse_encode_StreamSink_rs_upload_event_Sse(sink, serializer); sse_encode_protocol_type(protocol, serializer); sse_encode_String(ip, serializer); sse_encode_u_16(port, serializer); @@ -939,7 +939,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { }, codec: SseCodec( decodeSuccessData: sse_decode_unit, - decodeErrorData: sse_decode_rs_http_client_error, + decodeErrorData: null, ), constMeta: kCrateApiHttpRsHttpClientUploadConstMeta, argValues: [that, sink, protocol, ip, port, publicKey, sessionId, fileId, token, binary, path, fileDescriptor, contentLength, cancelToken], @@ -1141,7 +1141,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { }, codec: SseCodec( decodeSuccessData: sse_decode_unit, - decodeErrorData: sse_decode_AnyhowException, + decodeErrorData: null, ), constMeta: kCrateApiServerRsHttpServerRespondFileUploadConstMeta, argValues: [that, sink, sessionId, fileId, path, fileDescriptor, fileSize], @@ -1885,7 +1885,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { }, codec: SseCodec( decodeSuccessData: sse_decode_unit, - decodeErrorData: sse_decode_AnyhowException, + decodeErrorData: null, ), constMeta: kCrateApiCryptoHashFileConstMeta, argValues: [sink, path, fileDescriptor, bytes, cancelToken], @@ -2478,6 +2478,12 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { throw UnimplementedError(); } + @protected + RustStreamSink dco_decode_StreamSink_rs_upload_event_Sse(dynamic raw) { + // Codec=Dco (DartCObject based), see doc to use other codecs + throw UnimplementedError(); + } + @protected RustStreamSink dco_decode_StreamSink_rtc_file_error_Sse(dynamic raw) { // Codec=Dco (DartCObject based), see doc to use other codecs @@ -2606,6 +2612,12 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { return dco_decode_rs_discovered_device(raw); } + @protected + RsHttpClientError dco_decode_box_autoadd_rs_http_client_error(dynamic raw) { + // Codec=Dco (DartCObject based), see doc to use other codecs + return dco_decode_rs_http_client_error(raw); + } + @protected RsStoredDevice dco_decode_box_autoadd_rs_stored_device(dynamic raw) { // Codec=Dco (DartCObject based), see doc to use other codecs @@ -3213,6 +3225,23 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { ); } + @protected + RsUploadEvent dco_decode_rs_upload_event(dynamic raw) { + // Codec=Dco (DartCObject based), see doc to use other codecs + switch (raw[0]) { + case 0: + return RsUploadEvent_Progress( + progress: dco_decode_f_64(raw[1]), + ); + case 1: + return RsUploadEvent_Failed( + error: dco_decode_box_autoadd_rs_http_client_error(raw[1]), + ); + default: + throw Exception('unreachable'); + } + } + @protected RTCFileError dco_decode_rtc_file_error(dynamic raw) { // Codec=Dco (DartCObject based), see doc to use other codecs @@ -3718,6 +3747,12 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { throw UnimplementedError('Unreachable ()'); } + @protected + RustStreamSink sse_decode_StreamSink_rs_upload_event_Sse(SseDeserializer deserializer) { + // Codec=Sse (Serialization based), see doc to use other codecs + throw UnimplementedError('Unreachable ()'); + } + @protected RustStreamSink sse_decode_StreamSink_rtc_file_error_Sse(SseDeserializer deserializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -3848,6 +3883,12 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { return (sse_decode_rs_discovered_device(deserializer)); } + @protected + RsHttpClientError sse_decode_box_autoadd_rs_http_client_error(SseDeserializer deserializer) { + // Codec=Sse (Serialization based), see doc to use other codecs + return (sse_decode_rs_http_client_error(deserializer)); + } + @protected RsStoredDevice sse_decode_box_autoadd_rs_stored_device(SseDeserializer deserializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -4549,6 +4590,23 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { ); } + @protected + RsUploadEvent sse_decode_rs_upload_event(SseDeserializer deserializer) { + // Codec=Sse (Serialization based), see doc to use other codecs + + var tag_ = sse_decode_i_32(deserializer); + switch (tag_) { + case 0: + var var_progress = sse_decode_f_64(deserializer); + return RsUploadEvent_Progress(progress: var_progress); + case 1: + var var_error = sse_decode_box_autoadd_rs_http_client_error(deserializer); + return RsUploadEvent_Failed(error: var_error); + default: + throw UnimplementedError(''); + } + } + @protected RTCFileError sse_decode_rtc_file_error(SseDeserializer deserializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -5133,6 +5191,20 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { ); } + @protected + void sse_encode_StreamSink_rs_upload_event_Sse(RustStreamSink self, SseSerializer serializer) { + // Codec=Sse (Serialization based), see doc to use other codecs + sse_encode_String( + self.setupAndSerialize( + codec: SseCodec( + decodeSuccessData: sse_decode_rs_upload_event, + decodeErrorData: sse_decode_AnyhowException, + ), + ), + serializer, + ); + } + @protected void sse_encode_StreamSink_rtc_file_error_Sse(RustStreamSink self, SseSerializer serializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -5286,6 +5358,12 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { sse_encode_rs_discovered_device(self, serializer); } + @protected + void sse_encode_box_autoadd_rs_http_client_error(RsHttpClientError self, SseSerializer serializer) { + // Codec=Sse (Serialization based), see doc to use other codecs + sse_encode_rs_http_client_error(self, serializer); + } + @protected void sse_encode_box_autoadd_rs_stored_device(RsStoredDevice self, SseSerializer serializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -5868,6 +5946,19 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { sse_encode_list_rs_device_channel(self.channels, serializer); } + @protected + void sse_encode_rs_upload_event(RsUploadEvent self, SseSerializer serializer) { + // Codec=Sse (Serialization based), see doc to use other codecs + switch (self) { + case RsUploadEvent_Progress(progress: final progress): + sse_encode_i_32(0, serializer); + sse_encode_f_64(progress, serializer); + case RsUploadEvent_Failed(error: final error): + sse_encode_i_32(1, serializer); + sse_encode_box_autoadd_rs_http_client_error(error, serializer); + } + } + @protected void sse_encode_rtc_file_error(RTCFileError self, SseSerializer serializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -6269,7 +6360,13 @@ class RsHttpClientImpl extends RustOpaque implements RsHttpClient { required RegisterDto payload, }) => RustLib.instance.api.crateApiHttpRsHttpClientRegister(that: this, protocol: protocol, ip: ip, port: port, payload: payload); - Stream upload({ + /// Uploads a single file, emitting [RsUploadEvent]s on [sink]. + /// + /// Failures are emitted as [RsUploadEvent::Failed] instead of being + /// returned: flutter_rust_bridge discards the returned `Result` of + /// functions taking a [StreamSink], so a returned error would become an + /// uncaught async error killing the calling isolate. + Stream upload({ required ProtocolType protocol, required String ip, required int port, @@ -6363,7 +6460,10 @@ class RsHttpServerImpl extends RustOpaque implements RsHttpServer { /// and waits until the file has been received completely. /// /// The progress (fraction of [file_size]) is emitted on [sink] - /// while the file is being received. + /// while the file is being received. Failures are emitted on [sink] as + /// well: flutter_rust_bridge discards the returned `Result` of functions + /// taking a [StreamSink], so a returned error would become an uncaught + /// async error killing the calling isolate. /// /// Timestamps provided in the sender's file metadata are applied to the /// written file by the server. diff --git a/packages/localsend_isolates/lib/rust/frb_generated.io.dart b/packages/localsend_isolates/lib/rust/frb_generated.io.dart index e96c3599..6799b7a7 100644 --- a/packages/localsend_isolates/lib/rust/frb_generated.io.dart +++ b/packages/localsend_isolates/lib/rust/frb_generated.io.dart @@ -202,6 +202,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RustStreamSink dco_decode_StreamSink_rs_stored_device_Sse(dynamic raw); + @protected + RustStreamSink dco_decode_StreamSink_rs_upload_event_Sse(dynamic raw); + @protected RustStreamSink dco_decode_StreamSink_rtc_file_error_Sse(dynamic raw); @@ -267,6 +270,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RsDiscoveredDevice dco_decode_box_autoadd_rs_discovered_device(dynamic raw); + @protected + RsHttpClientError dco_decode_box_autoadd_rs_http_client_error(dynamic raw); + @protected RsStoredDevice dco_decode_box_autoadd_rs_stored_device(dynamic raw); @@ -451,6 +457,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RsStoredDevice dco_decode_rs_stored_device(dynamic raw); + @protected + RsUploadEvent dco_decode_rs_upload_event(dynamic raw); + @protected RTCFileError dco_decode_rtc_file_error(dynamic raw); @@ -660,6 +669,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RustStreamSink sse_decode_StreamSink_rs_stored_device_Sse(SseDeserializer deserializer); + @protected + RustStreamSink sse_decode_StreamSink_rs_upload_event_Sse(SseDeserializer deserializer); + @protected RustStreamSink sse_decode_StreamSink_rtc_file_error_Sse(SseDeserializer deserializer); @@ -725,6 +737,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RsDiscoveredDevice sse_decode_box_autoadd_rs_discovered_device(SseDeserializer deserializer); + @protected + RsHttpClientError sse_decode_box_autoadd_rs_http_client_error(SseDeserializer deserializer); + @protected RsStoredDevice sse_decode_box_autoadd_rs_stored_device(SseDeserializer deserializer); @@ -911,6 +926,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RsStoredDevice sse_decode_rs_stored_device(SseDeserializer deserializer); + @protected + RsUploadEvent sse_decode_rs_upload_event(SseDeserializer deserializer); + @protected RTCFileError sse_decode_rtc_file_error(SseDeserializer deserializer); @@ -1164,6 +1182,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void sse_encode_StreamSink_rs_stored_device_Sse(RustStreamSink self, SseSerializer serializer); + @protected + void sse_encode_StreamSink_rs_upload_event_Sse(RustStreamSink self, SseSerializer serializer); + @protected void sse_encode_StreamSink_rtc_file_error_Sse(RustStreamSink self, SseSerializer serializer); @@ -1230,6 +1251,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void sse_encode_box_autoadd_rs_discovered_device(RsDiscoveredDevice self, SseSerializer serializer); + @protected + void sse_encode_box_autoadd_rs_http_client_error(RsHttpClientError self, SseSerializer serializer); + @protected void sse_encode_box_autoadd_rs_stored_device(RsStoredDevice self, SseSerializer serializer); @@ -1417,6 +1441,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void sse_encode_rs_stored_device(RsStoredDevice self, SseSerializer serializer); + @protected + void sse_encode_rs_upload_event(RsUploadEvent self, SseSerializer serializer); + @protected void sse_encode_rtc_file_error(RTCFileError self, SseSerializer serializer); diff --git a/packages/localsend_isolates/lib/rust/frb_generated.web.dart b/packages/localsend_isolates/lib/rust/frb_generated.web.dart index dafd558d..d4c79868 100644 --- a/packages/localsend_isolates/lib/rust/frb_generated.web.dart +++ b/packages/localsend_isolates/lib/rust/frb_generated.web.dart @@ -204,6 +204,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RustStreamSink dco_decode_StreamSink_rs_stored_device_Sse(dynamic raw); + @protected + RustStreamSink dco_decode_StreamSink_rs_upload_event_Sse(dynamic raw); + @protected RustStreamSink dco_decode_StreamSink_rtc_file_error_Sse(dynamic raw); @@ -269,6 +272,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RsDiscoveredDevice dco_decode_box_autoadd_rs_discovered_device(dynamic raw); + @protected + RsHttpClientError dco_decode_box_autoadd_rs_http_client_error(dynamic raw); + @protected RsStoredDevice dco_decode_box_autoadd_rs_stored_device(dynamic raw); @@ -453,6 +459,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RsStoredDevice dco_decode_rs_stored_device(dynamic raw); + @protected + RsUploadEvent dco_decode_rs_upload_event(dynamic raw); + @protected RTCFileError dco_decode_rtc_file_error(dynamic raw); @@ -662,6 +671,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RustStreamSink sse_decode_StreamSink_rs_stored_device_Sse(SseDeserializer deserializer); + @protected + RustStreamSink sse_decode_StreamSink_rs_upload_event_Sse(SseDeserializer deserializer); + @protected RustStreamSink sse_decode_StreamSink_rtc_file_error_Sse(SseDeserializer deserializer); @@ -727,6 +739,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RsDiscoveredDevice sse_decode_box_autoadd_rs_discovered_device(SseDeserializer deserializer); + @protected + RsHttpClientError sse_decode_box_autoadd_rs_http_client_error(SseDeserializer deserializer); + @protected RsStoredDevice sse_decode_box_autoadd_rs_stored_device(SseDeserializer deserializer); @@ -913,6 +928,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected RsStoredDevice sse_decode_rs_stored_device(SseDeserializer deserializer); + @protected + RsUploadEvent sse_decode_rs_upload_event(SseDeserializer deserializer); + @protected RTCFileError sse_decode_rtc_file_error(SseDeserializer deserializer); @@ -1166,6 +1184,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void sse_encode_StreamSink_rs_stored_device_Sse(RustStreamSink self, SseSerializer serializer); + @protected + void sse_encode_StreamSink_rs_upload_event_Sse(RustStreamSink self, SseSerializer serializer); + @protected void sse_encode_StreamSink_rtc_file_error_Sse(RustStreamSink self, SseSerializer serializer); @@ -1232,6 +1253,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void sse_encode_box_autoadd_rs_discovered_device(RsDiscoveredDevice self, SseSerializer serializer); + @protected + void sse_encode_box_autoadd_rs_http_client_error(RsHttpClientError self, SseSerializer serializer); + @protected void sse_encode_box_autoadd_rs_stored_device(RsStoredDevice self, SseSerializer serializer); @@ -1419,6 +1443,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void sse_encode_rs_stored_device(RsStoredDevice self, SseSerializer serializer); + @protected + void sse_encode_rs_upload_event(RsUploadEvent self, SseSerializer serializer); + @protected void sse_encode_rtc_file_error(RTCFileError self, SseSerializer serializer); diff --git a/packages/localsend_isolates/lib/src/isolate/child/main.dart b/packages/localsend_isolates/lib/src/isolate/child/main.dart index 7f5b26ff..09f7302a 100644 --- a/packages/localsend_isolates/lib/src/isolate/child/main.dart +++ b/packages/localsend_isolates/lib/src/isolate/child/main.dart @@ -40,25 +40,34 @@ Future setupChildIsolateHelper({ }) async { initLogger(initialData.logLevel); - _isolateContainer.set( - syncProvider.overrideWithNotifier( - (ref) => SyncService( - initial: initialData.syncState, - ), - ), + // Uncaught async errors are fatal to the isolate by default, which would + // silently kill the server/upload loop; log them instead. + await runZonedGuarded( + () async { + _isolateContainer.set( + syncProvider.overrideWithNotifier( + (ref) => SyncService( + initial: initialData.syncState, + ), + ), + ); + + await RustLib.init(); + + if (init != null) { + await init(_isolateContainer); + } + + _logger.info('Child isolate is ready: $debugLabel (logLevel: ${initialData.logLevel})'); + + await for (final message in receiveFromMain) { + _handleMessage(debugLabel, message, handler); + } + }, + (e, st) { + _logger.severe('Uncaught error in $debugLabel', e, st); + }, ); - - await RustLib.init(); - - if (init != null) { - await init(_isolateContainer); - } - - _logger.info('Child isolate is ready: $debugLabel (logLevel: ${initialData.logLevel})'); - - await for (final message in receiveFromMain) { - _handleMessage(debugLabel, message, handler); - } } // separate function to avoid blocking the for loop diff --git a/packages/localsend_isolates/lib/src/task/upload/http_upload.dart b/packages/localsend_isolates/lib/src/task/upload/http_upload.dart index ceb05b2d..251881f4 100644 --- a/packages/localsend_isolates/lib/src/task/upload/http_upload.dart +++ b/packages/localsend_isolates/lib/src/task/upload/http_upload.dart @@ -48,7 +48,15 @@ class HttpUploadService { contentLength: BigInt.from(contentLength), cancelToken: cancelToken, ) - .forEach(onSendProgress); + .forEach((event) { + switch (event) { + case RsUploadEvent_Progress(:final progress): + onSendProgress(progress); + case RsUploadEvent_Failed(:final error): + // Fails [uploadFuture] with the typed client error. + throw error; + } + }); try { await for (final chunk in stream ?? const Stream>.empty()) { diff --git a/packages/localsend_isolates/lib/util/rust.dart b/packages/localsend_isolates/lib/util/rust.dart index 86d1ed52..738d6620 100644 --- a/packages/localsend_isolates/lib/util/rust.dart +++ b/packages/localsend_isolates/lib/util/rust.dart @@ -1,3 +1,4 @@ +import 'package:flutter_rust_bridge/flutter_rust_bridge.dart' show AnyhowException; import 'package:localsend_isolates/constants.dart'; import 'package:localsend_isolates/model/device.dart'; import 'package:localsend_isolates/model/dto/file_dto.dart'; @@ -105,6 +106,7 @@ extension HumanErrorMessageExt on Object { final e = this; return switch (e) { rust_http.RsHttpClientError_StatusCode(:final status, :final message) when message != null => '[$status] $message', + AnyhowException(:final message) => message, _ => e.toString(), }; } diff --git a/packages/localsend_isolates/pubspec.lock b/packages/localsend_isolates/pubspec.lock deleted file mode 100644 index 551b4435..00000000 --- a/packages/localsend_isolates/pubspec.lock +++ /dev/null @@ -1,736 +0,0 @@ -# Generated by pub -# See https://dart.dev/tools/pub/glossary#lockfile -packages: - _fe_analyzer_shared: - dependency: transitive - description: - name: _fe_analyzer_shared - sha256: "8d7ff3948166b8ec5da0fbb5962000926b8e02f2ed9b3e51d1738905fbd4c98d" - url: "https://pub.dev" - source: hosted - version: "93.0.0" - analyzer: - dependency: transitive - description: - name: analyzer - sha256: de7148ed2fcec579b19f122c1800933dfa028f6d9fd38a152b04b1516cec120b - url: "https://pub.dev" - source: hosted - version: "10.0.1" - ansicolor: - dependency: transitive - description: - name: ansicolor - sha256: "50e982d500bc863e1d703448afdbf9e5a72eb48840a4f766fa361ffd6877055f" - url: "https://pub.dev" - source: hosted - version: "2.0.3" - args: - dependency: transitive - description: - name: args - sha256: d0481093c50b1da8910eb0bb301626d4d8eb7284aa739614d2b394ee09e3ea04 - url: "https://pub.dev" - source: hosted - version: "2.7.0" - async: - dependency: transitive - description: - name: async - sha256: e2eb0491ba5ddb6177742d2da23904574082139b07c1e33b8503b9f46f3e1a37 - url: "https://pub.dev" - source: hosted - version: "2.13.1" - boolean_selector: - dependency: transitive - description: - name: boolean_selector - sha256: "8aab1771e1243a5063b8b0ff68042d67334e3feab9e95b9490f9a6ebf73b42ea" - url: "https://pub.dev" - source: hosted - version: "2.1.2" - build: - dependency: transitive - description: - name: build - sha256: "45d14a0fb23e018d8287c32fc98d726ce466b231928ed9b9200f29bd3ccd39ae" - url: "https://pub.dev" - source: hosted - version: "4.0.7" - build_cli_annotations: - dependency: transitive - description: - name: build_cli_annotations - sha256: e563c2e01de8974566a1998410d3f6f03521788160a02503b0b1f1a46c7b3d95 - url: "https://pub.dev" - source: hosted - version: "2.1.1" - build_config: - dependency: transitive - description: - name: build_config - sha256: f2c223156a26eea323e6244b85141d76413a80aeee9fe0b380773789fabaf8ae - url: "https://pub.dev" - source: hosted - version: "1.3.1" - build_daemon: - dependency: transitive - description: - name: build_daemon - sha256: fd754058c342243718d5171a95f352cfc9fcf0cba8cfa26df67cb13a5836db78 - url: "https://pub.dev" - source: hosted - version: "4.1.2" - build_runner: - dependency: "direct dev" - description: - name: build_runner - sha256: "5367e521935b102bdf1e735d2aab461e36b2edca6517662d088dd04cc39f8d16" - url: "https://pub.dev" - source: hosted - version: "2.15.1" - built_collection: - dependency: transitive - description: - name: built_collection - sha256: "376e3dd27b51ea877c28d525560790aee2e6fbb5f20e2f85d5081027d94e2100" - url: "https://pub.dev" - source: hosted - version: "5.1.1" - built_value: - dependency: transitive - description: - name: built_value - sha256: "34e4067d30ce212937df995f03b69992eea683539ceeac7f679a1f1eba055b56" - url: "https://pub.dev" - source: hosted - version: "8.12.6" - characters: - dependency: transitive - description: - name: characters - sha256: faf38497bda5ead2a8c7615f4f7939df04333478bf32e4173fcb06d428b5716b - url: "https://pub.dev" - source: hosted - version: "1.4.1" - checked_yaml: - dependency: transitive - description: - name: checked_yaml - sha256: "959525d3162f249993882720d52b7e0c833978df229be20702b33d48d91de70f" - url: "https://pub.dev" - source: hosted - version: "2.0.4" - clock: - dependency: transitive - description: - name: clock - sha256: fddb70d9b5277016c77a80201021d40a2247104d9f4aa7bab7157b7e3f05b84b - url: "https://pub.dev" - source: hosted - version: "1.1.2" - collection: - dependency: "direct main" - description: - name: collection - sha256: "2f5709ae4d3d59dd8f7cd309b4e023046b57d8a6c82130785d2b0e5868084e76" - url: "https://pub.dev" - source: hosted - version: "1.19.1" - convert: - dependency: transitive - description: - name: convert - sha256: b30acd5944035672bc15c6b7a8b47d773e41e2f17de064350988c5d02adb1c68 - url: "https://pub.dev" - source: hosted - version: "3.1.2" - crypto: - dependency: transitive - description: - name: crypto - sha256: c8ea0233063ba03258fbcf2ca4d6dadfefe14f02fab57702265467a19f27fadf - url: "https://pub.dev" - source: hosted - version: "3.0.7" - dart_mappable: - dependency: "direct main" - description: - name: dart_mappable - sha256: "960746478faaa68ed6b9d3c6fd03c87c7b8614e6c33e75fe1b0c6d7a60adcf29" - url: "https://pub.dev" - source: hosted - version: "4.8.0" - dart_mappable_builder: - dependency: "direct dev" - description: - name: dart_mappable_builder - sha256: eb89efebad96bae52333ae4bd6b8fdcfb44c921f8a38f44893f2c72ed15c860c - url: "https://pub.dev" - source: hosted - version: "4.8.0" - dart_style: - dependency: transitive - description: - name: dart_style - sha256: "29f7ecc274a86d32920b1d9cfc7502fa87220da41ec60b55f329559d5732e2b2" - url: "https://pub.dev" - source: hosted - version: "3.1.7" - fake_async: - dependency: transitive - description: - name: fake_async - sha256: "5368f224a74523e8d2e7399ea1638b37aecfca824a3cc4dfdf77bf1fa905ac44" - url: "https://pub.dev" - source: hosted - version: "1.3.3" - ffi: - dependency: transitive - description: - name: ffi - sha256: "6d7fd89431262d8f3125e81b50d3847a091d846eafcd4fdb88dd06f36d705a45" - url: "https://pub.dev" - source: hosted - version: "2.2.0" - file: - dependency: transitive - description: - name: file - sha256: a3b4f84adafef897088c160faf7dfffb7696046cb13ae90b508c2cbc95d3b8d4 - url: "https://pub.dev" - source: hosted - version: "7.0.1" - fixnum: - dependency: transitive - description: - name: fixnum - sha256: b6dc7065e46c974bc7c5f143080a6764ec7a4be6da1285ececdc37be96de53be - url: "https://pub.dev" - source: hosted - version: "1.1.1" - flutter: - dependency: "direct main" - description: flutter - source: sdk - version: "0.0.0" - flutter_foreground_task: - dependency: "direct main" - description: - name: flutter_foreground_task - sha256: fc5c01a5e1b8f7bb51d0c737714f0c50440dbdf1aeddc5f8cbba313aa6fd4856 - url: "https://pub.dev" - source: hosted - version: "9.2.2" - flutter_lints: - dependency: "direct dev" - description: - name: flutter_lints - sha256: "3105dc8492f6183fb076ccf1f351ac3d60564bff92e20bfc4af9cc1651f4e7e1" - url: "https://pub.dev" - source: hosted - version: "6.0.0" - flutter_rust_bridge: - dependency: "direct main" - description: - name: flutter_rust_bridge - sha256: e87d6b9ee934dcd24a128ccb2bd91905d2d5fe5c06245d6a8f5477d4907a437a - url: "https://pub.dev" - source: hosted - version: "2.12.0" - flutter_test: - dependency: "direct dev" - description: flutter - source: sdk - version: "0.0.0" - flutter_web_plugins: - dependency: transitive - description: flutter - source: sdk - version: "0.0.0" - freezed: - dependency: "direct dev" - description: - name: freezed - sha256: f23ea33b3863f119b58ed1b586e881a46bd28715ddcc4dbc33104524e3434131 - url: "https://pub.dev" - source: hosted - version: "3.2.5" - freezed_annotation: - dependency: "direct main" - description: - name: freezed_annotation - sha256: "7294967ff0a6d98638e7acb774aac3af2550777accd8149c90af5b014e6d44d8" - url: "https://pub.dev" - source: hosted - version: "3.1.0" - gal: - dependency: "direct main" - description: - name: gal - sha256: "969598f986789127fd407a750413249e1352116d4c2be66e81837ffeeaafdfee" - url: "https://pub.dev" - source: hosted - version: "2.3.2" - glob: - dependency: transitive - description: - name: glob - sha256: c3f1ee72c96f8f78935e18aa8cecced9ab132419e8625dc187e1c2408efc20de - url: "https://pub.dev" - source: hosted - version: "2.1.3" - graphs: - dependency: transitive - description: - name: graphs - sha256: "741bbf84165310a68ff28fe9e727332eef1407342fca52759cb21ad8177bb8d0" - url: "https://pub.dev" - source: hosted - version: "2.3.2" - http_multi_server: - dependency: transitive - description: - name: http_multi_server - sha256: aa6199f908078bb1c5efb8d8638d4ae191aac11b311132c3ef48ce352fb52ef8 - url: "https://pub.dev" - source: hosted - version: "3.2.2" - http_parser: - dependency: transitive - description: - name: http_parser - sha256: "178d74305e7866013777bab2c3d8726205dc5a4dd935297175b19a23a2e66571" - url: "https://pub.dev" - source: hosted - version: "4.1.2" - io: - dependency: transitive - description: - name: io - sha256: dfd5a80599cf0165756e3181807ed3e77daf6dd4137caaad72d0b7931597650b - url: "https://pub.dev" - source: hosted - version: "1.0.5" - json_annotation: - dependency: transitive - description: - name: json_annotation - sha256: "2a743920d81b7910627f68ee2c9ac1fc0bfee32b9fc3403587d7c6791ca12f80" - url: "https://pub.dev" - source: hosted - version: "4.12.0" - leak_tracker: - dependency: transitive - description: - name: leak_tracker - sha256: "33e2e26bdd85a0112ec15400c8cbffea70d0f9c3407491f672a2fad47915e2de" - url: "https://pub.dev" - source: hosted - version: "11.0.2" - leak_tracker_flutter_testing: - dependency: transitive - description: - name: leak_tracker_flutter_testing - sha256: "1dbc140bb5a23c75ea9c4811222756104fbcd1a27173f0c34ca01e16bea473c1" - url: "https://pub.dev" - source: hosted - version: "3.0.10" - leak_tracker_testing: - dependency: transitive - description: - name: leak_tracker_testing - sha256: "8d5a2d49f4a66b49744b23b018848400d23e54caf9463f4eb20df3eb8acb2eb1" - url: "https://pub.dev" - source: hosted - version: "3.0.2" - lints: - dependency: transitive - description: - name: lints - sha256: "12f842a479589fea194fe5c5a3095abc7be0c1f2ddfa9a0e76aed1dbd26a87df" - url: "https://pub.dev" - source: hosted - version: "6.1.0" - logging: - dependency: "direct main" - description: - name: logging - sha256: c8245ada5f1717ed44271ed1c26b8ce85ca3228fd2ffdb75468ab01979309d61 - url: "https://pub.dev" - source: hosted - version: "1.3.0" - matcher: - dependency: transitive - description: - name: matcher - sha256: dc0b7dc7651697ea4ff3e69ef44b0407ea32c487a39fff6a4004fa585e901861 - url: "https://pub.dev" - source: hosted - version: "0.12.19" - material_color_utilities: - dependency: transitive - description: - name: material_color_utilities - sha256: "9c337007e82b1889149c82ed242ed1cb24a66044e30979c44912381e9be4c48b" - url: "https://pub.dev" - source: hosted - version: "0.13.0" - meta: - dependency: transitive - description: - name: meta - sha256: "23f08335362185a5ea2ad3a4e597f1375e78bce8a040df5c600c8d3552ef2394" - url: "https://pub.dev" - source: hosted - version: "1.17.0" - mime: - dependency: "direct main" - description: - name: mime - sha256: "41a20518f0cb1256669420fdba0cd90d21561e560ac240f26ef8322e45bb7ed6" - url: "https://pub.dev" - source: hosted - version: "2.0.0" - package_config: - dependency: transitive - description: - name: package_config - sha256: f096c55ebb7deb7e384101542bfba8c52696c1b56fca2eb62827989ef2353bbc - url: "https://pub.dev" - source: hosted - version: "2.2.0" - path: - dependency: "direct main" - description: - name: path - sha256: "75cca69d1490965be98c73ceaea117e8a04dd21217b37b292c9ddbec0d955bc5" - url: "https://pub.dev" - source: hosted - version: "1.9.1" - path_provider_linux: - dependency: transitive - description: - name: path_provider_linux - sha256: "58c2005f147315b11e9b4a7bc889cd5203e250cba8e3f012dae259b4972b5c16" - url: "https://pub.dev" - source: hosted - version: "2.2.2" - path_provider_platform_interface: - dependency: transitive - description: - name: path_provider_platform_interface - sha256: "484838772624c3a4b94f1e44a3e19897fee738f2d5c4ce448443b0417f7c9dda" - url: "https://pub.dev" - source: hosted - version: "2.1.3" - path_provider_windows: - dependency: transitive - description: - name: path_provider_windows - sha256: bd6f00dbd873bfb70d0761682da2b3a2c2fccc2b9e84c495821639601d81afe7 - url: "https://pub.dev" - source: hosted - version: "2.3.0" - platform: - dependency: transitive - description: - name: platform - sha256: "5d6b1b0036a5f331ebc77c850ebc8506cbc1e9416c27e59b439f917a902a4984" - url: "https://pub.dev" - source: hosted - version: "3.1.6" - plugin_platform_interface: - dependency: transitive - description: - name: plugin_platform_interface - sha256: "4820fbfdb9478b1ebae27888254d445073732dae3d6ea81f0b7e06d5dedc3f02" - url: "https://pub.dev" - source: hosted - version: "2.1.8" - pool: - dependency: "direct main" - description: - name: pool - sha256: "978783255c543aa3586a1b3c21f6e9d720eb315376a915872c61ef8b5c20177d" - url: "https://pub.dev" - source: hosted - version: "1.5.2" - pub_semver: - dependency: transitive - description: - name: pub_semver - sha256: "5bfcf68ca79ef689f8990d1160781b4bad40a3bd5e5218ad4076ddb7f4081585" - url: "https://pub.dev" - source: hosted - version: "2.2.0" - pubspec_parse: - dependency: transitive - description: - name: pubspec_parse - sha256: "0560ba233314abbed0a48a2956f7f022cce7c3e1e73df540277da7544cad4082" - url: "https://pub.dev" - source: hosted - version: "1.5.0" - refena: - dependency: transitive - description: - name: refena - sha256: d2bb36fd39412563a4515b8228271cf8786936f77d4a6261b05eecf648e59b6b - url: "https://pub.dev" - source: hosted - version: "3.2.2" - refena_flutter: - dependency: "direct main" - description: - name: refena_flutter - sha256: "18f704c3ba38a30aa05ae71b3c55777b8d09864570e3b88b2f8814644381e461" - url: "https://pub.dev" - source: hosted - version: "3.2.1" - rust_lib_localsend_app: - dependency: "direct main" - description: - path: rust_builder - relative: true - source: path - version: "0.0.1" - shared_preferences: - dependency: transitive - description: - name: shared_preferences - sha256: c3025c5534b01739267eb7d76959bbc25a6d10f6988e1c2a3036940133dd10bf - url: "https://pub.dev" - source: hosted - version: "2.5.5" - shared_preferences_android: - dependency: transitive - description: - name: shared_preferences_android - sha256: e8d4762b1e2e8578fc4d0fd548cebf24afd24f49719c08974df92834565e2c53 - url: "https://pub.dev" - source: hosted - version: "2.4.23" - shared_preferences_foundation: - dependency: transitive - description: - name: shared_preferences_foundation - sha256: "4e7eaffc2b17ba398759f1151415869a34771ba11ebbccd1b0145472a619a64f" - url: "https://pub.dev" - source: hosted - version: "2.5.6" - shared_preferences_linux: - dependency: transitive - description: - name: shared_preferences_linux - sha256: "580abfd40f415611503cae30adf626e6656dfb2f0cee8f465ece7b6defb40f2f" - url: "https://pub.dev" - source: hosted - version: "2.4.1" - shared_preferences_platform_interface: - dependency: transitive - description: - name: shared_preferences_platform_interface - sha256: "649dc798a33931919ea356c4305c2d1f81619ea6e92244070b520187b5140ef9" - url: "https://pub.dev" - source: hosted - version: "2.4.2" - shared_preferences_web: - dependency: transitive - description: - name: shared_preferences_web - sha256: c49bd060261c9a3f0ff445892695d6212ff603ef3115edbb448509d407600019 - url: "https://pub.dev" - source: hosted - version: "2.4.3" - shared_preferences_windows: - dependency: transitive - description: - name: shared_preferences_windows - sha256: "94ef0f72b2d71bc3e700e025db3710911bd51a71cefb65cc609dd0d9a982e3c1" - url: "https://pub.dev" - source: hosted - version: "2.4.1" - shelf: - dependency: transitive - description: - name: shelf - sha256: e7dd780a7ffb623c57850b33f43309312fc863fb6aa3d276a754bb299839ef12 - url: "https://pub.dev" - source: hosted - version: "1.4.2" - shelf_web_socket: - dependency: transitive - description: - name: shelf_web_socket - sha256: "3632775c8e90d6c9712f883e633716432a27758216dfb61bd86a8321c0580925" - url: "https://pub.dev" - source: hosted - version: "3.0.0" - sky_engine: - dependency: transitive - description: flutter - source: sdk - version: "0.0.0" - source_gen: - dependency: transitive - description: - name: source_gen - sha256: ec37cc0e6694374cbef59ed79685572c870a54ede6fa30a3e420feb3adffea02 - url: "https://pub.dev" - source: hosted - version: "4.2.3" - source_span: - dependency: transitive - description: - name: source_span - sha256: "56a02f1f4cd1a2d96303c0144c93bd6d909eea6bee6bf5a0e0b685edbd4c47ab" - url: "https://pub.dev" - source: hosted - version: "1.10.2" - stack_trace: - dependency: transitive - description: - name: stack_trace - sha256: "8b27215b45d22309b5cddda1aa2b19bdfec9df0e765f2de506401c071d38d1b1" - url: "https://pub.dev" - source: hosted - version: "1.12.1" - stream_channel: - dependency: transitive - description: - name: stream_channel - sha256: "969e04c80b8bcdf826f8f16579c7b14d780458bd97f56d107d3950fdbeef059d" - url: "https://pub.dev" - source: hosted - version: "2.1.4" - stream_transform: - dependency: transitive - description: - name: stream_transform - sha256: ad47125e588cfd37a9a7f86c7d6356dde8dfe89d071d293f80ca9e9273a33871 - url: "https://pub.dev" - source: hosted - version: "2.1.1" - string_scanner: - dependency: transitive - description: - name: string_scanner - sha256: "921cd31725b72fe181906c6a94d987c78e3b98c2e205b397ea399d4054872b43" - url: "https://pub.dev" - source: hosted - version: "1.4.1" - term_glyph: - dependency: transitive - description: - name: term_glyph - sha256: "7f554798625ea768a7518313e58f83891c7f5024f88e46e7182a4558850a4b8e" - url: "https://pub.dev" - source: hosted - version: "1.2.2" - test_api: - dependency: transitive - description: - name: test_api - sha256: "8161c84903fd860b26bfdefb7963b3f0b68fee7adea0f59ef805ecca346f0c7a" - url: "https://pub.dev" - source: hosted - version: "0.7.10" - type_plus: - dependency: transitive - description: - name: type_plus - sha256: d5d1019471f0d38b91603adb9b5fd4ce7ab903c879d2fbf1a3f80a630a03fcc9 - url: "https://pub.dev" - source: hosted - version: "2.1.1" - typed_data: - dependency: transitive - description: - name: typed_data - sha256: f9049c039ebfeb4cf7a7104a675823cd72dba8297f264b6637062516699fa006 - url: "https://pub.dev" - source: hosted - version: "1.4.0" - typed_isolates: - dependency: "direct main" - description: - path: "../typed_isolates" - relative: true - source: path - version: "1.0.0" - uuid: - dependency: "direct main" - description: - name: uuid - sha256: "1fef9e8e11e2991bb773070d4656b7bd5d850967a2456cfc83cf47925ba79489" - url: "https://pub.dev" - source: hosted - version: "4.5.3" - vector_math: - dependency: transitive - description: - name: vector_math - sha256: d530bd74fea330e6e364cda7a85019c434070188383e1cd8d9777ee586914c5b - url: "https://pub.dev" - source: hosted - version: "2.2.0" - vm_service: - dependency: transitive - description: - name: vm_service - sha256: "0016aef94fc66495ac78af5859181e3f3bf2026bd8eecc72b9565601e19ab360" - url: "https://pub.dev" - source: hosted - version: "15.2.0" - watcher: - dependency: transitive - description: - name: watcher - sha256: "1398c9f081a753f9226febe8900fce8f7d0a67163334e1c94a2438339d79d635" - url: "https://pub.dev" - source: hosted - version: "1.2.1" - web: - dependency: transitive - description: - name: web - sha256: "868d88a33d8a87b18ffc05f9f030ba328ffefba92d6c127917a2ba740f9cfe4a" - url: "https://pub.dev" - source: hosted - version: "1.1.1" - web_socket: - dependency: transitive - description: - name: web_socket - sha256: "34d64019aa8e36bf9842ac014bb5d2f5586ca73df5e4d9bf5c936975cae6982c" - url: "https://pub.dev" - source: hosted - version: "1.0.1" - web_socket_channel: - dependency: transitive - description: - name: web_socket_channel - sha256: d645757fb0f4773d602444000a8131ff5d48c9e47adfe9772652dd1a4f2d45c8 - url: "https://pub.dev" - source: hosted - version: "3.0.3" - xdg_directories: - dependency: transitive - description: - name: xdg_directories - sha256: "7a3f37b05d989967cdddcbb571f1ea834867ae2faa29725fd085180e0883aa15" - url: "https://pub.dev" - source: hosted - version: "1.1.0" - yaml: - dependency: transitive - description: - name: yaml - sha256: b9da305ac7c39faa3f030eccd175340f968459dae4af175130b3fc47e40d76ce - url: "https://pub.dev" - source: hosted - version: "3.1.3" -sdks: - dart: ">=3.11.0 <4.0.0" - flutter: ">=3.41.0 <4.0.0" diff --git a/packages/localsend_isolates/rust/src/api/crypto.rs b/packages/localsend_isolates/rust/src/api/crypto.rs index 688ccfd6..78bab262 100644 --- a/packages/localsend_isolates/rust/src/api/crypto.rs +++ b/packages/localsend_isolates/rust/src/api/crypto.rs @@ -60,6 +60,11 @@ pub enum RsHashFileEvent { /// Cancelling [cancel_token] aborts the read, so hashing a large file does not /// have to be waited out. /// +/// Failures (including cancellation) are emitted as errors on the stream: +/// flutter_rust_bridge discards the returned `Result` of functions taking a +/// [StreamSink], so a returned error would become an uncaught async error +/// killing the calling isolate. +/// /// Exactly one content source must be provided: /// a [path] to a regular file, a [file_descriptor] (Android only), or [bytes] /// for a file that only lives in memory. @@ -69,47 +74,54 @@ pub async fn hash_file( file_descriptor: Option, bytes: Option>, cancel_token: &RsCancellationToken, -) -> anyhow::Result<()> { - let content = match (path, file_descriptor, bytes) { - (Some(path), None, None) => localsend::model::transfer::FileContent::Path(path.into()), - (None, Some(file_descriptor), None) => { - #[cfg(target_os = "android")] - { - localsend::model::transfer::FileContent::Fd(file_descriptor) - } - #[cfg(not(target_os = "android"))] - { - let _ = file_descriptor; - anyhow::bail!("File descriptors are only supported on Android"); - } - } - (None, None, Some(bytes)) => { - let hash = localsend::crypto::hash::sha256_hex(&bytes); - let _ = sink.add(RsHashFileEvent::Done { hash }); - return Ok(()); - } - _ => anyhow::bail!("Exactly one content source must be provided"), - }; - - // Progress events with throttling - let last_emit = std::cell::Cell::new(None::); - let progress = { - let sink = sink.clone(); - move |hashed| { - let now = std::time::Instant::now(); - if let Some(last) = last_emit.get() { - if now.duration_since(last) < std::time::Duration::from_millis(20) { - return; +) { + let result = async { + let content = match (path, file_descriptor, bytes) { + (Some(path), None, None) => localsend::model::transfer::FileContent::Path(path.into()), + (None, Some(file_descriptor), None) => { + #[cfg(target_os = "android")] + { + localsend::model::transfer::FileContent::Fd(file_descriptor) + } + #[cfg(not(target_os = "android"))] + { + let _ = file_descriptor; + anyhow::bail!("File descriptors are only supported on Android"); } } - last_emit.set(Some(now)); - let _ = sink.add(RsHashFileEvent::Progress { bytes: hashed }); - } - }; + (None, None, Some(bytes)) => { + let hash = localsend::crypto::hash::sha256_hex(&bytes); + let _ = sink.add(RsHashFileEvent::Done { hash }); + return Ok(()); + } + _ => anyhow::bail!("Exactly one content source must be provided"), + }; - let hash = - localsend::crypto::hash::sha256_file_content(content, &cancel_token.inner, progress) - .await?; - let _ = sink.add(RsHashFileEvent::Done { hash }); - Ok(()) + // Progress events with throttling + let last_emit = std::cell::Cell::new(None::); + let progress = { + let sink = sink.clone(); + move |hashed| { + let now = std::time::Instant::now(); + if let Some(last) = last_emit.get() { + if now.duration_since(last) < std::time::Duration::from_millis(20) { + return; + } + } + last_emit.set(Some(now)); + let _ = sink.add(RsHashFileEvent::Progress { bytes: hashed }); + } + }; + + let hash = + localsend::crypto::hash::sha256_file_content(content, &cancel_token.inner, progress) + .await?; + let _ = sink.add(RsHashFileEvent::Done { hash }); + Ok(()) + } + .await; + + if let Err(err) = result { + let _ = sink.add_error(err); + } } diff --git a/packages/localsend_isolates/rust/src/api/http.rs b/packages/localsend_isolates/rust/src/api/http.rs index 889edc6d..6d5e617d 100644 --- a/packages/localsend_isolates/rust/src/api/http.rs +++ b/packages/localsend_isolates/rust/src/api/http.rs @@ -87,9 +87,15 @@ impl RsHttpClient { Ok(response) } + /// Uploads a single file, emitting [RsUploadEvent]s on [sink]. + /// + /// Failures are emitted as [RsUploadEvent::Failed] instead of being + /// returned: flutter_rust_bridge discards the returned `Result` of + /// functions taking a [StreamSink], so a returned error would become an + /// uncaught async error killing the calling isolate. pub async fn upload( &self, - sink: StreamSink, + sink: StreamSink, protocol: ProtocolType, ip: &str, port: u16, @@ -102,45 +108,53 @@ impl RsHttpClient { file_descriptor: Option, content_length: u64, cancel_token: &RsCancellationToken, - ) -> Result<(), RsHttpClientError> { - let content = resolve_file_content(binary, path, file_descriptor)?; - let last_emit = std::cell::Cell::new(None::); - let progress = move |sent| { - let now = std::time::Instant::now(); - let is_final = sent >= content_length; - if !is_final { - if let Some(last) = last_emit.get() { - if now.duration_since(last) < std::time::Duration::from_millis(20) { - return; + ) { + let result = async { + let content = resolve_file_content(binary, path, file_descriptor)?; + let last_emit = std::cell::Cell::new(None::); + let progress_sink = sink.clone(); + let progress = move |sent| { + let now = std::time::Instant::now(); + let is_final = sent >= content_length; + if !is_final { + if let Some(last) = last_emit.get() { + if now.duration_since(last) < std::time::Duration::from_millis(20) { + return; + } } } - } - last_emit.set(Some(now)); - let progress = if content_length == 0 { - 1.0 - } else { - (sent as f64 / content_length as f64).min(1.0) + last_emit.set(Some(now)); + let progress = if content_length == 0 { + 1.0 + } else { + (sent as f64 / content_length as f64).min(1.0) + }; + let _ = progress_sink.add(RsUploadEvent::Progress { progress }); }; - let _ = sink.add(progress); - }; - self.inner - .upload( - protocol, - ip, - port, - public_key, - session_id, - file_id, - token, - content, - progress, - cancel_token.inner.clone(), - ) - .await - .map_err(RsHttpClientError::from)?; + self.inner + .upload( + protocol, + ip, + port, + public_key, + session_id, + file_id, + token, + content, + progress, + cancel_token.inner.clone(), + ) + .await + .map_err(RsHttpClientError::from)?; - Ok(()) + Ok(()) + } + .await; + + if let Err(error) = result { + let _ = sink.add(RsUploadEvent::Failed { error }); + } } pub async fn cancel( @@ -188,6 +202,17 @@ fn resolve_file_content( } } +/// An event emitted while a file is being uploaded by [RsHttpClient::upload]. +#[derive(Clone)] +pub enum RsUploadEvent { + /// The upload progress as a fraction (0.0 to 1.0). Throttled. + Progress { progress: f64 }, + + /// The upload failed. Always the last event of the stream. + Failed { error: RsHttpClientError }, +} + +#[derive(Clone)] pub enum RsHttpClientError { StatusCode { status: u16, diff --git a/packages/localsend_isolates/rust/src/api/server.rs b/packages/localsend_isolates/rust/src/api/server.rs index 80b804dc..72abd74b 100644 --- a/packages/localsend_isolates/rust/src/api/server.rs +++ b/packages/localsend_isolates/rust/src/api/server.rs @@ -486,7 +486,10 @@ impl RsHttpServer { /// and waits until the file has been received completely. /// /// The progress (fraction of [file_size]) is emitted on [sink] - /// while the file is being received. + /// while the file is being received. Failures are emitted on [sink] as + /// well: flutter_rust_bridge discards the returned `Result` of functions + /// taking a [StreamSink], so a returned error would become an uncaught + /// async error killing the calling isolate. /// /// Timestamps provided in the sender's file metadata are applied to the /// written file by the server. @@ -498,50 +501,58 @@ impl RsHttpServer { path: Option, file_descriptor: Option, file_size: u64, - ) -> anyhow::Result<()> { - let Some(target_tx) = self - .pending_uploads - .lock() - .await - .remove(&(session_id, file_id)) - else { - return Err(anyhow::anyhow!("No pending file upload for this file")); - }; + ) { + let result = async { + let Some(target_tx) = self + .pending_uploads + .lock() + .await + .remove(&(session_id, file_id)) + else { + return Err(anyhow::anyhow!("No pending file upload for this file")); + }; - let (progress_tx, mut progress_rx) = mpsc::channel::(16); - tokio::spawn(async move { - let mut last_emit = None::; - while let Some(written) = progress_rx.recv().await { - let now = std::time::Instant::now(); - let is_final = written >= file_size; - if !is_final { - if let Some(last) = last_emit { - if now.duration_since(last) < std::time::Duration::from_millis(20) { - continue; + let (progress_tx, mut progress_rx) = mpsc::channel::(16); + let progress_sink = sink.clone(); + tokio::spawn(async move { + let mut last_emit = None::; + while let Some(written) = progress_rx.recv().await { + let now = std::time::Instant::now(); + let is_final = written >= file_size; + if !is_final { + if let Some(last) = last_emit { + if now.duration_since(last) < std::time::Duration::from_millis(20) { + continue; + } } } + last_emit = Some(now); + let progress = if file_size == 0 { + 1.0 + } else { + (written as f64 / file_size as f64).min(1.0) + }; + let _ = progress_sink.add(progress); } - last_emit = Some(now); - let progress = if file_size == 0 { - 1.0 - } else { - (written as f64 / file_size as f64).min(1.0) - }; - let _ = sink.add(progress); + }); + + let (result_tx, result_rx) = oneshot::channel::>(); + let target = resolve_upload_target(path, file_descriptor, result_tx, progress_tx)?; + + target_tx + .send(target) + .map_err(|_| anyhow::anyhow!("Upload request already ended"))?; + + match result_rx.await { + Ok(Ok(())) => Ok(()), + Ok(Err(err)) => Err(anyhow::anyhow!(err)), + Err(_) => Err(anyhow::anyhow!("Upload request aborted")), } - }); + } + .await; - let (result_tx, result_rx) = oneshot::channel::>(); - let target = resolve_upload_target(path, file_descriptor, result_tx, progress_tx)?; - - target_tx - .send(target) - .map_err(|_| anyhow::anyhow!("Upload request already ended"))?; - - match result_rx.await { - Ok(Ok(())) => Ok(()), - Ok(Err(err)) => Err(anyhow::anyhow!(err)), - Err(_) => Err(anyhow::anyhow!("Upload request aborted")), + if let Err(err) = result { + let _ = sink.add_error(err); } } diff --git a/packages/localsend_isolates/rust/src/frb_generated.rs b/packages/localsend_isolates/rust/src/frb_generated.rs index d3310369..522e3657 100644 --- a/packages/localsend_isolates/rust/src/frb_generated.rs +++ b/packages/localsend_isolates/rust/src/frb_generated.rs @@ -1128,10 +1128,10 @@ fn wire__crate__api__http__RsHttpClient_upload_impl( let api_that = , >>::sse_decode(&mut deserializer); - let api_sink = - >::sse_decode( - &mut deserializer, - ); + let api_sink = >::sse_decode(&mut deserializer); let api_protocol = ::sse_decode(&mut deserializer); let api_ip = ::sse_decode(&mut deserializer); let api_port = ::sse_decode(&mut deserializer); @@ -1148,7 +1148,7 @@ fn wire__crate__api__http__RsHttpClient_upload_impl( >>::sse_decode(&mut deserializer); deserializer.end(); move |context| async move { - transform_result_sse::<_, crate::api::http::RsHttpClientError>( + transform_result_sse::<_, ()>( (move || async move { let mut api_that_guard = None; let mut api_cancel_token_guard = None; @@ -1180,23 +1180,25 @@ fn wire__crate__api__http__RsHttpClient_upload_impl( } let api_that_guard = api_that_guard.unwrap(); let api_cancel_token_guard = api_cancel_token_guard.unwrap(); - let output_ok = crate::api::http::RsHttpClient::upload( - &*api_that_guard, - api_sink, - api_protocol, - &api_ip, - api_port, - api_public_key, - &api_session_id, - &api_file_id, - &api_token, - api_binary, - api_path, - api_file_descriptor, - api_content_length, - &*api_cancel_token_guard, - ) - .await?; + let output_ok = Result::<_, ()>::Ok({ + crate::api::http::RsHttpClient::upload( + &*api_that_guard, + api_sink, + api_protocol, + &api_ip, + api_port, + api_public_key, + &api_session_id, + &api_file_id, + &api_token, + api_binary, + api_path, + api_file_descriptor, + api_content_length, + &*api_cancel_token_guard, + ) + .await; + })?; Ok(output_ok) })() .await, @@ -1559,7 +1561,7 @@ fn wire__crate__api__server__RsHttpServer_respond_file_upload_impl( let api_file_size = ::sse_decode(&mut deserializer); deserializer.end(); move |context| async move { - transform_result_sse::<_, flutter_rust_bridge::for_generated::anyhow::Error>( + transform_result_sse::<_, ()>( (move || async move { let mut api_that_guard = None; let decode_indices_ = @@ -1578,16 +1580,18 @@ fn wire__crate__api__server__RsHttpServer_respond_file_upload_impl( } } let api_that_guard = api_that_guard.unwrap(); - let output_ok = crate::api::server::RsHttpServer::respond_file_upload( - &*api_that_guard, - api_sink, - api_session_id, - api_file_id, - api_path, - api_file_descriptor, - api_file_size, - ) - .await?; + let output_ok = Result::<_, ()>::Ok({ + crate::api::server::RsHttpServer::respond_file_upload( + &*api_that_guard, + api_sink, + api_session_id, + api_file_id, + api_path, + api_file_descriptor, + api_file_size, + ) + .await; + })?; Ok(output_ok) })() .await, @@ -3005,7 +3009,7 @@ fn wire__crate__api__crypto__hash_file_impl( >>::sse_decode(&mut deserializer); deserializer.end(); move |context| async move { - transform_result_sse::<_, flutter_rust_bridge::for_generated::anyhow::Error>( + transform_result_sse::<_, ()>( (move || async move { let mut api_cancel_token_guard = None; let decode_indices_ = @@ -3026,14 +3030,16 @@ fn wire__crate__api__crypto__hash_file_impl( } } let api_cancel_token_guard = api_cancel_token_guard.unwrap(); - let output_ok = crate::api::crypto::hash_file( - api_sink, - api_path, - api_file_descriptor, - api_bytes, - &*api_cancel_token_guard, - ) - .await?; + let output_ok = Result::<_, ()>::Ok({ + crate::api::crypto::hash_file( + api_sink, + api_path, + api_file_descriptor, + api_bytes, + &*api_cancel_token_guard, + ) + .await; + })?; Ok(output_ok) })() .await, @@ -3822,6 +3828,16 @@ impl SseDecode } } +impl SseDecode + for StreamSink +{ + // Codec=Sse (Serialization based), see doc to use other codecs + fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { + let mut inner = ::sse_decode(deserializer); + return StreamSink::deserialize(inner); + } +} + impl SseDecode for StreamSink { @@ -4683,6 +4699,28 @@ impl SseDecode for crate::api::discovery::RsStoredDevice { } } +impl SseDecode for crate::api::http::RsUploadEvent { + // Codec=Sse (Serialization based), see doc to use other codecs + fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { + let mut tag_ = ::sse_decode(deserializer); + match tag_ { + 0 => { + let mut var_progress = ::sse_decode(deserializer); + return crate::api::http::RsUploadEvent::Progress { + progress: var_progress, + }; + } + 1 => { + let mut var_error = ::sse_decode(deserializer); + return crate::api::http::RsUploadEvent::Failed { error: var_error }; + } + _ => { + unimplemented!(""); + } + } + } +} + impl SseDecode for crate::api::webrtc::RTCFileError { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { @@ -6030,6 +6068,33 @@ impl flutter_rust_bridge::IntoIntoDart } } // Codec=Dco (DartCObject based), see doc to use other codecs +impl flutter_rust_bridge::IntoDart for crate::api::http::RsUploadEvent { + fn into_dart(self) -> flutter_rust_bridge::for_generated::DartAbi { + match self { + crate::api::http::RsUploadEvent::Progress { progress } => { + [0.into_dart(), progress.into_into_dart().into_dart()].into_dart() + } + crate::api::http::RsUploadEvent::Failed { error } => { + [1.into_dart(), error.into_into_dart().into_dart()].into_dart() + } + _ => { + unimplemented!(""); + } + } + } +} +impl flutter_rust_bridge::for_generated::IntoDartExceptPrimitive + for crate::api::http::RsUploadEvent +{ +} +impl flutter_rust_bridge::IntoIntoDart + for crate::api::http::RsUploadEvent +{ + fn into_into_dart(self) -> crate::api::http::RsUploadEvent { + self + } +} +// Codec=Dco (DartCObject based), see doc to use other codecs impl flutter_rust_bridge::IntoDart for FrbWrapper { fn into_dart(self) -> flutter_rust_bridge::for_generated::DartAbi { [ @@ -6614,6 +6679,15 @@ impl SseEncode } } +impl SseEncode + for StreamSink +{ + // Codec=Sse (Serialization based), see doc to use other codecs + fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { + unimplemented!("") + } +} + impl SseEncode for StreamSink { @@ -7307,6 +7381,25 @@ impl SseEncode for crate::api::discovery::RsStoredDevice { } } +impl SseEncode for crate::api::http::RsUploadEvent { + // Codec=Sse (Serialization based), see doc to use other codecs + fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { + match self { + crate::api::http::RsUploadEvent::Progress { progress } => { + ::sse_encode(0, serializer); + ::sse_encode(progress, serializer); + } + crate::api::http::RsUploadEvent::Failed { error } => { + ::sse_encode(1, serializer); + ::sse_encode(error, serializer); + } + _ => { + unimplemented!(""); + } + } + } +} + impl SseEncode for crate::api::webrtc::RTCFileError { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { diff --git a/packages/localsend_isolates/test/task/server/cancel_session_repro_test.dart b/packages/localsend_isolates/test/task/server/cancel_session_repro_test.dart new file mode 100644 index 00000000..3ac0d9ab --- /dev/null +++ b/packages/localsend_isolates/test/task/server/cancel_session_repro_test.dart @@ -0,0 +1,239 @@ +// Regression test: the receiver cancels the session mid-transfer (the sender +// aborts its in-flight uploads), then the sender starts a new session. +// +// The aborted uploads used to fail `respond_file_upload` with a returned +// `Err`, which flutter_rust_bridge discards as an uncaught async error. That +// killed the server isolate, leaving the Rust server without a Dart listener, +// so every following prepare-upload request was answered with 500. +// +// Loads the real Rust dylib and replicates the event loop of +// `server_isolate.dart` without isolates. Skipped when the dylib is missing; +// build it with `cargo build -p rust_lib_localsend_app` from the repo root. +@Timeout(Duration(minutes: 2)) +library; + +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_rust_bridge/flutter_rust_bridge_for_generated.dart' show ExternalLibrary; +import 'package:flutter_test/flutter_test.dart'; +import 'package:localsend_isolates/rust/api/server.dart'; +import 'package:localsend_isolates/rust/frb_generated.dart'; +import 'package:localsend_isolates/util/future_queue.dart'; + +const _port = 40901; + +class _ReceiveSession { + final String sessionId; + final Set acceptedIds; + final Map uploads = {}; + + _ReceiveSession(this.sessionId, this.acceptedIds); +} + +void main() { + test('new session works after receiver-side cancel during transfer', () async { + final dylib = File('${Directory.current.path}/../../target/debug/librust_lib_localsend_app.dylib'); + if (!dylib.existsSync()) { + markTestSkipped('Rust dylib not built (cargo build -p rust_lib_localsend_app)'); + return; + } + await RustLib.init(externalLibrary: ExternalLibrary.open(dylib.path)); + + final tempDir = await Directory.systemTemp.createTemp('cancel_repro'); + + final server = await startServer( + port: _port, + tls: null, + alias: 'Receiver', + version: '2.1', + deviceModel: 'Test', + deviceType: null, + fingerprint: 'RECEIVER-FINGERPRINT', + pin: null, + verifyChecksums: true, + web: null, + showToken: null, + ); + + _ReceiveSession? session; + final uploadErrors = []; + var receiverCancelled = false; + + // Set when the receiver cancels; the "sender" reacts by aborting everything. + final senderAbort = Completer(); + void triggerReceiverCancel(String sessionId) { + if (receiverCancelled) return; + receiverCancelled = true; + session = null; + unawaited(server.cancelSession(sessionId: sessionId)); + senderAbort.complete(); + } + + // Mirrors the event loop in server_isolate.dart. + final eventLoop = () async { + await for (final event in server.listen()) { + switch (event) { + case RsServerEvent_PrepareUpload(:final sessionId, :final files): + session = null; + // The app answers via a separate task; a microtask gap is enough here. + unawaited( + Future(() async { + session = _ReceiveSession(sessionId, files.keys.toSet()); + await server.respondPrepareUpload(acceptedFileIds: files.keys.toList()); + }), + ); + case RsServerEvent_FileUpload(:final sessionId, :final fileId, :final file): + final s = session; + if (s == null || s.sessionId != sessionId || !s.acceptedIds.contains(fileId)) { + // Same as server_isolate.dart: reject by cancelling. + unawaited(server.cancelSession(sessionId: sessionId)); + break; + } + final queue = s.uploads.putIfAbsent(fileId, () => FutureQueue()); + queue.add(() async { + try { + final progressStream = server.respondFileUpload( + sessionId: sessionId, + fileId: fileId, + path: '${tempDir.path}/$fileId', + fileDescriptor: null, + fileSize: file.size, + ); + await for (final progress in progressStream) { + if (progress > 0.2) { + // Receiver-side cancel mid-transfer (HttpServerCancelSessionTask). + triggerReceiverCancel(sessionId); + } + } + } catch (e) { + // Mirrors _handleFileUpload: the failure surfaces on the + // progress stream and the file is marked as failed. + uploadErrors.add(e); + } + }); + default: + break; + } + } + }(); + + // --- Sender side (plain HTTP) --- + final client = HttpClient(); + + Future<(int, String)> prepareUpload(List fileIds, {required int size}) async { + final req = await client.postUrl(Uri.parse('http://127.0.0.1:$_port/api/localsend/v2/prepare-upload')); + req.headers.contentType = ContentType.json; + req.write( + jsonEncode({ + 'info': { + 'alias': 'Sender', + 'version': '2.1', + 'fingerprint': 'SENDER-FINGERPRINT', + 'port': 1, + 'protocol': 'http', + 'download': false, + }, + 'files': { + for (final id in fileIds) + id: { + 'id': id, + 'fileName': '$id.bin', + 'size': size, + 'fileType': 'application/octet-stream', + }, + }, + }), + ); + final res = await req.close(); + final body = await utf8.decodeStream(res); + return (res.statusCode, body); + } + + // Uploads with pauses between chunks; aborts the request when + // [senderAbort] completes (like the sender's cancellation token). + Future upload(String sessionId, String fileId, String token, int size, {bool abortable = false}) async { + final req = await client.postUrl( + Uri.parse('http://127.0.0.1:$_port/api/localsend/v2/upload?sessionId=$sessionId&fileId=$fileId&token=$token'), + ); + req.headers.contentLength = size; + var aborted = false; + if (abortable) { + unawaited( + senderAbort.future.then((_) { + aborted = true; + req.abort(); + }), + ); + } + try { + const chunk = 64 * 1024; + var sent = 0; + while (sent < size && !aborted) { + final n = (size - sent).clamp(0, chunk); + req.add(List.filled(n, 7)); + await req.flush(); + sent += n; + if (abortable) { + await Future.delayed(const Duration(milliseconds: 20)); + } + } + final res = await req.close(); + await res.drain(); + return res.statusCode; + } catch (_) { + // Aborted by the "cancellation token". + return -1; + } + } + + // Session A: several files, 2 uploaded in parallel like the upload isolate, + // cancelled by the receiver mid-transfer; the sender then aborts everything. + const size = 4 * 1024 * 1024; + final fileIds = List.generate(4, (i) => 'file-a$i'); + final (statusA, bodyA) = await prepareUpload(fileIds, size: size); + expect(statusA, 200); + final resA = jsonDecode(bodyA) as Map; + final sessionA = resA['sessionId'] as String; + final tokensA = (resA['files'] as Map).cast(); + + final pending = [...fileIds]; + Future worker() async { + while (pending.isNotEmpty && !senderAbort.isCompleted) { + final id = pending.removeAt(0); + await upload(sessionA, id, tokensA[id]!, size, abortable: true); + } + } + + await Future.wait([worker(), worker()]); + expect(receiverCancelled, true, reason: 'the transfer should have been cancelled mid-flight'); + + // Give in-flight events time to settle. + await Future.delayed(const Duration(milliseconds: 500)); + + // The aborted uploads must fail via the progress stream + // (not as uncaught async errors, which would kill the server isolate). + expect(uploadErrors, isNotEmpty); + + // Session B: this returned 500 before the fix. + final (statusB, bodyB) = await prepareUpload(['file-b'], size: 1024).timeout( + const Duration(seconds: 10), + onTimeout: () => (-1, 'timed out'), + ); + expect(statusB, 200, reason: 'prepare-upload after cancel responded: $bodyB'); + + final resB = jsonDecode(bodyB) as Map; + final sessionB = resB['sessionId'] as String; + final tokenB = (resB['files'] as Map)['file-b'] as String; + final uploadStatusB = await upload(sessionB, 'file-b', tokenB, 1024); + expect(uploadStatusB, 200); + + client.close(force: true); + await server.stop(); + // The event stream does not close on stop (the server handle keeps the + // event channel alive), matching the app which never awaits it either. + unawaited(eventLoop); + await tempDir.delete(recursive: true); + }); +}