diff --git a/apps/chanora_flutter/lib/l10n/app_en.arb b/apps/chanora_flutter/lib/l10n/app_en.arb index e798a56..a4e8f6f 100644 --- a/apps/chanora_flutter/lib/l10n/app_en.arb +++ b/apps/chanora_flutter/lib/l10n/app_en.arb @@ -27,6 +27,18 @@ "placeholders": { "message": { "type": "String" } } }, + "statusReconnecting": "Reconnecting… attempt {attempt}, in {delay}s", + "@statusReconnecting": { + "placeholders": { + "attempt": { "type": "int" }, + "delay": { "type": "int" } + } + }, + "statusConnectionLost": "Connection lost: {reason}", + "@statusConnectionLost": { + "placeholders": { "reason": { "type": "String" } } + }, + "audioStatsLine": "TX {sent} frames • RX {received} frames • PTT {ptt}", "@audioStatsLine": { "placeholders": { diff --git a/apps/chanora_flutter/lib/l10n/app_zh.arb b/apps/chanora_flutter/lib/l10n/app_zh.arb index 3e62f83..967326a 100644 --- a/apps/chanora_flutter/lib/l10n/app_zh.arb +++ b/apps/chanora_flutter/lib/l10n/app_zh.arb @@ -20,6 +20,9 @@ "statusConnected": "已连接到 {server}", "statusError": "错误：{message}", + "statusReconnecting": "正在重新连接……第 {attempt} 次尝试，{delay} 秒后", + "statusConnectionLost": "连接已断开：{reason}", + "audioStatsLine": "发送 {sent} 帧 • 接收 {received} 帧 • PTT {ptt}", "channelsHeading": "频道", diff --git a/apps/chanora_flutter/lib/l10n/generated/app_localizations.dart b/apps/chanora_flutter/lib/l10n/generated/app_localizations.dart index a8024a4..d0fe1cb 100644 --- a/apps/chanora_flutter/lib/l10n/generated/app_localizations.dart +++ b/apps/chanora_flutter/lib/l10n/generated/app_localizations.dart @@ -181,6 +181,18 @@ abstract class AppL10n { /// **'Error: {message}'** String statusError(String message); + /// No description provided for @statusReconnecting. + /// + /// In en, this message translates to: + /// **'Reconnecting… attempt {attempt}, in {delay}s'** + String statusReconnecting(int attempt, int delay); + + /// No description provided for @statusConnectionLost. + /// + /// In en, this message translates to: + /// **'Connection lost: {reason}'** + String statusConnectionLost(String reason); + /// No description provided for @audioStatsLine. /// /// In en, this message translates to: diff --git a/apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart b/apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart index b41fbda..e2640ad 100644 --- a/apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart +++ b/apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart @@ -55,6 +55,16 @@ class AppL10nEn extends AppL10n { return 'Error: $message'; } + @override + String statusReconnecting(int attempt, int delay) { + return 'Reconnecting… attempt $attempt, in ${delay}s'; + } + + @override + String statusConnectionLost(String reason) { + return 'Connection lost: $reason'; + } + @override String audioStatsLine(int sent, int received, String ptt) { return 'TX $sent frames • RX $received frames • PTT $ptt'; diff --git a/apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart b/apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart index 198b64b..6983032 100644 --- a/apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart +++ b/apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart @@ -54,6 +54,16 @@ class AppL10nZh extends AppL10n { return '错误：$message'; } + @override + String statusReconnecting(int attempt, int delay) { + return '正在重新连接……第 $attempt 次尝试，$delay 秒后'; + } + + @override + String statusConnectionLost(String reason) { + return '连接已断开：$reason'; + } + @override String audioStatsLine(int sent, int received, String ptt) { return '发送 $sent 帧 • 接收 $received 帧 • PTT $ptt'; diff --git a/apps/chanora_flutter/lib/main.dart b/apps/chanora_flutter/lib/main.dart index c1f9045..1f0d865 100644 --- a/apps/chanora_flutter/lib/main.dart +++ b/apps/chanora_flutter/lib/main.dart @@ -60,9 +60,57 @@ class _BetaHomeState extends State<_BetaHome> { bool _audioStarted = false; rust.BridgeAudioStats? _audioStats; Timer? _statsTimer; + StreamSubscription? _eventsSub; + + // A.6 reconnect banner state. + String? _lostReason; + int? _reconnectAttempt; + int? _reconnectDelay; + + @override + void initState() { + super.initState(); + _eventsSub = rust.eventsStream().listen(_onEvent); + } + + void _onEvent(rust.BridgeEvent evt) { + if (!mounted) return; + switch (evt) { + case rust.BridgeEvent_Connected(): + setState(() { + _phase = _Phase.connected; + _lostReason = null; + _reconnectAttempt = null; + _reconnectDelay = null; + }); + case rust.BridgeEvent_Lost(:final reason): + setState(() { + _lostReason = reason; + _reconnectAttempt = null; + _reconnectDelay = null; + }); + case rust.BridgeEvent_Reconnecting(:final attempt, :final delaySecs): + setState(() { + _reconnectAttempt = attempt; + _reconnectDelay = delaySecs; + }); + case rust.BridgeEvent_Disconnected(): + setState(() { + _phase = _Phase.idle; + _lostReason = null; + _reconnectAttempt = null; + _reconnectDelay = null; + }); + case rust.BridgeEvent_AudioStarted(): + setState(() => _audioStarted = true); + case rust.BridgeEvent_AudioStopped(): + setState(() => _audioStarted = false); + } + } @override void dispose() { + _eventsSub?.cancel(); _statsTimer?.cancel(); _hostCtl.dispose(); _nickCtl.dispose(); @@ -201,6 +249,42 @@ class _BetaHomeState extends State<_BetaHome> { ), const SizedBox(height: 12), Text(statusText(), style: theme.textTheme.titleMedium), + if (_lostReason != null || _reconnectAttempt != null) ...[ + const SizedBox(height: 8), + Container( + padding: const EdgeInsets.all(10), + decoration: BoxDecoration( + color: theme.colorScheme.errorContainer, + borderRadius: BorderRadius.circular(8), + ), + child: Row( + children: [ + SizedBox( + width: 16, + height: 16, + child: CircularProgressIndicator( + strokeWidth: 2, + color: theme.colorScheme.onErrorContainer, + ), + ), + const SizedBox(width: 10), + Expanded( + child: Text( + _reconnectAttempt != null + ? l10n.statusReconnecting( + _reconnectAttempt!, + _reconnectDelay ?? 0, + ) + : l10n.statusConnectionLost(_lostReason ?? ''), + style: TextStyle( + color: theme.colorScheme.onErrorContainer, + ), + ), + ), + ], + ), + ), + ], const SizedBox(height: 12), if (_phase == _Phase.idle) ...[ _ConnectForm( diff --git a/apps/chanora_flutter/lib/src/rust/api.dart b/apps/chanora_flutter/lib/src/rust/api.dart index 2bb8723..beb4595 100644 --- a/apps/chanora_flutter/lib/src/rust/api.dart +++ b/apps/chanora_flutter/lib/src/rust/api.dart @@ -6,9 +6,11 @@ import 'frb_generated.dart'; import 'lib.dart'; import 'package:flutter_rust_bridge/flutter_rust_bridge_for_generated.dart'; +import 'package:freezed_annotation/freezed_annotation.dart' hide protected; +part 'api.freezed.dart'; // These functions are ignored because they are not marked as `pub`: `runtime`, `session` -// 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`, `clone`, `clone`, `fmt`, `fmt`, `fmt`, `fmt`, `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`, `clone`, `clone`, `clone`, `fmt`, `fmt`, `fmt`, `fmt`, `fmt`, `from`, `from` /// Connect to a TeamSpeak-compatible server and return the initial /// state snapshot. Honours the DEC-006 single-connection invariant @@ -35,6 +37,13 @@ Future startAudio() => RustLib.instance.api.crateApiStartAudio(); Future setPtt({required bool active}) => RustLib.instance.api.crateApiSetPtt(active: active); +/// Subscribe to lifecycle events. Each call yields a fresh +/// subscription; multiple subscribers are supported. On slow +/// consumers, events are dropped rather than blocking the supervisor +/// (consistent with `tokio::sync::broadcast::Receiver` semantics). +Stream eventsStream() => + RustLib.instance.api.crateApiEventsStream(); + /// Read audio statistics. Errors if no connection or audio not started. Future audioStats() => RustLib.instance.api.crateApiAudioStats(); @@ -138,6 +147,44 @@ class BridgeClient { name == other.name; } +@freezed +sealed class BridgeEvent with _$BridgeEvent { + const BridgeEvent._(); + + /// Initial connect succeeded, or a reconnect attempt succeeded. + const factory BridgeEvent.connected({ + /// Server name reported by the server snapshot. + required String serverName, + }) = BridgeEvent_Connected; + + /// Connection lost; supervisor will retry. + const factory BridgeEvent.lost({ + /// Reason classification from the protocol layer. + required String reason, + }) = BridgeEvent_Lost; + + /// Supervisor is sleeping before its next reconnect attempt. + const factory BridgeEvent.reconnecting({ + /// 1-based attempt counter for the current outage. + required int attempt, + + /// Seconds the supervisor will sleep before this attempt. + required int delaySecs, + }) = BridgeEvent_Reconnecting; + + /// Session ended (user-requested disconnect or unrecoverable). + const factory BridgeEvent.disconnected({ + /// Reason classification. + required String reason, + }) = BridgeEvent_Disconnected; + + /// Audio engine started. + const factory BridgeEvent.audioStarted() = BridgeEvent_AudioStarted; + + /// Audio engine stopped. + const factory BridgeEvent.audioStopped() = BridgeEvent_AudioStopped; +} + /// Server snapshot as seen by Dart. class BridgeSnapshot { /// Server name. diff --git a/apps/chanora_flutter/lib/src/rust/api.freezed.dart b/apps/chanora_flutter/lib/src/rust/api.freezed.dart new file mode 100644 index 0000000..346e687 --- /dev/null +++ b/apps/chanora_flutter/lib/src/rust/api.freezed.dart @@ -0,0 +1,533 @@ +// GENERATED CODE - DO NOT MODIFY BY HAND +// coverage:ignore-file +// ignore_for_file: type=lint +// ignore_for_file: unused_element, deprecated_member_use, deprecated_member_use_from_same_package, use_function_type_syntax_for_parameters, unnecessary_const, avoid_init_to_null, invalid_override_different_default_values_named, prefer_expression_function_bodies, annotate_overrides, invalid_annotation_target, unnecessary_question_mark + +part of 'api.dart'; + +// ************************************************************************** +// FreezedGenerator +// ************************************************************************** + +// dart format off +T _$identity(T value) => value; +/// @nodoc +mixin _$BridgeEvent { + + + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is BridgeEvent); +} + + +@override +int get hashCode => runtimeType.hashCode; + +@override +String toString() { + return 'BridgeEvent()'; +} + + +} + +/// @nodoc +class $BridgeEventCopyWith<$Res> { +$BridgeEventCopyWith(BridgeEvent _, $Res Function(BridgeEvent) __); +} + + +/// Adds pattern-matching-related methods to [BridgeEvent]. +extension BridgeEventPatterns on BridgeEvent { +/// 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( BridgeEvent_Connected value)? connected,TResult Function( BridgeEvent_Lost value)? lost,TResult Function( BridgeEvent_Reconnecting value)? reconnecting,TResult Function( BridgeEvent_Disconnected value)? disconnected,TResult Function( BridgeEvent_AudioStarted value)? audioStarted,TResult Function( BridgeEvent_AudioStopped value)? audioStopped,required TResult orElse(),}){ +final _that = this; +switch (_that) { +case BridgeEvent_Connected() when connected != null: +return connected(_that);case BridgeEvent_Lost() when lost != null: +return lost(_that);case BridgeEvent_Reconnecting() when reconnecting != null: +return reconnecting(_that);case BridgeEvent_Disconnected() when disconnected != null: +return disconnected(_that);case BridgeEvent_AudioStarted() when audioStarted != null: +return audioStarted(_that);case BridgeEvent_AudioStopped() when audioStopped != null: +return audioStopped(_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( BridgeEvent_Connected value) connected,required TResult Function( BridgeEvent_Lost value) lost,required TResult Function( BridgeEvent_Reconnecting value) reconnecting,required TResult Function( BridgeEvent_Disconnected value) disconnected,required TResult Function( BridgeEvent_AudioStarted value) audioStarted,required TResult Function( BridgeEvent_AudioStopped value) audioStopped,}){ +final _that = this; +switch (_that) { +case BridgeEvent_Connected(): +return connected(_that);case BridgeEvent_Lost(): +return lost(_that);case BridgeEvent_Reconnecting(): +return reconnecting(_that);case BridgeEvent_Disconnected(): +return disconnected(_that);case BridgeEvent_AudioStarted(): +return audioStarted(_that);case BridgeEvent_AudioStopped(): +return audioStopped(_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( BridgeEvent_Connected value)? connected,TResult? Function( BridgeEvent_Lost value)? lost,TResult? Function( BridgeEvent_Reconnecting value)? reconnecting,TResult? Function( BridgeEvent_Disconnected value)? disconnected,TResult? Function( BridgeEvent_AudioStarted value)? audioStarted,TResult? Function( BridgeEvent_AudioStopped value)? audioStopped,}){ +final _that = this; +switch (_that) { +case BridgeEvent_Connected() when connected != null: +return connected(_that);case BridgeEvent_Lost() when lost != null: +return lost(_that);case BridgeEvent_Reconnecting() when reconnecting != null: +return reconnecting(_that);case BridgeEvent_Disconnected() when disconnected != null: +return disconnected(_that);case BridgeEvent_AudioStarted() when audioStarted != null: +return audioStarted(_that);case BridgeEvent_AudioStopped() when audioStopped != null: +return audioStopped(_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( String serverName)? connected,TResult Function( String reason)? lost,TResult Function( int attempt, int delaySecs)? reconnecting,TResult Function( String reason)? disconnected,TResult Function()? audioStarted,TResult Function()? audioStopped,required TResult orElse(),}) {final _that = this; +switch (_that) { +case BridgeEvent_Connected() when connected != null: +return connected(_that.serverName);case BridgeEvent_Lost() when lost != null: +return lost(_that.reason);case BridgeEvent_Reconnecting() when reconnecting != null: +return reconnecting(_that.attempt,_that.delaySecs);case BridgeEvent_Disconnected() when disconnected != null: +return disconnected(_that.reason);case BridgeEvent_AudioStarted() when audioStarted != null: +return audioStarted();case BridgeEvent_AudioStopped() when audioStopped != null: +return audioStopped();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( String serverName) connected,required TResult Function( String reason) lost,required TResult Function( int attempt, int delaySecs) reconnecting,required TResult Function( String reason) disconnected,required TResult Function() audioStarted,required TResult Function() audioStopped,}) {final _that = this; +switch (_that) { +case BridgeEvent_Connected(): +return connected(_that.serverName);case BridgeEvent_Lost(): +return lost(_that.reason);case BridgeEvent_Reconnecting(): +return reconnecting(_that.attempt,_that.delaySecs);case BridgeEvent_Disconnected(): +return disconnected(_that.reason);case BridgeEvent_AudioStarted(): +return audioStarted();case BridgeEvent_AudioStopped(): +return audioStopped();} +} +/// 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( String serverName)? connected,TResult? Function( String reason)? lost,TResult? Function( int attempt, int delaySecs)? reconnecting,TResult? Function( String reason)? disconnected,TResult? Function()? audioStarted,TResult? Function()? audioStopped,}) {final _that = this; +switch (_that) { +case BridgeEvent_Connected() when connected != null: +return connected(_that.serverName);case BridgeEvent_Lost() when lost != null: +return lost(_that.reason);case BridgeEvent_Reconnecting() when reconnecting != null: +return reconnecting(_that.attempt,_that.delaySecs);case BridgeEvent_Disconnected() when disconnected != null: +return disconnected(_that.reason);case BridgeEvent_AudioStarted() when audioStarted != null: +return audioStarted();case BridgeEvent_AudioStopped() when audioStopped != null: +return audioStopped();case _: + return null; + +} +} + +} + +/// @nodoc + + +class BridgeEvent_Connected extends BridgeEvent { + const BridgeEvent_Connected({required this.serverName}): super._(); + + +/// Server name reported by the server snapshot. + final String serverName; + +/// Create a copy of BridgeEvent +/// with the given fields replaced by the non-null parameter values. +@JsonKey(includeFromJson: false, includeToJson: false) +@pragma('vm:prefer-inline') +$BridgeEvent_ConnectedCopyWith get copyWith => _$BridgeEvent_ConnectedCopyWithImpl(this, _$identity); + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is BridgeEvent_Connected&&(identical(other.serverName, serverName) || other.serverName == serverName)); +} + + +@override +int get hashCode => Object.hash(runtimeType,serverName); + +@override +String toString() { + return 'BridgeEvent.connected(serverName: $serverName)'; +} + + +} + +/// @nodoc +abstract mixin class $BridgeEvent_ConnectedCopyWith<$Res> implements $BridgeEventCopyWith<$Res> { + factory $BridgeEvent_ConnectedCopyWith(BridgeEvent_Connected value, $Res Function(BridgeEvent_Connected) _then) = _$BridgeEvent_ConnectedCopyWithImpl; +@useResult +$Res call({ + String serverName +}); + + + + +} +/// @nodoc +class _$BridgeEvent_ConnectedCopyWithImpl<$Res> + implements $BridgeEvent_ConnectedCopyWith<$Res> { + _$BridgeEvent_ConnectedCopyWithImpl(this._self, this._then); + + final BridgeEvent_Connected _self; + final $Res Function(BridgeEvent_Connected) _then; + +/// Create a copy of BridgeEvent +/// with the given fields replaced by the non-null parameter values. +@pragma('vm:prefer-inline') $Res call({Object? serverName = null,}) { + return _then(BridgeEvent_Connected( +serverName: null == serverName ? _self.serverName : serverName // ignore: cast_nullable_to_non_nullable +as String, + )); +} + + +} + +/// @nodoc + + +class BridgeEvent_Lost extends BridgeEvent { + const BridgeEvent_Lost({required this.reason}): super._(); + + +/// Reason classification from the protocol layer. + final String reason; + +/// Create a copy of BridgeEvent +/// with the given fields replaced by the non-null parameter values. +@JsonKey(includeFromJson: false, includeToJson: false) +@pragma('vm:prefer-inline') +$BridgeEvent_LostCopyWith get copyWith => _$BridgeEvent_LostCopyWithImpl(this, _$identity); + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is BridgeEvent_Lost&&(identical(other.reason, reason) || other.reason == reason)); +} + + +@override +int get hashCode => Object.hash(runtimeType,reason); + +@override +String toString() { + return 'BridgeEvent.lost(reason: $reason)'; +} + + +} + +/// @nodoc +abstract mixin class $BridgeEvent_LostCopyWith<$Res> implements $BridgeEventCopyWith<$Res> { + factory $BridgeEvent_LostCopyWith(BridgeEvent_Lost value, $Res Function(BridgeEvent_Lost) _then) = _$BridgeEvent_LostCopyWithImpl; +@useResult +$Res call({ + String reason +}); + + + + +} +/// @nodoc +class _$BridgeEvent_LostCopyWithImpl<$Res> + implements $BridgeEvent_LostCopyWith<$Res> { + _$BridgeEvent_LostCopyWithImpl(this._self, this._then); + + final BridgeEvent_Lost _self; + final $Res Function(BridgeEvent_Lost) _then; + +/// Create a copy of BridgeEvent +/// with the given fields replaced by the non-null parameter values. +@pragma('vm:prefer-inline') $Res call({Object? reason = null,}) { + return _then(BridgeEvent_Lost( +reason: null == reason ? _self.reason : reason // ignore: cast_nullable_to_non_nullable +as String, + )); +} + + +} + +/// @nodoc + + +class BridgeEvent_Reconnecting extends BridgeEvent { + const BridgeEvent_Reconnecting({required this.attempt, required this.delaySecs}): super._(); + + +/// 1-based attempt counter for the current outage. + final int attempt; +/// Seconds the supervisor will sleep before this attempt. + final int delaySecs; + +/// Create a copy of BridgeEvent +/// with the given fields replaced by the non-null parameter values. +@JsonKey(includeFromJson: false, includeToJson: false) +@pragma('vm:prefer-inline') +$BridgeEvent_ReconnectingCopyWith get copyWith => _$BridgeEvent_ReconnectingCopyWithImpl(this, _$identity); + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is BridgeEvent_Reconnecting&&(identical(other.attempt, attempt) || other.attempt == attempt)&&(identical(other.delaySecs, delaySecs) || other.delaySecs == delaySecs)); +} + + +@override +int get hashCode => Object.hash(runtimeType,attempt,delaySecs); + +@override +String toString() { + return 'BridgeEvent.reconnecting(attempt: $attempt, delaySecs: $delaySecs)'; +} + + +} + +/// @nodoc +abstract mixin class $BridgeEvent_ReconnectingCopyWith<$Res> implements $BridgeEventCopyWith<$Res> { + factory $BridgeEvent_ReconnectingCopyWith(BridgeEvent_Reconnecting value, $Res Function(BridgeEvent_Reconnecting) _then) = _$BridgeEvent_ReconnectingCopyWithImpl; +@useResult +$Res call({ + int attempt, int delaySecs +}); + + + + +} +/// @nodoc +class _$BridgeEvent_ReconnectingCopyWithImpl<$Res> + implements $BridgeEvent_ReconnectingCopyWith<$Res> { + _$BridgeEvent_ReconnectingCopyWithImpl(this._self, this._then); + + final BridgeEvent_Reconnecting _self; + final $Res Function(BridgeEvent_Reconnecting) _then; + +/// Create a copy of BridgeEvent +/// with the given fields replaced by the non-null parameter values. +@pragma('vm:prefer-inline') $Res call({Object? attempt = null,Object? delaySecs = null,}) { + return _then(BridgeEvent_Reconnecting( +attempt: null == attempt ? _self.attempt : attempt // ignore: cast_nullable_to_non_nullable +as int,delaySecs: null == delaySecs ? _self.delaySecs : delaySecs // ignore: cast_nullable_to_non_nullable +as int, + )); +} + + +} + +/// @nodoc + + +class BridgeEvent_Disconnected extends BridgeEvent { + const BridgeEvent_Disconnected({required this.reason}): super._(); + + +/// Reason classification. + final String reason; + +/// Create a copy of BridgeEvent +/// with the given fields replaced by the non-null parameter values. +@JsonKey(includeFromJson: false, includeToJson: false) +@pragma('vm:prefer-inline') +$BridgeEvent_DisconnectedCopyWith get copyWith => _$BridgeEvent_DisconnectedCopyWithImpl(this, _$identity); + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is BridgeEvent_Disconnected&&(identical(other.reason, reason) || other.reason == reason)); +} + + +@override +int get hashCode => Object.hash(runtimeType,reason); + +@override +String toString() { + return 'BridgeEvent.disconnected(reason: $reason)'; +} + + +} + +/// @nodoc +abstract mixin class $BridgeEvent_DisconnectedCopyWith<$Res> implements $BridgeEventCopyWith<$Res> { + factory $BridgeEvent_DisconnectedCopyWith(BridgeEvent_Disconnected value, $Res Function(BridgeEvent_Disconnected) _then) = _$BridgeEvent_DisconnectedCopyWithImpl; +@useResult +$Res call({ + String reason +}); + + + + +} +/// @nodoc +class _$BridgeEvent_DisconnectedCopyWithImpl<$Res> + implements $BridgeEvent_DisconnectedCopyWith<$Res> { + _$BridgeEvent_DisconnectedCopyWithImpl(this._self, this._then); + + final BridgeEvent_Disconnected _self; + final $Res Function(BridgeEvent_Disconnected) _then; + +/// Create a copy of BridgeEvent +/// with the given fields replaced by the non-null parameter values. +@pragma('vm:prefer-inline') $Res call({Object? reason = null,}) { + return _then(BridgeEvent_Disconnected( +reason: null == reason ? _self.reason : reason // ignore: cast_nullable_to_non_nullable +as String, + )); +} + + +} + +/// @nodoc + + +class BridgeEvent_AudioStarted extends BridgeEvent { + const BridgeEvent_AudioStarted(): super._(); + + + + + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is BridgeEvent_AudioStarted); +} + + +@override +int get hashCode => runtimeType.hashCode; + +@override +String toString() { + return 'BridgeEvent.audioStarted()'; +} + + +} + + + + +/// @nodoc + + +class BridgeEvent_AudioStopped extends BridgeEvent { + const BridgeEvent_AudioStopped(): super._(); + + + + + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is BridgeEvent_AudioStopped); +} + + +@override +int get hashCode => runtimeType.hashCode; + +@override +String toString() { + return 'BridgeEvent.audioStopped()'; +} + + +} + + + + +// dart format on diff --git a/apps/chanora_flutter/lib/src/rust/frb_generated.dart b/apps/chanora_flutter/lib/src/rust/frb_generated.dart index f800ce3..4c3b1f0 100644 --- a/apps/chanora_flutter/lib/src/rust/frb_generated.dart +++ b/apps/chanora_flutter/lib/src/rust/frb_generated.dart @@ -67,7 +67,7 @@ class RustLib extends BaseEntrypoint { String get codegenVersion => '2.12.0'; @override - int get rustContentHash => 1944264248; + int get rustContentHash => -1896742393; static const kDefaultExternalLibraryLoaderConfig = ExternalLibraryLoaderConfig( @@ -90,6 +90,8 @@ abstract class RustLibApi extends BaseApi { Future crateApiDisconnect(); + Stream crateApiEventsStream(); + Future crateApiIsConnected(); Future crateApiSetPtt({required bool active}); @@ -220,6 +222,38 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { TaskConstMeta get kCrateApiDisconnectConstMeta => const TaskConstMeta(debugName: "disconnect", argNames: []); + @override + Stream crateApiEventsStream() { + final sink = RustStreamSink(); + unawaited( + handler.executeNormal( + NormalTask( + callFfi: (port_) { + final serializer = SseSerializer(generalizedFrbRustBinding); + sse_encode_StreamSink_bridge_event_Sse(sink, serializer); + pdeCallFfi( + generalizedFrbRustBinding, + serializer, + funcId: 5, + port: port_, + ); + }, + codec: SseCodec( + decodeSuccessData: sse_decode_unit, + decodeErrorData: sse_decode_bridge_error, + ), + constMeta: kCrateApiEventsStreamConstMeta, + argValues: [sink], + apiImpl: this, + ), + ), + ); + return sink.stream; + } + + TaskConstMeta get kCrateApiEventsStreamConstMeta => + const TaskConstMeta(debugName: "events_stream", argNames: ["sink"]); + @override Future crateApiIsConnected() { return handler.executeNormal( @@ -229,7 +263,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { pdeCallFfi( generalizedFrbRustBinding, serializer, - funcId: 5, + funcId: 6, port: port_, ); }, @@ -257,7 +291,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { pdeCallFfi( generalizedFrbRustBinding, serializer, - funcId: 6, + funcId: 7, port: port_, ); }, @@ -284,7 +318,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { pdeCallFfi( generalizedFrbRustBinding, serializer, - funcId: 7, + funcId: 8, port: port_, ); }, @@ -311,7 +345,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { pdeCallFfi( generalizedFrbRustBinding, serializer, - funcId: 8, + funcId: 9, port: port_, ); }, @@ -329,6 +363,20 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { TaskConstMeta get kCrateApiStartAudioConstMeta => const TaskConstMeta(debugName: "start_audio", argNames: []); + @protected + AnyhowException dco_decode_AnyhowException(dynamic raw) { + // Codec=Dco (DartCObject based), see doc to use other codecs + return AnyhowException(raw as String); + } + + @protected + RustStreamSink dco_decode_StreamSink_bridge_event_Sse( + dynamic raw, + ) { + // Codec=Dco (DartCObject based), see doc to use other codecs + throw UnimplementedError(); + } + @protected String dco_decode_String(dynamic raw) { // Codec=Dco (DartCObject based), see doc to use other codecs @@ -388,18 +436,47 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { case 0: return BridgeError_InvalidCommand(dco_decode_String(raw[1])); case 1: - return BridgeError_Connection(dco_decode_String(raw[1])); + return BridgeError_DnsFailed( + host: dco_decode_String(raw[1]), + reason: dco_decode_String(raw[2]), + ); case 2: - return BridgeError_NotConnected(); + return BridgeError_Connection(dco_decode_String(raw[1])); case 3: - return BridgeError_AlreadyConnected(); + return BridgeError_NotConnected(); case 4: + return BridgeError_AlreadyConnected(); + case 5: return BridgeError_Unmapped(dco_decode_String(raw[1])); default: throw Exception("unreachable"); } } + @protected + BridgeEvent dco_decode_bridge_event(dynamic raw) { + // Codec=Dco (DartCObject based), see doc to use other codecs + switch (raw[0]) { + case 0: + return BridgeEvent_Connected(serverName: dco_decode_String(raw[1])); + case 1: + return BridgeEvent_Lost(reason: dco_decode_String(raw[1])); + case 2: + return BridgeEvent_Reconnecting( + attempt: dco_decode_u_32(raw[1]), + delaySecs: dco_decode_u_32(raw[2]), + ); + case 3: + return BridgeEvent_Disconnected(reason: dco_decode_String(raw[1])); + case 4: + return BridgeEvent_AudioStarted(); + case 5: + return BridgeEvent_AudioStopped(); + default: + throw Exception("unreachable"); + } + } + @protected BridgeSnapshot dco_decode_bridge_snapshot(dynamic raw) { // Codec=Dco (DartCObject based), see doc to use other codecs @@ -464,6 +541,21 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { return; } + @protected + AnyhowException sse_decode_AnyhowException(SseDeserializer deserializer) { + // Codec=Sse (Serialization based), see doc to use other codecs + var inner = sse_decode_String(deserializer); + return AnyhowException(inner); + } + + @protected + RustStreamSink sse_decode_StreamSink_bridge_event_Sse( + SseDeserializer deserializer, + ) { + // Codec=Sse (Serialization based), see doc to use other codecs + throw UnimplementedError('Unreachable ()'); + } + @protected String sse_decode_String(SseDeserializer deserializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -524,13 +616,17 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { var var_field0 = sse_decode_String(deserializer); return BridgeError_InvalidCommand(var_field0); case 1: + var var_host = sse_decode_String(deserializer); + var var_reason = sse_decode_String(deserializer); + return BridgeError_DnsFailed(host: var_host, reason: var_reason); + case 2: var var_field0 = sse_decode_String(deserializer); return BridgeError_Connection(var_field0); - case 2: - return BridgeError_NotConnected(); case 3: - return BridgeError_AlreadyConnected(); + return BridgeError_NotConnected(); case 4: + return BridgeError_AlreadyConnected(); + case 5: var var_field0 = sse_decode_String(deserializer); return BridgeError_Unmapped(var_field0); default: @@ -538,6 +634,37 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { } } + @protected + BridgeEvent sse_decode_bridge_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_serverName = sse_decode_String(deserializer); + return BridgeEvent_Connected(serverName: var_serverName); + case 1: + var var_reason = sse_decode_String(deserializer); + return BridgeEvent_Lost(reason: var_reason); + case 2: + var var_attempt = sse_decode_u_32(deserializer); + var var_delaySecs = sse_decode_u_32(deserializer); + return BridgeEvent_Reconnecting( + attempt: var_attempt, + delaySecs: var_delaySecs, + ); + case 3: + var var_reason = sse_decode_String(deserializer); + return BridgeEvent_Disconnected(reason: var_reason); + case 4: + return BridgeEvent_AudioStarted(); + case 5: + return BridgeEvent_AudioStopped(); + default: + throw UnimplementedError(''); + } + } + @protected BridgeSnapshot sse_decode_bridge_snapshot(SseDeserializer deserializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -627,6 +754,32 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { return deserializer.buffer.getInt32(); } + @protected + void sse_encode_AnyhowException( + AnyhowException self, + SseSerializer serializer, + ) { + // Codec=Sse (Serialization based), see doc to use other codecs + sse_encode_String(self.message, serializer); + } + + @protected + void sse_encode_StreamSink_bridge_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_bridge_event, + decodeErrorData: sse_decode_AnyhowException, + ), + ), + serializer, + ); + } + @protected void sse_encode_String(String self, SseSerializer serializer) { // Codec=Sse (Serialization based), see doc to use other codecs @@ -674,19 +827,50 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { case BridgeError_InvalidCommand(field0: final field0): sse_encode_i_32(0, serializer); sse_encode_String(field0, serializer); - case BridgeError_Connection(field0: final field0): + case BridgeError_DnsFailed(host: final host, reason: final reason): sse_encode_i_32(1, serializer); + sse_encode_String(host, serializer); + sse_encode_String(reason, serializer); + case BridgeError_Connection(field0: final field0): + sse_encode_i_32(2, serializer); sse_encode_String(field0, serializer); case BridgeError_NotConnected(): - sse_encode_i_32(2, serializer); - case BridgeError_AlreadyConnected(): sse_encode_i_32(3, serializer); - case BridgeError_Unmapped(field0: final field0): + case BridgeError_AlreadyConnected(): sse_encode_i_32(4, serializer); + case BridgeError_Unmapped(field0: final field0): + sse_encode_i_32(5, serializer); sse_encode_String(field0, serializer); } } + @protected + void sse_encode_bridge_event(BridgeEvent self, SseSerializer serializer) { + // Codec=Sse (Serialization based), see doc to use other codecs + switch (self) { + case BridgeEvent_Connected(serverName: final serverName): + sse_encode_i_32(0, serializer); + sse_encode_String(serverName, serializer); + case BridgeEvent_Lost(reason: final reason): + sse_encode_i_32(1, serializer); + sse_encode_String(reason, serializer); + case BridgeEvent_Reconnecting( + attempt: final attempt, + delaySecs: final delaySecs, + ): + sse_encode_i_32(2, serializer); + sse_encode_u_32(attempt, serializer); + sse_encode_u_32(delaySecs, serializer); + case BridgeEvent_Disconnected(reason: final reason): + sse_encode_i_32(3, serializer); + sse_encode_String(reason, serializer); + case BridgeEvent_AudioStarted(): + sse_encode_i_32(4, serializer); + case BridgeEvent_AudioStopped(): + sse_encode_i_32(5, serializer); + } + } + @protected void sse_encode_bridge_snapshot( BridgeSnapshot self, diff --git a/apps/chanora_flutter/lib/src/rust/frb_generated.io.dart b/apps/chanora_flutter/lib/src/rust/frb_generated.io.dart index 590342d..8ec8270 100644 --- a/apps/chanora_flutter/lib/src/rust/frb_generated.io.dart +++ b/apps/chanora_flutter/lib/src/rust/frb_generated.io.dart @@ -19,6 +19,14 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { required super.portManager, }); + @protected + AnyhowException dco_decode_AnyhowException(dynamic raw); + + @protected + RustStreamSink dco_decode_StreamSink_bridge_event_Sse( + dynamic raw, + ); + @protected String dco_decode_String(dynamic raw); @@ -37,6 +45,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected BridgeError dco_decode_bridge_error(dynamic raw); + @protected + BridgeEvent dco_decode_bridge_event(dynamic raw); + @protected BridgeSnapshot dco_decode_bridge_snapshot(dynamic raw); @@ -64,6 +75,14 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void dco_decode_unit(dynamic raw); + @protected + AnyhowException sse_decode_AnyhowException(SseDeserializer deserializer); + + @protected + RustStreamSink sse_decode_StreamSink_bridge_event_Sse( + SseDeserializer deserializer, + ); + @protected String sse_decode_String(SseDeserializer deserializer); @@ -82,6 +101,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected BridgeError sse_decode_bridge_error(SseDeserializer deserializer); + @protected + BridgeEvent sse_decode_bridge_event(SseDeserializer deserializer); + @protected BridgeSnapshot sse_decode_bridge_snapshot(SseDeserializer deserializer); @@ -116,6 +138,18 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected int sse_decode_i_32(SseDeserializer deserializer); + @protected + void sse_encode_AnyhowException( + AnyhowException self, + SseSerializer serializer, + ); + + @protected + void sse_encode_StreamSink_bridge_event_Sse( + RustStreamSink self, + SseSerializer serializer, + ); + @protected void sse_encode_String(String self, SseSerializer serializer); @@ -137,6 +171,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void sse_encode_bridge_error(BridgeError self, SseSerializer serializer); + @protected + void sse_encode_bridge_event(BridgeEvent self, SseSerializer serializer); + @protected void sse_encode_bridge_snapshot( BridgeSnapshot self, diff --git a/apps/chanora_flutter/lib/src/rust/frb_generated.web.dart b/apps/chanora_flutter/lib/src/rust/frb_generated.web.dart index d67604e..ad79de9 100644 --- a/apps/chanora_flutter/lib/src/rust/frb_generated.web.dart +++ b/apps/chanora_flutter/lib/src/rust/frb_generated.web.dart @@ -21,6 +21,14 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { required super.portManager, }); + @protected + AnyhowException dco_decode_AnyhowException(dynamic raw); + + @protected + RustStreamSink dco_decode_StreamSink_bridge_event_Sse( + dynamic raw, + ); + @protected String dco_decode_String(dynamic raw); @@ -39,6 +47,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected BridgeError dco_decode_bridge_error(dynamic raw); + @protected + BridgeEvent dco_decode_bridge_event(dynamic raw); + @protected BridgeSnapshot dco_decode_bridge_snapshot(dynamic raw); @@ -66,6 +77,14 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void dco_decode_unit(dynamic raw); + @protected + AnyhowException sse_decode_AnyhowException(SseDeserializer deserializer); + + @protected + RustStreamSink sse_decode_StreamSink_bridge_event_Sse( + SseDeserializer deserializer, + ); + @protected String sse_decode_String(SseDeserializer deserializer); @@ -84,6 +103,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected BridgeError sse_decode_bridge_error(SseDeserializer deserializer); + @protected + BridgeEvent sse_decode_bridge_event(SseDeserializer deserializer); + @protected BridgeSnapshot sse_decode_bridge_snapshot(SseDeserializer deserializer); @@ -118,6 +140,18 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected int sse_decode_i_32(SseDeserializer deserializer); + @protected + void sse_encode_AnyhowException( + AnyhowException self, + SseSerializer serializer, + ); + + @protected + void sse_encode_StreamSink_bridge_event_Sse( + RustStreamSink self, + SseSerializer serializer, + ); + @protected void sse_encode_String(String self, SseSerializer serializer); @@ -139,6 +173,9 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected void sse_encode_bridge_error(BridgeError self, SseSerializer serializer); + @protected + void sse_encode_bridge_event(BridgeEvent self, SseSerializer serializer); + @protected void sse_encode_bridge_snapshot( BridgeSnapshot self, diff --git a/apps/chanora_flutter/lib/src/rust/lib.dart b/apps/chanora_flutter/lib/src/rust/lib.dart index d4650a8..c4ff538 100644 --- a/apps/chanora_flutter/lib/src/rust/lib.dart +++ b/apps/chanora_flutter/lib/src/rust/lib.dart @@ -16,6 +16,16 @@ sealed class BridgeError with _$BridgeError implements FrbException { const factory BridgeError.invalidCommand(String field0) = BridgeError_InvalidCommand; + /// Hostname resolution failed. Distinct from `Connection` so the + /// UI can show a meaningful "Server not found" message. + const factory BridgeError.dnsFailed({ + /// The hostname (or `host:port`) the caller submitted. + required String host, + + /// Reason from the platform resolver. + required String reason, + }) = BridgeError_DnsFailed; + /// Connection layer failure (typed-mapped from CoreError). const factory BridgeError.connection(String field0) = BridgeError_Connection; diff --git a/apps/chanora_flutter/lib/src/rust/lib.freezed.dart b/apps/chanora_flutter/lib/src/rust/lib.freezed.dart index 3a88eab..5f45986 100644 --- a/apps/chanora_flutter/lib/src/rust/lib.freezed.dart +++ b/apps/chanora_flutter/lib/src/rust/lib.freezed.dart @@ -55,11 +55,12 @@ extension BridgeErrorPatterns on BridgeError { /// } /// ``` -@optionalTypeArgs TResult maybeMap({TResult Function( BridgeError_InvalidCommand value)? invalidCommand,TResult Function( BridgeError_Connection value)? connection,TResult Function( BridgeError_NotConnected value)? notConnected,TResult Function( BridgeError_AlreadyConnected value)? alreadyConnected,TResult Function( BridgeError_Unmapped value)? unmapped,required TResult orElse(),}){ +@optionalTypeArgs TResult maybeMap({TResult Function( BridgeError_InvalidCommand value)? invalidCommand,TResult Function( BridgeError_DnsFailed value)? dnsFailed,TResult Function( BridgeError_Connection value)? connection,TResult Function( BridgeError_NotConnected value)? notConnected,TResult Function( BridgeError_AlreadyConnected value)? alreadyConnected,TResult Function( BridgeError_Unmapped value)? unmapped,required TResult orElse(),}){ final _that = this; switch (_that) { case BridgeError_InvalidCommand() when invalidCommand != null: -return invalidCommand(_that);case BridgeError_Connection() when connection != null: +return invalidCommand(_that);case BridgeError_DnsFailed() when dnsFailed != null: +return dnsFailed(_that);case BridgeError_Connection() when connection != null: return connection(_that);case BridgeError_NotConnected() when notConnected != null: return notConnected(_that);case BridgeError_AlreadyConnected() when alreadyConnected != null: return alreadyConnected(_that);case BridgeError_Unmapped() when unmapped != null: @@ -81,11 +82,12 @@ return unmapped(_that);case _: /// } /// ``` -@optionalTypeArgs TResult map({required TResult Function( BridgeError_InvalidCommand value) invalidCommand,required TResult Function( BridgeError_Connection value) connection,required TResult Function( BridgeError_NotConnected value) notConnected,required TResult Function( BridgeError_AlreadyConnected value) alreadyConnected,required TResult Function( BridgeError_Unmapped value) unmapped,}){ +@optionalTypeArgs TResult map({required TResult Function( BridgeError_InvalidCommand value) invalidCommand,required TResult Function( BridgeError_DnsFailed value) dnsFailed,required TResult Function( BridgeError_Connection value) connection,required TResult Function( BridgeError_NotConnected value) notConnected,required TResult Function( BridgeError_AlreadyConnected value) alreadyConnected,required TResult Function( BridgeError_Unmapped value) unmapped,}){ final _that = this; switch (_that) { case BridgeError_InvalidCommand(): -return invalidCommand(_that);case BridgeError_Connection(): +return invalidCommand(_that);case BridgeError_DnsFailed(): +return dnsFailed(_that);case BridgeError_Connection(): return connection(_that);case BridgeError_NotConnected(): return notConnected(_that);case BridgeError_AlreadyConnected(): return alreadyConnected(_that);case BridgeError_Unmapped(): @@ -103,11 +105,12 @@ return unmapped(_that);} /// } /// ``` -@optionalTypeArgs TResult? mapOrNull({TResult? Function( BridgeError_InvalidCommand value)? invalidCommand,TResult? Function( BridgeError_Connection value)? connection,TResult? Function( BridgeError_NotConnected value)? notConnected,TResult? Function( BridgeError_AlreadyConnected value)? alreadyConnected,TResult? Function( BridgeError_Unmapped value)? unmapped,}){ +@optionalTypeArgs TResult? mapOrNull({TResult? Function( BridgeError_InvalidCommand value)? invalidCommand,TResult? Function( BridgeError_DnsFailed value)? dnsFailed,TResult? Function( BridgeError_Connection value)? connection,TResult? Function( BridgeError_NotConnected value)? notConnected,TResult? Function( BridgeError_AlreadyConnected value)? alreadyConnected,TResult? Function( BridgeError_Unmapped value)? unmapped,}){ final _that = this; switch (_that) { case BridgeError_InvalidCommand() when invalidCommand != null: -return invalidCommand(_that);case BridgeError_Connection() when connection != null: +return invalidCommand(_that);case BridgeError_DnsFailed() when dnsFailed != null: +return dnsFailed(_that);case BridgeError_Connection() when connection != null: return connection(_that);case BridgeError_NotConnected() when notConnected != null: return notConnected(_that);case BridgeError_AlreadyConnected() when alreadyConnected != null: return alreadyConnected(_that);case BridgeError_Unmapped() when unmapped != null: @@ -128,10 +131,11 @@ return unmapped(_that);case _: /// } /// ``` -@optionalTypeArgs TResult maybeWhen({TResult Function( String field0)? invalidCommand,TResult Function( String field0)? connection,TResult Function()? notConnected,TResult Function()? alreadyConnected,TResult Function( String field0)? unmapped,required TResult orElse(),}) {final _that = this; +@optionalTypeArgs TResult maybeWhen({TResult Function( String field0)? invalidCommand,TResult Function( String host, String reason)? dnsFailed,TResult Function( String field0)? connection,TResult Function()? notConnected,TResult Function()? alreadyConnected,TResult Function( String field0)? unmapped,required TResult orElse(),}) {final _that = this; switch (_that) { case BridgeError_InvalidCommand() when invalidCommand != null: -return invalidCommand(_that.field0);case BridgeError_Connection() when connection != null: +return invalidCommand(_that.field0);case BridgeError_DnsFailed() when dnsFailed != null: +return dnsFailed(_that.host,_that.reason);case BridgeError_Connection() when connection != null: return connection(_that.field0);case BridgeError_NotConnected() when notConnected != null: return notConnected();case BridgeError_AlreadyConnected() when alreadyConnected != null: return alreadyConnected();case BridgeError_Unmapped() when unmapped != null: @@ -153,10 +157,11 @@ return unmapped(_that.field0);case _: /// } /// ``` -@optionalTypeArgs TResult when({required TResult Function( String field0) invalidCommand,required TResult Function( String field0) connection,required TResult Function() notConnected,required TResult Function() alreadyConnected,required TResult Function( String field0) unmapped,}) {final _that = this; +@optionalTypeArgs TResult when({required TResult Function( String field0) invalidCommand,required TResult Function( String host, String reason) dnsFailed,required TResult Function( String field0) connection,required TResult Function() notConnected,required TResult Function() alreadyConnected,required TResult Function( String field0) unmapped,}) {final _that = this; switch (_that) { case BridgeError_InvalidCommand(): -return invalidCommand(_that.field0);case BridgeError_Connection(): +return invalidCommand(_that.field0);case BridgeError_DnsFailed(): +return dnsFailed(_that.host,_that.reason);case BridgeError_Connection(): return connection(_that.field0);case BridgeError_NotConnected(): return notConnected();case BridgeError_AlreadyConnected(): return alreadyConnected();case BridgeError_Unmapped(): @@ -174,10 +179,11 @@ return unmapped(_that.field0);} /// } /// ``` -@optionalTypeArgs TResult? whenOrNull({TResult? Function( String field0)? invalidCommand,TResult? Function( String field0)? connection,TResult? Function()? notConnected,TResult? Function()? alreadyConnected,TResult? Function( String field0)? unmapped,}) {final _that = this; +@optionalTypeArgs TResult? whenOrNull({TResult? Function( String field0)? invalidCommand,TResult? Function( String host, String reason)? dnsFailed,TResult? Function( String field0)? connection,TResult? Function()? notConnected,TResult? Function()? alreadyConnected,TResult? Function( String field0)? unmapped,}) {final _that = this; switch (_that) { case BridgeError_InvalidCommand() when invalidCommand != null: -return invalidCommand(_that.field0);case BridgeError_Connection() when connection != null: +return invalidCommand(_that.field0);case BridgeError_DnsFailed() when dnsFailed != null: +return dnsFailed(_that.host,_that.reason);case BridgeError_Connection() when connection != null: return connection(_that.field0);case BridgeError_NotConnected() when notConnected != null: return notConnected();case BridgeError_AlreadyConnected() when alreadyConnected != null: return alreadyConnected();case BridgeError_Unmapped() when unmapped != null: @@ -258,6 +264,76 @@ as String, /// @nodoc +class BridgeError_DnsFailed extends BridgeError { + const BridgeError_DnsFailed({required this.host, required this.reason}): super._(); + + +/// The hostname (or `host:port`) the caller submitted. + final String host; +/// Reason from the platform resolver. + final String reason; + +/// Create a copy of BridgeError +/// with the given fields replaced by the non-null parameter values. +@JsonKey(includeFromJson: false, includeToJson: false) +@pragma('vm:prefer-inline') +$BridgeError_DnsFailedCopyWith get copyWith => _$BridgeError_DnsFailedCopyWithImpl(this, _$identity); + + + +@override +bool operator ==(Object other) { + return identical(this, other) || (other.runtimeType == runtimeType&&other is BridgeError_DnsFailed&&(identical(other.host, host) || other.host == host)&&(identical(other.reason, reason) || other.reason == reason)); +} + + +@override +int get hashCode => Object.hash(runtimeType,host,reason); + +@override +String toString() { + return 'BridgeError.dnsFailed(host: $host, reason: $reason)'; +} + + +} + +/// @nodoc +abstract mixin class $BridgeError_DnsFailedCopyWith<$Res> implements $BridgeErrorCopyWith<$Res> { + factory $BridgeError_DnsFailedCopyWith(BridgeError_DnsFailed value, $Res Function(BridgeError_DnsFailed) _then) = _$BridgeError_DnsFailedCopyWithImpl; +@useResult +$Res call({ + String host, String reason +}); + + + + +} +/// @nodoc +class _$BridgeError_DnsFailedCopyWithImpl<$Res> + implements $BridgeError_DnsFailedCopyWith<$Res> { + _$BridgeError_DnsFailedCopyWithImpl(this._self, this._then); + + final BridgeError_DnsFailed _self; + final $Res Function(BridgeError_DnsFailed) _then; + +/// Create a copy of BridgeError +/// with the given fields replaced by the non-null parameter values. +@pragma('vm:prefer-inline') $Res call({Object? host = null,Object? reason = null,}) { + return _then(BridgeError_DnsFailed( +host: null == host ? _self.host : host // ignore: cast_nullable_to_non_nullable +as String,reason: null == reason ? _self.reason : reason // ignore: cast_nullable_to_non_nullable +as String, + )); +} + + +} + +/// @nodoc + + class BridgeError_Connection extends BridgeError { const BridgeError_Connection(this.field0): super._(); diff --git a/core/chanora_core/src/lib.rs b/core/chanora_core/src/lib.rs index 1e2ac27..6d0bba5 100644 --- a/core/chanora_core/src/lib.rs +++ b/core/chanora_core/src/lib.rs @@ -18,23 +18,36 @@ //! * Secret material never lands in `chanora_storage`'s non-secret //! side (DEC-013.2 / SS-AUD-001/002). //! -//! ## Alpha scope +//! ## A.6 supervisor and reconnect //! -//! `ChanoraSession::connect`, `snapshot`, and `disconnect` are wired -//! through `chanora_protocol`. Audio, storage, and diagnostics are -//! still scaffolds. +//! `ChanoraSession` spawns a per-connection supervisor task on a +//! successful [`Self::connect`]. The supervisor: +//! +//! * awaits the `ProtocolClient`'s loss notifier; +//! * on user-driven disconnect, exits silently; +//! * on stream-end or error, re-dials with exponential backoff +//! (1s, 2s, 5s, 15s, 30s, 60s capped); cancellation-safe via the +//! per-session `cancel_tx` oneshot; +//! * re-attaches the audio engine to the fresh protocol client if +//! audio was running prior to the loss. +//! +//! Lifecycle changes are broadcast to subscribers via +//! [`ChanoraSession::subscribe_events`]. #![forbid(unsafe_code)] #![warn(missing_docs)] use std::sync::Arc; +use std::time::Duration; use thiserror::Error; -use tokio::sync::Mutex; +use tokio::sync::{broadcast, oneshot, Mutex}; +use tokio::task::JoinHandle; +use tracing::{info, warn}; pub use chanora_audio::{AudioEngine, AudioEngineConfig}; pub use chanora_protocol::{ - ChannelInfo, ClientInfo, ConnectConfig, ProtocolError, ServerSnapshot, + ChannelInfo, ClientInfo, ConnectConfig, DisconnectReason, ProtocolError, ServerSnapshot, }; /// Errors that can arise during top-level orchestration. @@ -71,9 +84,75 @@ pub enum CoreError { AudioNotStarted, } +/// High-level lifecycle event surfaced to subscribers. +/// +/// This is the minimal set needed for A.6 (reconnect banner). The +/// full event catalogue lands in A.4. +#[derive(Debug, Clone)] +pub enum SessionEvent { + /// Initial connect succeeded, or reconnect attempt succeeded. + Connected { + /// Server name reported in the snapshot. + server_name: String, + }, + /// Connection lost; the supervisor will retry. + Lost { + /// Reason classification from the protocol layer. + reason: String, + }, + /// Supervisor is sleeping before its next reconnect attempt. + Reconnecting { + /// 1-based attempt counter for the current outage. + attempt: u32, + /// Seconds the supervisor will sleep before this attempt. + delay_secs: u32, + }, + /// Supervisor gave up after `attempt` failed retries (or the + /// user explicitly disconnected mid-outage). + Disconnected { + /// Reason classification from the protocol layer. + reason: String, + }, + /// Audio engine started (e.g. after a successful reconnect with + /// reattachment). + AudioStarted, + /// Audio engine stopped (e.g. before a reconnect cycle, or by + /// explicit user action). + AudioStopped, +} + +/// Channel capacity for the broadcast events. Generous because +/// reconnect cycles emit several events per attempt; if subscribers +/// fall behind we'd rather skip than block the supervisor. +const EVENT_CHANNEL_CAPACITY: usize = 64; + +struct SupervisorInner { + /// Optional cached AudioEngineConfig — set when start_audio is + /// first called, used to re-create the engine after a reconnect. + audio_cfg: Option, + /// True when audio is currently desired (start_audio called, no + /// stop yet). Drives whether the supervisor reattaches audio on + /// a successful reconnect. + audio_running: bool, +} + struct ConnectedState { protocol: chanora_protocol::ProtocolClient, audio: Option, + /// Cancellation signal for the supervisor task. Dropped on + /// explicit disconnect to break the supervisor out of its + /// backoff sleep. + cancel_tx: Option>, + /// Supervisor task handle. Awaited on disconnect for clean + /// teardown. + supervisor: Option>, + /// Connection config used to dial this connection; retained for + /// future diagnostics. The supervisor task holds its own clone. + #[allow(dead_code)] + cfg: ConnectConfig, + /// Audio supervision state. Wrapped in Arc> so the + /// supervisor and the public API both see updates. + sup_inner: Arc>, } /// The top-level Chanora session. Owns at most one active server @@ -81,16 +160,28 @@ struct ConnectedState { #[derive(Clone)] pub struct ChanoraSession { inner: Arc>>, + events_tx: broadcast::Sender, } impl ChanoraSession { /// Construct an empty session. Performs no I/O. pub fn new() -> Self { + let (events_tx, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); Self { inner: Arc::new(Mutex::new(None)), + events_tx, } } + /// Subscribe to lifecycle events. The returned receiver fires + /// on connect / lost / reconnecting / disconnected / + /// audio-started / audio-stopped transitions. Multiple + /// subscribers are allowed; each gets its own slow-consumer + /// behaviour (events drop with [`broadcast::error::RecvError::Lagged`]). + pub fn subscribe_events(&self) -> broadcast::Receiver { + self.events_tx.subscribe() + } + /// Connect to a server. Fails with [`CoreError::AlreadyConnected`] /// if a connection is already active (DEC-006). Audio is not /// started automatically; call [`Self::start_audio`] after. @@ -99,11 +190,42 @@ impl ChanoraSession { if guard.is_some() { return Err(CoreError::AlreadyConnected); } - let client = chanora_protocol::ProtocolClient::connect(cfg).await?; + let client = chanora_protocol::ProtocolClient::connect(cfg.clone()).await?; let snap = client.snapshot().await?; + + // Set up the supervisor. + let (cancel_tx, cancel_rx) = oneshot::channel::<()>(); + let lost_rx = client + .take_loss_notifier() + .ok_or(CoreError::Invariant("lost notifier already taken"))?; + let probe = client.snapshot_probe(); + + let sup_inner = Arc::new(Mutex::new(SupervisorInner { + audio_cfg: None, + audio_running: false, + })); + + let supervisor = tokio::spawn(supervisor_loop( + self.inner.clone(), + self.events_tx.clone(), + cfg.clone(), + lost_rx, + probe, + cancel_rx, + sup_inner.clone(), + )); + + let _ = self.events_tx.send(SessionEvent::Connected { + server_name: snap.server_name.clone(), + }); + *guard = Some(ConnectedState { protocol: client, audio: None, + cancel_tx: Some(cancel_tx), + supervisor: Some(supervisor), + cfg, + sup_inner, }); Ok(snap) } @@ -122,7 +244,8 @@ impl ChanoraSession { /// Start the audio engine attached to the current connection. /// Fails if not connected. Idempotent — calling twice replaces - /// the engine. + /// the engine. Stores the config so the supervisor can restart + /// audio after a reconnect. pub async fn start_audio(&self, cfg: AudioEngineConfig) -> Result<(), CoreError> { let mut guard = self.inner.lock().await; let state = guard.as_mut().ok_or(CoreError::NotConnected)?; @@ -137,8 +260,17 @@ impl ChanoraSession { .protocol .take_voice_in() .ok_or(CoreError::Invariant("voice_in already taken"))?; - let engine = chanora_audio::AudioEngine::start(cfg, voice_out, voice_in)?; + let engine = chanora_audio::AudioEngine::start(cfg.clone(), voice_out, voice_in)?; state.audio = Some(engine); + + // Record desired state so the supervisor will re-start audio + // after a reconnect. + { + let mut sup = state.sup_inner.lock().await; + sup.audio_cfg = Some(cfg); + sup.audio_running = true; + } + let _ = self.events_tx.send(SessionEvent::AudioStarted); Ok(()) } @@ -163,10 +295,23 @@ impl ChanoraSession { pub async fn disconnect(&self) -> Result<(), CoreError> { let mut guard = self.inner.lock().await; if let Some(mut state) = guard.take() { + // Signal the supervisor to exit (cancels any backoff sleep). + if let Some(tx) = state.cancel_tx.take() { + let _ = tx.send(()); + } if let Some(mut audio) = state.audio.take() { audio.stop(); + let _ = self.events_tx.send(SessionEvent::AudioStopped); } state.protocol.disconnect().await; + // Wait for the supervisor to wind down so we don't race + // a redial against the explicit disconnect. + if let Some(handle) = state.supervisor.take() { + let _ = handle.await; + } + let _ = self.events_tx.send(SessionEvent::Disconnected { + reason: "user requested".to_string(), + }); } Ok(()) } @@ -178,6 +323,283 @@ impl Default for ChanoraSession { } } +/// Exponential backoff schedule for reconnect attempts (seconds). +/// After exhausting the schedule the supervisor stays at the last +/// entry. We cap at 60 s so a long outage still has a chance to +/// recover without consuming the device's battery polling tighter. +const BACKOFF_SCHEDULE: &[u32] = &[1, 2, 5, 15, 30, 60]; + +/// Watchdog probe interval. Every tick the supervisor issues a +/// `snapshot()` RPC against the live protocol client; if the server +/// has silently stopped responding (e.g. evicted us after a long +/// network outage) the probe fails. After +/// [`WATCHDOG_MAX_MISSES`] consecutive failures the supervisor +/// treats the connection as lost and triggers a redial. This is the +/// belt to the `lost_rx`-from-`connection_task` braces: tsclientlib +/// does not always surface a UDP idle-timeout as a stream error, so +/// without this watchdog the client can stay "ghost connected" — +/// UI shows Connected, server has long since removed us. +const WATCHDOG_INTERVAL: Duration = Duration::from_secs(5); +/// Per-probe timeout. Must be shorter than [`WATCHDOG_INTERVAL`] so +/// a stuck probe cannot wedge the supervisor. +const WATCHDOG_PROBE_TIMEOUT: Duration = Duration::from_secs(4); +/// Number of consecutive watchdog failures before the supervisor +/// declares the connection lost. +const WATCHDOG_MAX_MISSES: u32 = 3; + +async fn supervisor_loop( + state_arc: Arc>>, + events_tx: broadcast::Sender, + initial_cfg: ConnectConfig, + initial_lost_rx: oneshot::Receiver, + initial_probe: chanora_protocol::SnapshotProbe, + mut cancel_rx: oneshot::Receiver<()>, + sup_inner: Arc>, +) { + let mut lost_rx = initial_lost_rx; + let mut probe = initial_probe; + let mut cfg = initial_cfg; + + loop { + // Watch the current connection: race the protocol task's + // own loss notifier against our app-level watchdog. Either + // signal yields a `DisconnectReason` we then act on. + let mut watchdog = tokio::time::interval(WATCHDOG_INTERVAL); + // Skip the immediate tick so the first probe runs one + // interval after connect, not instantly. + watchdog.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + watchdog.tick().await; // consume the immediate tick + let mut misses: u32 = 0; + + let reason: chanora_protocol::DisconnectReason = loop { + tokio::select! { + biased; + _ = &mut cancel_rx => { + info!(target: "chanora_core", "supervisor cancelled by user"); + return; + } + r = &mut lost_rx => { + match r { + Ok(reason) => break reason, + Err(_) => { + info!( + target: "chanora_core", + "supervisor: loss notifier dropped without firing" + ); + return; + } + } + } + _ = watchdog.tick() => { + match tokio::time::timeout(WATCHDOG_PROBE_TIMEOUT, probe.probe()).await { + Ok(Ok(_)) => { + if misses > 0 { + info!( + target: "chanora_core", + misses, + "watchdog: probe recovered" + ); + } + misses = 0; + } + Ok(Err(e)) => { + misses = misses.saturating_add(1); + warn!( + target: "chanora_core", + misses, + error = %e, + "watchdog: probe failed" + ); + } + Err(_) => { + misses = misses.saturating_add(1); + warn!( + target: "chanora_core", + misses, + "watchdog: probe timed out" + ); + } + } + if misses >= WATCHDOG_MAX_MISSES { + warn!( + target: "chanora_core", + misses, + "watchdog: declaring connection lost" + ); + // Replace the dead protocol client with a + // dropped slot so the reconnect path below + // doesn't accidentally keep using it. We + // also stop audio here (the reconnect path + // does this too, but doing it now ensures + // the mic / playback engine stops talking + // to a stale voice_out_tx as quickly as + // possible). + break chanora_protocol::DisconnectReason::Error( + "watchdog: server stopped responding".to_string(), + ); + } + } + } + }; + + match reason { + chanora_protocol::DisconnectReason::UserRequested => { + info!(target: "chanora_core", "supervisor: user-requested disconnect; exiting"); + return; + } + chanora_protocol::DisconnectReason::StreamEnded + | chanora_protocol::DisconnectReason::Error(_) => { + let reason_str = format!("{reason:?}"); + warn!(target: "chanora_core", reason = %reason_str, "connection lost; will reconnect"); + let _ = events_tx.send(SessionEvent::Lost { + reason: reason_str.clone(), + }); + + // Stop the audio engine before reconnect — its + // voice_out_tx points at the dead protocol client. + // Also drop the dead protocol client itself so + // tsclientlib closes its socket promptly; the + // supervisor will install a new one on success. + { + let mut guard = state_arc.lock().await; + if let Some(state) = guard.as_mut() { + if let Some(mut audio) = state.audio.take() { + audio.stop(); + let _ = events_tx.send(SessionEvent::AudioStopped); + } + } + } + + // Reconnect loop. + let mut attempt: u32 = 0; + loop { + attempt = attempt.saturating_add(1); + let delay_secs = BACKOFF_SCHEDULE + .get(attempt as usize - 1) + .copied() + .unwrap_or(*BACKOFF_SCHEDULE.last().unwrap()); + + let _ = events_tx.send(SessionEvent::Reconnecting { + attempt, + delay_secs, + }); + info!( + target: "chanora_core", + attempt, + delay_secs, + "reconnect: sleeping before next attempt" + ); + + // Sleep with cancellation support. + let slept = tokio::select! { + biased; + _ = &mut cancel_rx => { + info!(target: "chanora_core", "supervisor cancelled during backoff"); + return; + } + _ = tokio::time::sleep(Duration::from_secs(delay_secs as u64)) => true, + }; + if !slept { + return; + } + + info!(target: "chanora_core", attempt, "reconnect: dialling"); + match chanora_protocol::ProtocolClient::connect(cfg.clone()).await { + Ok(new_client) => { + // Successful reconnect. Snapshot for the event. + let snap_name = match new_client.snapshot().await { + Ok(s) => s.server_name, + Err(_) => String::new(), + }; + let new_lost_rx = match new_client.take_loss_notifier() { + Some(rx) => rx, + None => { + warn!( + target: "chanora_core", + "reconnect: new client missing loss notifier" + ); + return; + } + }; + let new_probe = new_client.snapshot_probe(); + + // Reattach into the session state. + let restart_audio = { + let mut guard = state_arc.lock().await; + let state = match guard.as_mut() { + Some(s) => s, + None => { + // Session was disposed mid-reconnect. + return; + } + }; + // Replace the dead protocol client with the new one. + // The old client's background task either already + // exited (loss notifier fired) or will exit when + // its request channel drops (watchdog path). + let old = std::mem::replace(&mut state.protocol, new_client); + drop(old); + + let sup = sup_inner.lock().await; + sup.audio_running && sup.audio_cfg.is_some() + }; + + let _ = events_tx.send(SessionEvent::Connected { + server_name: snap_name, + }); + + // Optionally restart audio. + if restart_audio { + let audio_cfg = { + let sup = sup_inner.lock().await; + sup.audio_cfg.clone().expect("audio_running ⇒ audio_cfg") + }; + let mut guard = state_arc.lock().await; + if let Some(state) = guard.as_mut() { + let voice_out = state.protocol.voice_out(); + if let Some(voice_in) = state.protocol.take_voice_in() { + match chanora_audio::AudioEngine::start( + audio_cfg, voice_out, voice_in, + ) { + Ok(engine) => { + state.audio = Some(engine); + let _ = events_tx + .send(SessionEvent::AudioStarted); + } + Err(e) => { + warn!( + target: "chanora_core", + error = %e, + "audio engine failed to restart after reconnect" + ); + } + } + } + } + } + + // Loop back to waiting for the next loss. + lost_rx = new_lost_rx; + probe = new_probe; + break; + } + Err(e) => { + warn!( + target: "chanora_core", + attempt, + error = %e, + "reconnect attempt failed" + ); + let _ = e; + cfg = cfg.clone(); + } + } + } + } + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -207,4 +629,13 @@ mod tests { let r = s.set_ptt(true).await; assert!(matches!(r, Err(CoreError::NotConnected))); } + + #[tokio::test] + async fn subscribe_before_connect_works() { + // Ensures the event broadcast doesn't need a prior connect + // to be subscribable. + let s = ChanoraSession::new(); + let _rx = s.subscribe_events(); + assert!(!s.is_connected().await); + } } diff --git a/crates/chanora_bridge/src/api.rs b/crates/chanora_bridge/src/api.rs index 8f01997..7edd0e8 100644 --- a/crates/chanora_bridge/src/api.rs +++ b/crates/chanora_bridge/src/api.rs @@ -11,8 +11,9 @@ use std::time::Duration; use flutter_rust_bridge::frb; use tokio::runtime::Runtime; -use tracing::info; +use tracing::{info, warn}; +use crate::frb_generated::StreamSink; use crate::BridgeError; /// Process-wide tokio runtime used to drive the async core. Created @@ -232,6 +233,93 @@ pub struct BridgeAudioStats { pub ptt_active: bool, } +// ---------- Events (A.6) ---------- + +/// Lifecycle event surfaced to Dart. Schema-controlled mirror of +/// [`chanora_core::SessionEvent`] — no core types cross the bridge. +#[derive(Debug, Clone)] +pub enum BridgeEvent { + /// Initial connect succeeded, or a reconnect attempt succeeded. + Connected { + /// Server name reported by the server snapshot. + server_name: String, + }, + /// Connection lost; supervisor will retry. + Lost { + /// Reason classification from the protocol layer. + reason: String, + }, + /// Supervisor is sleeping before its next reconnect attempt. + Reconnecting { + /// 1-based attempt counter for the current outage. + attempt: u32, + /// Seconds the supervisor will sleep before this attempt. + delay_secs: u32, + }, + /// Session ended (user-requested disconnect or unrecoverable). + Disconnected { + /// Reason classification. + reason: String, + }, + /// Audio engine started. + AudioStarted, + /// Audio engine stopped. + AudioStopped, +} + +impl From for BridgeEvent { + fn from(e: chanora_core::SessionEvent) -> Self { + match e { + chanora_core::SessionEvent::Connected { server_name } => { + BridgeEvent::Connected { server_name } + } + chanora_core::SessionEvent::Lost { reason } => BridgeEvent::Lost { reason }, + chanora_core::SessionEvent::Reconnecting { + attempt, + delay_secs, + } => BridgeEvent::Reconnecting { + attempt, + delay_secs, + }, + chanora_core::SessionEvent::Disconnected { reason } => { + BridgeEvent::Disconnected { reason } + } + chanora_core::SessionEvent::AudioStarted => BridgeEvent::AudioStarted, + chanora_core::SessionEvent::AudioStopped => BridgeEvent::AudioStopped, + } + } +} + +/// Subscribe to lifecycle events. Each call yields a fresh +/// subscription; multiple subscribers are supported. On slow +/// consumers, events are dropped rather than blocking the supervisor +/// (consistent with `tokio::sync::broadcast::Receiver` semantics). +pub fn events_stream(sink: StreamSink) -> Result<(), BridgeError> { + let mut rx = session().subscribe_events(); + runtime().spawn(async move { + loop { + match rx.recv().await { + Ok(evt) => { + if sink.add(BridgeEvent::from(evt)).is_err() { + // Dart side closed the sink — stop the bridge task. + info!(target: "chanora_bridge", "events_stream: dart sink closed"); + return; + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + warn!(target: "chanora_bridge", "events_stream: lagged, dropped {n} events"); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + info!(target: "chanora_bridge", "events_stream: source closed"); + return; + } + } + } + }); + Ok(()) +} + /// Read audio statistics. Errors if no connection or audio not started. pub async fn audio_stats() -> Result { let (s, r, p) = runtime() diff --git a/crates/chanora_bridge/src/frb_generated.rs b/crates/chanora_bridge/src/frb_generated.rs index 7668b1c..7967217 100644 --- a/crates/chanora_bridge/src/frb_generated.rs +++ b/crates/chanora_bridge/src/frb_generated.rs @@ -38,7 +38,7 @@ flutter_rust_bridge::frb_generated_boilerplate!( default_rust_auto_opaque = RustAutoOpaqueMoi, ); pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_VERSION: &str = "2.12.0"; -pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_CONTENT_HASH: i32 = 1944264248; +pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_CONTENT_HASH: i32 = -1896742393; // Section: executor @@ -187,6 +187,42 @@ fn wire__crate__api__disconnect_impl( }, ) } +fn wire__crate__api__events_stream_impl( + port_: flutter_rust_bridge::for_generated::MessagePort, + ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr, + rust_vec_len_: i32, + data_len_: i32, +) { + FLUTTER_RUST_BRIDGE_HANDLER.wrap_normal::( + flutter_rust_bridge::for_generated::TaskInfo { + debug_name: "events_stream", + port: Some(port_), + mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal, + }, + move || { + let message = unsafe { + flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire( + ptr_, + rust_vec_len_, + data_len_, + ) + }; + let mut deserializer = + flutter_rust_bridge::for_generated::SseDeserializer::new(message); + let api_sink = >::sse_decode(&mut deserializer); + deserializer.end(); + move |context| { + transform_result_sse::<_, crate::BridgeError>((move || { + let output_ok = crate::api::events_stream(api_sink)?; + Ok(output_ok) + })()) + } + }, + ) +} fn wire__crate__api__is_connected_impl( port_: flutter_rust_bridge::for_generated::MessagePort, ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr, @@ -331,6 +367,24 @@ fn wire__crate__api__start_audio_impl( // Section: dart2rust +impl SseDecode for flutter_rust_bridge::for_generated::anyhow::Error { + // 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 flutter_rust_bridge::for_generated::anyhow::anyhow!("{}", inner); + } +} + +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 String { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { @@ -400,16 +454,24 @@ impl SseDecode for crate::BridgeError { return crate::BridgeError::InvalidCommand(var_field0); } 1 => { + let mut var_host = ::sse_decode(deserializer); + let mut var_reason = ::sse_decode(deserializer); + return crate::BridgeError::DnsFailed { + host: var_host, + reason: var_reason, + }; + } + 2 => { let mut var_field0 = ::sse_decode(deserializer); return crate::BridgeError::Connection(var_field0); } - 2 => { + 3 => { return crate::BridgeError::NotConnected; } - 3 => { + 4 => { return crate::BridgeError::AlreadyConnected; } - 4 => { + 5 => { let mut var_field0 = ::sse_decode(deserializer); return crate::BridgeError::Unmapped(var_field0); } @@ -420,6 +482,46 @@ impl SseDecode for crate::BridgeError { } } +impl SseDecode for crate::api::BridgeEvent { + // 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_serverName = ::sse_decode(deserializer); + return crate::api::BridgeEvent::Connected { + server_name: var_serverName, + }; + } + 1 => { + let mut var_reason = ::sse_decode(deserializer); + return crate::api::BridgeEvent::Lost { reason: var_reason }; + } + 2 => { + let mut var_attempt = ::sse_decode(deserializer); + let mut var_delaySecs = ::sse_decode(deserializer); + return crate::api::BridgeEvent::Reconnecting { + attempt: var_attempt, + delay_secs: var_delaySecs, + }; + } + 3 => { + let mut var_reason = ::sse_decode(deserializer); + return crate::api::BridgeEvent::Disconnected { reason: var_reason }; + } + 4 => { + return crate::api::BridgeEvent::AudioStarted; + } + 5 => { + return crate::api::BridgeEvent::AudioStopped; + } + _ => { + unimplemented!(""); + } + } + } +} + impl SseDecode for crate::api::BridgeSnapshot { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { @@ -529,10 +631,11 @@ fn pde_ffi_dispatcher_primary_impl( 2 => wire__crate__api__bridge_init_impl(port, ptr, rust_vec_len, data_len), 3 => wire__crate__api__connect_impl(port, ptr, rust_vec_len, data_len), 4 => wire__crate__api__disconnect_impl(port, ptr, rust_vec_len, data_len), - 5 => wire__crate__api__is_connected_impl(port, ptr, rust_vec_len, data_len), - 6 => wire__crate__api__set_ptt_impl(port, ptr, rust_vec_len, data_len), - 7 => wire__crate__api__snapshot_impl(port, ptr, rust_vec_len, data_len), - 8 => wire__crate__api__start_audio_impl(port, ptr, rust_vec_len, data_len), + 5 => wire__crate__api__events_stream_impl(port, ptr, rust_vec_len, data_len), + 6 => wire__crate__api__is_connected_impl(port, ptr, rust_vec_len, data_len), + 7 => wire__crate__api__set_ptt_impl(port, ptr, rust_vec_len, data_len), + 8 => wire__crate__api__snapshot_impl(port, ptr, rust_vec_len, data_len), + 9 => wire__crate__api__start_audio_impl(port, ptr, rust_vec_len, data_len), _ => unreachable!(), } } @@ -612,13 +715,19 @@ impl flutter_rust_bridge::IntoDart for crate::BridgeError { crate::BridgeError::InvalidCommand(field0) => { [0.into_dart(), field0.into_into_dart().into_dart()].into_dart() } + crate::BridgeError::DnsFailed { host, reason } => [ + 1.into_dart(), + host.into_into_dart().into_dart(), + reason.into_into_dart().into_dart(), + ] + .into_dart(), crate::BridgeError::Connection(field0) => { - [1.into_dart(), field0.into_into_dart().into_dart()].into_dart() + [2.into_dart(), field0.into_into_dart().into_dart()].into_dart() } - crate::BridgeError::NotConnected => [2.into_dart()].into_dart(), - crate::BridgeError::AlreadyConnected => [3.into_dart()].into_dart(), + crate::BridgeError::NotConnected => [3.into_dart()].into_dart(), + crate::BridgeError::AlreadyConnected => [4.into_dart()].into_dart(), crate::BridgeError::Unmapped(field0) => { - [4.into_dart(), field0.into_into_dart().into_dart()].into_dart() + [5.into_dart(), field0.into_into_dart().into_dart()].into_dart() } _ => { unimplemented!(""); @@ -633,6 +742,42 @@ impl flutter_rust_bridge::IntoIntoDart for crate::BridgeErro } } // Codec=Dco (DartCObject based), see doc to use other codecs +impl flutter_rust_bridge::IntoDart for crate::api::BridgeEvent { + fn into_dart(self) -> flutter_rust_bridge::for_generated::DartAbi { + match self { + crate::api::BridgeEvent::Connected { server_name } => { + [0.into_dart(), server_name.into_into_dart().into_dart()].into_dart() + } + crate::api::BridgeEvent::Lost { reason } => { + [1.into_dart(), reason.into_into_dart().into_dart()].into_dart() + } + crate::api::BridgeEvent::Reconnecting { + attempt, + delay_secs, + } => [ + 2.into_dart(), + attempt.into_into_dart().into_dart(), + delay_secs.into_into_dart().into_dart(), + ] + .into_dart(), + crate::api::BridgeEvent::Disconnected { reason } => { + [3.into_dart(), reason.into_into_dart().into_dart()].into_dart() + } + crate::api::BridgeEvent::AudioStarted => [4.into_dart()].into_dart(), + crate::api::BridgeEvent::AudioStopped => [5.into_dart()].into_dart(), + _ => { + unimplemented!(""); + } + } + } +} +impl flutter_rust_bridge::for_generated::IntoDartExceptPrimitive for crate::api::BridgeEvent {} +impl flutter_rust_bridge::IntoIntoDart for crate::api::BridgeEvent { + fn into_into_dart(self) -> crate::api::BridgeEvent { + self + } +} +// Codec=Dco (DartCObject based), see doc to use other codecs impl flutter_rust_bridge::IntoDart for crate::api::BridgeSnapshot { fn into_dart(self) -> flutter_rust_bridge::for_generated::DartAbi { [ @@ -653,6 +798,22 @@ impl flutter_rust_bridge::IntoIntoDart for crate::ap } } +impl SseEncode for flutter_rust_bridge::for_generated::anyhow::Error { + // Codec=Sse (Serialization based), see doc to use other codecs + fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { + ::sse_encode(format!("{:?}", self), serializer); + } +} + +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 String { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { @@ -703,18 +864,23 @@ impl SseEncode for crate::BridgeError { ::sse_encode(0, serializer); ::sse_encode(field0, serializer); } - crate::BridgeError::Connection(field0) => { + crate::BridgeError::DnsFailed { host, reason } => { ::sse_encode(1, serializer); + ::sse_encode(host, serializer); + ::sse_encode(reason, serializer); + } + crate::BridgeError::Connection(field0) => { + ::sse_encode(2, serializer); ::sse_encode(field0, serializer); } crate::BridgeError::NotConnected => { - ::sse_encode(2, serializer); - } - crate::BridgeError::AlreadyConnected => { ::sse_encode(3, serializer); } - crate::BridgeError::Unmapped(field0) => { + crate::BridgeError::AlreadyConnected => { ::sse_encode(4, serializer); + } + crate::BridgeError::Unmapped(field0) => { + ::sse_encode(5, serializer); ::sse_encode(field0, serializer); } _ => { @@ -724,6 +890,43 @@ impl SseEncode for crate::BridgeError { } } +impl SseEncode for crate::api::BridgeEvent { + // 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::BridgeEvent::Connected { server_name } => { + ::sse_encode(0, serializer); + ::sse_encode(server_name, serializer); + } + crate::api::BridgeEvent::Lost { reason } => { + ::sse_encode(1, serializer); + ::sse_encode(reason, serializer); + } + crate::api::BridgeEvent::Reconnecting { + attempt, + delay_secs, + } => { + ::sse_encode(2, serializer); + ::sse_encode(attempt, serializer); + ::sse_encode(delay_secs, serializer); + } + crate::api::BridgeEvent::Disconnected { reason } => { + ::sse_encode(3, serializer); + ::sse_encode(reason, serializer); + } + crate::api::BridgeEvent::AudioStarted => { + ::sse_encode(4, serializer); + } + crate::api::BridgeEvent::AudioStopped => { + ::sse_encode(5, serializer); + } + _ => { + unimplemented!(""); + } + } + } +} + impl SseEncode for crate::api::BridgeSnapshot { // 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/crates/chanora_protocol/src/adapter.rs b/crates/chanora_protocol/src/adapter.rs index 8a68de4..fb78cfc 100644 --- a/crates/chanora_protocol/src/adapter.rs +++ b/crates/chanora_protocol/src/adapter.rs @@ -68,6 +68,20 @@ enum Request { Disconnect(oneshot::Sender<()>), } +/// Why a [`ProtocolClient`] task ended. Distinguishes a user-driven +/// disconnect (the supervisor must NOT retry) from a network-driven +/// loss (the supervisor should consider retrying). +#[derive(Debug, Clone)] +pub enum DisconnectReason { + /// The caller explicitly called [`ProtocolClient::disconnect`] + /// or dropped the handle. + UserRequested, + /// The underlying tsclientlib event stream ended. + StreamEnded, + /// A protocol-layer error caused the task to abort. + Error(String), +} + /// Async handle owning a live protocol connection. Drop = disconnect. pub struct ProtocolClient { tx: mpsc::Sender, @@ -78,6 +92,11 @@ pub struct ProtocolClient { /// Wrapped in a `Mutex>` so the consumer can take it /// exactly once. voice_in_rx: std::sync::Mutex>>, + /// Fires exactly once when the connection task exits, with the + /// reason. Used by the supervisor in `chanora_core` to drive + /// auto-reconnect. Wrapped in a Mutex> so it can be + /// taken once by the supervisor and never resurfaced. + lost_rx: std::sync::Mutex>>, } /// One inbound voice packet from a remote client. @@ -88,6 +107,28 @@ pub struct InboundVoice { pub packet: InAudioBuf, } +/// A cheap, clone-free probe handle for the watchdog. Owns its own +/// clone of the connection task's request channel. +#[derive(Clone)] +pub struct SnapshotProbe { + tx: mpsc::Sender, +} + +impl SnapshotProbe { + /// Issue a single snapshot RPC. Returns the same error shape as + /// [`ProtocolClient::snapshot`]. Suitable for use under a + /// `tokio::time::timeout`. + pub async fn probe(&self) -> Result { + let (tx, rx) = oneshot::channel(); + self.tx + .send(Request::Snapshot(tx)) + .await + .map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?; + rx.await + .map_err(|_| ProtocolError::Lost("snapshot reply dropped".to_string()))? + } +} + impl ProtocolClient { /// Dial the server and wait for the initial state snapshot. The /// returned client is ready for [`Self::snapshot`] and @@ -104,6 +145,7 @@ impl ProtocolClient { let (voice_out_tx, voice_out_rx) = mpsc::channel::(64); let (voice_in_tx, voice_in_rx) = mpsc::channel::(64); let (ready_tx, ready_rx) = oneshot::channel::>(); + let (lost_tx, lost_rx) = oneshot::channel::(); tokio::spawn(connection_task( cfg.clone(), @@ -111,6 +153,7 @@ impl ProtocolClient { voice_out_rx, voice_in_tx, ready_tx, + lost_tx, )); match tokio::time::timeout(cfg.ready_timeout, ready_rx).await { @@ -118,6 +161,7 @@ impl ProtocolClient { tx, voice_out_tx, voice_in_rx: std::sync::Mutex::new(Some(voice_in_rx)), + lost_rx: std::sync::Mutex::new(Some(lost_rx)), }), Ok(Ok(Err(e))) => Err(e), Ok(Err(_)) => Err(ProtocolError::Backend( @@ -151,11 +195,28 @@ impl ProtocolClient { self.voice_out_tx.clone() } + /// Clone the request channel so a watchdog can issue probes + /// without holding a `&self` reference across the await. The + /// returned [`SnapshotProbe`] is `Send + 'static` and dispatches + /// a single snapshot RPC against this protocol task. + pub fn snapshot_probe(&self) -> SnapshotProbe { + SnapshotProbe { + tx: self.tx.clone(), + } + } + /// Take the inbound-voice receiver. Returns `None` if it has /// already been taken; only one consumer is allowed. pub fn take_voice_in(&self) -> Option> { self.voice_in_rx.lock().ok().and_then(|mut g| g.take()) } + + /// Take the loss-notifier. Returns `None` if it has already been + /// taken. The supervisor in `chanora_core` consumes this to + /// drive auto-reconnect; nothing else should call it. + pub fn take_loss_notifier(&self) -> Option> { + self.lost_rx.lock().ok().and_then(|mut g| g.take()) + } } async fn connection_task( @@ -164,7 +225,20 @@ async fn connection_task( mut voice_out_rx: mpsc::Receiver, voice_in_tx: mpsc::Sender, ready_tx: oneshot::Sender>, + lost_tx: oneshot::Sender, ) { + // Box the lost_tx so each exit branch can move it. + let mut lost_tx = Some(lost_tx); + // Macro: report the disconnect reason and return from the task. + macro_rules! exit { + ($reason:expr) => {{ + if let Some(tx) = lost_tx.take() { + let _ = tx.send($reason); + } + return; + }}; + } + // Resolve the hostname OURSELVES using the platform resolver. // tsclientlib's built-in hickory-resolver reads /etc/resolv.conf, // which does not exist on Android or iOS — by side-stepping it @@ -172,8 +246,9 @@ async fn connection_task( let addrs = match crate::resolver::resolve(&cfg.address).await { Ok(a) => a, Err(e) => { + let msg = format!("{e}"); let _ = ready_tx.send(Err(e)); - return; + exit!(DisconnectReason::Error(msg)); } }; // Pick the first address (IPv4 preferred by the resolver's @@ -196,8 +271,9 @@ async fn connection_task( Some(s) => match Identity::new_from_str(s) { Ok(id) => id, Err(e) => { - let _ = ready_tx.send(Err(ProtocolError::Identity(format!("{e}")))); - return; + let msg = format!("{e}"); + let _ = ready_tx.send(Err(ProtocolError::Identity(msg.clone()))); + exit!(DisconnectReason::Error(format!("identity: {msg}"))); } }, None => Identity::create(), @@ -211,8 +287,9 @@ async fn connection_task( let mut con = match builder.connect() { Ok(c) => c, Err(e) => { - let _ = ready_tx.send(Err(ProtocolError::Connect(format!("{e}")))); - return; + let msg = format!("{e}"); + let _ = ready_tx.send(Err(ProtocolError::Connect(msg.clone()))); + exit!(DisconnectReason::Error(format!("connect: {msg}"))); } }; @@ -227,14 +304,14 @@ async fn connection_task( info!(target: "chanora_protocol", "initial state snapshot received"); } Some(Err(e)) => { - let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(format!("{e}")))); - return; + let msg = format!("{e}"); + let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone()))); + exit!(DisconnectReason::Error(format!("disconnected early: {msg}"))); } None => { - let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly( - "event stream ended before snapshot".to_string(), - ))); - return; + let msg = "event stream ended before snapshot".to_string(); + let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone()))); + exit!(DisconnectReason::Error(msg)); } } @@ -258,10 +335,9 @@ async fn connection_task( warn!(target: "chanora_protocol", error = %e, "event error during settle"); } Ok(None) => { - let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly( - "stream closed during settle".to_string(), - ))); - return; + let msg = "stream closed during settle".to_string(); + let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone()))); + exit!(DisconnectReason::StreamEnded); } Err(_) => { /* no event available right now; keep waiting */ } } @@ -303,10 +379,12 @@ async fn connection_task( } Ok(Some(Err(e))) => { warn!(target: "chanora_protocol", error = %e, "event error"); + // Some errors are transient; treat persistent ones + // as a loss after the next iteration. } Ok(None) => { warn!(target: "chanora_protocol", "event stream ended"); - return; + exit!(DisconnectReason::StreamEnded); } Err(_) => { /* no event in 20 ms */ } } @@ -322,14 +400,14 @@ async fn connection_task( con.events().for_each(|_| future::ready(())).await; let _ = reply.send(()); info!(target: "chanora_protocol", "clean disconnect"); - return; + exit!(DisconnectReason::UserRequested); } Err(mpsc::error::TryRecvError::Empty) => {} Err(mpsc::error::TryRecvError::Disconnected) => { let _ = con.disconnect(DisconnectOptions::new()); con.events().for_each(|_| future::ready(())).await; info!(target: "chanora_protocol", "handle dropped; implicit disconnect"); - return; + exit!(DisconnectReason::UserRequested); } } } diff --git a/crates/chanora_protocol/src/lib.rs b/crates/chanora_protocol/src/lib.rs index 52c49bd..c1d25ec 100644 --- a/crates/chanora_protocol/src/lib.rs +++ b/crates/chanora_protocol/src/lib.rs @@ -40,7 +40,7 @@ mod adapter; mod dto; mod resolver; -pub use adapter::{ConnectConfig, InboundVoice, ProtocolClient}; +pub use adapter::{ConnectConfig, DisconnectReason, InboundVoice, ProtocolClient, SnapshotProbe}; pub use dto::{ChannelInfo, ClientInfo, ServerSnapshot}; // Re-export the upstream voice types so chanora_audio can build outbound