ihompadmin/tasq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Per channel skeleton

Browse Source
This commit is contained in:
Marc Rejohn Castillano
2026-03-01 20:10:38 +08:00
parent b9153a070f
commit 029e671367
13 changed files with 380 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/providers/stream_recovery.dart
+214 -88
View File
@@ -1,4 +1,5 @@
import 'dart:async';
import 'dart:math' as math;
import 'package:flutter/foundation.dart';
/// Connection status for a single stream subscription.
@@ -58,17 +59,37 @@ class StreamRecoveryConfig {
});
}
/// Callback type for per-channel status change notifications.
///
/// Used by [StreamRecoveryWrapper] to notify [RealtimeController] about
/// individual channel recovery state so the UI can show per-channel skeletons.
typedef ChannelStatusCallback =
void Function(String channel, StreamConnectionStatus status);
/// Wraps a Supabase realtime stream with automatic recovery, polling fallback,
/// and connection status tracking. Provides graceful degradation when the
/// realtime connection fails.
///
/// Error handling:
/// - **Timeout**: detected and handled internally with exponential backoff.
/// - **ChannelRateLimitReached**: detected and handled with a longer minimum
/// delay (5 s) before retrying. During recovery, a REST poll keeps data
/// fresh so the UI shows shimmer/skeleton instead of an error.
/// - **Generic errors**: same recovery flow with standard backoff.
///
/// Errors are **never** forwarded to consumers; instead the wrapper emits
/// [StreamRecoveryResult] events with appropriate [connectionStatus] so the
/// UI can react per-channel (e.g. show skeleton shimmer).
///
/// Usage:
/// ```dart
/// final wrappedStream = StreamRecoveryWrapper(
/// final wrapper = StreamRecoveryWrapper<Task>(
/// stream: client.from('tasks').stream(primaryKey: ['id']),
/// onPollData: () => fetchTasksViaRest(),
/// fromMap: Task.fromMap,
/// channelName: 'tasks',
/// onStatusChanged: (channel, status) { ... },
/// );
/// // wrappedStream.stream emits data with connection status in metadata
/// ```
class StreamRecoveryWrapper<T> {
final Stream<List<Map<String, dynamic>>> _realtimeStream;
@@ -76,140 +97,245 @@ class StreamRecoveryWrapper<T> {
final T Function(Map<String, dynamic>) _fromMap;
final StreamRecoveryConfig _config;
/// Human-readable channel name for logging and per-channel status tracking.
final String channelName;
/// Optional callback invoked whenever this channel's connection status
/// changes. Used to integrate with [RealtimeController] for per-channel
/// skeleton indicators in the UI.
final ChannelStatusCallback? _onStatusChanged;
StreamConnectionStatus _connectionStatus = StreamConnectionStatus.connected;
int _recoveryAttempts = 0;
Timer? _pollingTimer;
StreamController<StreamConnectionStatus>? _statusController;
Stream<StreamRecoveryResult<T>>? _cachedStream;
Timer? _recoveryTimer;
StreamSubscription<List<Map<String, dynamic>>>? _realtimeSub;
StreamController<StreamRecoveryResult<T>>? _controller;
bool _disposed = false;
bool _listening = false;
StreamRecoveryWrapper({
required Stream<List<Map<String, dynamic>>> stream,
required Future<List<T>> Function() onPollData,
required T Function(Map<String, dynamic>) fromMap,
StreamRecoveryConfig config = const StreamRecoveryConfig(),
this.channelName = 'unknown',
ChannelStatusCallback? onStatusChanged,
}) : _realtimeStream = stream,
_onPollData = onPollData,
_fromMap = fromMap,
_config = config;
_config = config,
_onStatusChanged = onStatusChanged;
/// The wrapped stream that emits recovery results with metadata.
Stream<StreamRecoveryResult<T>> get stream =>
_cachedStream ??= _buildStream();
///
/// Lazily initializes the internal controller and starts listening to the
/// realtime stream on first access. Errors from the realtime channel are
/// handled internally — consumers only see [StreamRecoveryResult] events.
Stream<StreamRecoveryResult<T>> get stream {
if (_controller == null) {
_controller = StreamController<StreamRecoveryResult<T>>.broadcast();
_startRealtimeSubscription();
}
return _controller!.stream;
}
/// Current connection status of this stream.
StreamConnectionStatus get connectionStatus => _connectionStatus;
/// Notifies listeners when connection status changes.
Stream<StreamConnectionStatus> get statusChanges {
_statusController ??= StreamController<StreamConnectionStatus>.broadcast();
return _statusController!.stream;
// ── Realtime subscription ───────────────────────────────────────────────
void _startRealtimeSubscription() {
if (_disposed || _listening) return;
_listening = true;
_realtimeSub?.cancel();
_realtimeSub = _realtimeStream.listen(
_onRealtimeData,
onError: _onRealtimeError,
onDone: _onRealtimeDone,
cancelOnError: false, // keep listening even after transient errors
);
}
/// Builds the wrapped stream with recovery and polling logic.
Stream<StreamRecoveryResult<T>> _buildStream() async* {
int delayMs = _config.initialDelayMs;
void _onRealtimeData(List<Map<String, dynamic>> rows) {
if (_disposed) return;
// Successful data — reset recovery counters.
_recoveryAttempts = 0;
_setStatus(StreamConnectionStatus.connected);
_emit(
StreamRecoveryResult<T>(
data: rows.map(_fromMap).toList(),
connectionStatus: StreamConnectionStatus.connected,
isStale: false,
),
);
}
while (true) {
try {
_setStatus(StreamConnectionStatus.connected);
void _onRealtimeError(Object error, [StackTrace? stack]) {
if (_disposed) return;
// Try realtime stream first
yield* _realtimeStream
.map(
(rows) => StreamRecoveryResult<T>(
data: rows.map(_fromMap).toList(),
connectionStatus: StreamConnectionStatus.connected,
isStale: false,
),
)
.handleError((error) {
debugPrint(
'StreamRecoveryWrapper: realtime stream error: $error',
);
_setStatus(StreamConnectionStatus.recovering);
throw error; // Propagate to outer handler
});
final isRateLimit = _isRateLimitError(error);
final isTimeout = _isTimeoutError(error);
final tag = isRateLimit
? ' (rate-limited)'
: isTimeout
? ' (timeout)'
: '';
// If we get here, stream completed normally (shouldn't happen)
break;
} catch (e) {
debugPrint(
'StreamRecoveryWrapper: realtime failed, error=$e, '
'attempts=$_recoveryAttempts/${_config.maxRecoveryAttempts}',
debugPrint('StreamRecoveryWrapper[$channelName]: stream error$tag: $error');
_setStatus(StreamConnectionStatus.recovering);
if (_recoveryAttempts >= _config.maxRecoveryAttempts) {
if (_config.enablePollingFallback) {
_startPollingFallback();
} else {
_setStatus(StreamConnectionStatus.failed);
_emit(
StreamRecoveryResult<T>(
data: const [],
connectionStatus: StreamConnectionStatus.failed,
isStale: true,
error: error.toString(),
),
);
// Exceeded max recovery attempts?
if (_recoveryAttempts >= _config.maxRecoveryAttempts) {
if (_config.enablePollingFallback) {
_setStatus(StreamConnectionStatus.polling);
yield* _pollingFallback();
break;
} else {
_setStatus(StreamConnectionStatus.failed);
yield StreamRecoveryResult<T>(
data: const [],
connectionStatus: StreamConnectionStatus.failed,
isStale: true,
error: e.toString(),
);
break;
}
}
// Exponential backoff before retry
_recoveryAttempts++;
await Future.delayed(Duration(milliseconds: delayMs));
delayMs = (delayMs * _config.backoffMultiplier).toInt();
if (delayMs > _config.maxDelayMs) {
delayMs = _config.maxDelayMs;
}
}
return;
}
_recoveryAttempts++;
// Compute backoff delay. Rate-limit errors get a longer floor (5 s).
final baseDelay =
_config.initialDelayMs *
math.pow(_config.backoffMultiplier, _recoveryAttempts - 1);
final effectiveDelay = isRateLimit
? math.max(baseDelay.toInt(), 5000)
: baseDelay.toInt();
final cappedDelay = math.min(effectiveDelay, _config.maxDelayMs);
debugPrint(
'StreamRecoveryWrapper[$channelName]: recovery attempt '
'$_recoveryAttempts/${_config.maxRecoveryAttempts}, '
'delay=${cappedDelay}ms$tag',
);
// Fire a single REST poll immediately so the UI can show fresh data
// under the skeleton shimmer while waiting for realtime to reconnect.
_pollOnce();
// Schedule re-subscription after backoff.
_recoveryTimer?.cancel();
_recoveryTimer = Timer(Duration(milliseconds: cappedDelay), () {
if (_disposed) return;
_listening = false;
_startRealtimeSubscription();
});
}
void _onRealtimeDone() {
if (_disposed) return;
debugPrint('StreamRecoveryWrapper[$channelName]: stream completed');
// Attempt to reconnect once if the stream closes unexpectedly.
if (_recoveryAttempts < _config.maxRecoveryAttempts) {
_onRealtimeError(StateError('realtime stream completed unexpectedly'));
}
}
/// Fallback to periodic REST polling when realtime is unavailable.
Stream<StreamRecoveryResult<T>> _pollingFallback() async* {
while (_connectionStatus == StreamConnectionStatus.polling) {
try {
final data = await _onPollData();
yield StreamRecoveryResult<T>(
// ── Polling fallback ──────────────────────────────────────────────────
void _startPollingFallback() {
_realtimeSub?.cancel();
_listening = false;
_setStatus(StreamConnectionStatus.polling);
_pollingTimer?.cancel();
_pollOnce(); // Immediate first poll
_pollingTimer = Timer.periodic(
Duration(milliseconds: _config.pollingIntervalMs),
(_) => _pollOnce(),
);
}
Future<void> _pollOnce() async {
if (_disposed) return;
try {
final data = await _onPollData();
_emit(
StreamRecoveryResult<T>(
data: data,
connectionStatus: StreamConnectionStatus.polling,
isStale: true, // Mark as stale since it's no longer live
);
await Future.delayed(Duration(milliseconds: _config.pollingIntervalMs));
} catch (e) {
debugPrint('StreamRecoveryWrapper: polling error: $e');
_setStatus(StreamConnectionStatus.stale);
yield StreamRecoveryResult<T>(
data: const [],
connectionStatus: StreamConnectionStatus.stale,
connectionStatus: _connectionStatus,
isStale: true,
error: e.toString(),
),
);
} catch (e) {
debugPrint('StreamRecoveryWrapper[$channelName]: poll error: $e');
if (_connectionStatus == StreamConnectionStatus.polling) {
_setStatus(StreamConnectionStatus.stale);
_emit(
StreamRecoveryResult<T>(
data: const [],
connectionStatus: StreamConnectionStatus.stale,
isStale: true,
error: e.toString(),
),
);
break;
}
}
}
/// Update connection status and notify listeners.
// ── Error classification ──────────────────────────────────────────────
/// Whether [error] indicates a Supabase channel rate limit.
static bool _isRateLimitError(Object error) {
final msg = error.toString().toLowerCase();
return msg.contains('rate limit') ||
msg.contains('rate_limit') ||
msg.contains('channelratelimitreached') ||
msg.contains('too many') ||
msg.contains('429');
}
/// Whether [error] indicates a subscription timeout.
static bool _isTimeoutError(Object error) {
if (error is TimeoutException) return true;
final msg = error.toString().toLowerCase();
return msg.contains('timeout') ||
msg.contains('timed out') ||
msg.contains('timed_out');
}
// ── Helpers ───────────────────────────────────────────────────────────
void _emit(StreamRecoveryResult<T> result) {
if (!_disposed && _controller != null && !_controller!.isClosed) {
_controller!.add(result);
}
}
/// Update connection status and notify the per-channel callback.
void _setStatus(StreamConnectionStatus status) {
if (_connectionStatus != status) {
_connectionStatus = status;
_statusController?.add(status);
_onStatusChanged?.call(channelName, status);
}
}
/// Manually trigger a recovery attempt.
void retry() {
_recoveryAttempts = 0;
_setStatus(StreamConnectionStatus.recovering);
_pollingTimer?.cancel();
_recoveryTimer?.cancel();
_listening = false;
_startRealtimeSubscription();
}
/// Clean up resources.
/// Clean up all resources.
void dispose() {
_disposed = true;
_pollingTimer?.cancel();
_statusController?.close();
_recoveryTimer?.cancel();
_realtimeSub?.cancel();
_controller?.close();
}
}