Initial commit
This commit is contained in:
gasaichandesu
@@ -0,0 +1,106 @@
|
||||
import 'dart:async';
|
||||
import 'package:injectable/injectable.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/download_exceptions.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/download_progress.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/download_request.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/use_cases/download_object_use_case.dart';
|
||||
|
||||
typedef DownloadId = String;
|
||||
|
||||
class _Session {
|
||||
_Session({required this.id, required this.request, required this.useCase})
|
||||
: progressController = StreamController<DownloadProgress>.broadcast();
|
||||
|
||||
final DownloadId id;
|
||||
final DownloadRequest request;
|
||||
final DownloadObjectUseCase useCase;
|
||||
|
||||
final StreamController<DownloadProgress> progressController;
|
||||
StreamSubscription<DownloadProgress>? sub;
|
||||
bool paused = false;
|
||||
bool cancelled = false;
|
||||
|
||||
Stream<DownloadProgress> get stream => progressController.stream;
|
||||
|
||||
Future<void> start() async {
|
||||
cancelled = false;
|
||||
paused = false;
|
||||
sub = useCase
|
||||
.call(
|
||||
DownloadObjectUseCaseInput(
|
||||
request: request,
|
||||
isCancelled: () async => cancelled || paused,
|
||||
),
|
||||
)
|
||||
.listen(
|
||||
progressController.add,
|
||||
onError: progressController.addError,
|
||||
onDone: () => progressController.close(),
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> pause() async {
|
||||
paused = true;
|
||||
await sub?.cancel();
|
||||
sub = null;
|
||||
}
|
||||
|
||||
Future<void> resume() async {
|
||||
if (cancelled) return;
|
||||
paused = false;
|
||||
await start(); // will resume from file size
|
||||
}
|
||||
|
||||
Future<void> cancel() async {
|
||||
cancelled = true;
|
||||
await sub?.cancel();
|
||||
sub = null;
|
||||
if (!progressController.isClosed) {
|
||||
progressController.addError(const CancelledException());
|
||||
await progressController.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@lazySingleton
|
||||
class DownloadManager {
|
||||
DownloadManager(this._useCase);
|
||||
|
||||
final DownloadObjectUseCase _useCase;
|
||||
|
||||
final Map<DownloadId, _Session> _sessions = {};
|
||||
|
||||
/// Start or replace a session with [id].
|
||||
Stream<DownloadProgress> start({
|
||||
required DownloadId id,
|
||||
required DownloadRequest request,
|
||||
}) {
|
||||
// dispose old session if exists
|
||||
_sessions[id]?.cancel();
|
||||
final s = _Session(id: id, request: request, useCase: _useCase);
|
||||
_sessions[id] = s;
|
||||
unawaited(s.start());
|
||||
return s.stream;
|
||||
}
|
||||
|
||||
Stream<DownloadProgress>? progress(DownloadId id) => _sessions[id]?.stream;
|
||||
|
||||
Future<void> pause(DownloadId id) => _sessions[id]?.pause() ?? Future.value();
|
||||
Future<void> resume(DownloadId id) =>
|
||||
_sessions[id]?.resume() ?? Future.value();
|
||||
Future<void> cancel(DownloadId id) async {
|
||||
await _sessions[id]?.cancel();
|
||||
_sessions.remove(id);
|
||||
}
|
||||
|
||||
bool isActive(DownloadId id) => _sessions.containsKey(id);
|
||||
Iterable<DownloadId> activeIds() => _sessions.keys;
|
||||
|
||||
/// Cancel and clear all sessions (e.g., on sign-out).
|
||||
Future<void> cancelAll() async {
|
||||
for (final s in _sessions.values) {
|
||||
await s.cancel();
|
||||
}
|
||||
_sessions.clear();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
import 'dart:async';
|
||||
import 'dart:io';
|
||||
|
||||
import 'package:crypto/crypto.dart' as c;
|
||||
import 'package:injectable/injectable.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/repositories/file_system.dart';
|
||||
|
||||
@LazySingleton(as: FileSystem)
|
||||
class IoFileSystem implements FileSystem {
|
||||
@override
|
||||
Future<void> ensureParentExists(String path) =>
|
||||
File(path).parent.create(recursive: true);
|
||||
|
||||
@override
|
||||
Future<bool> exists(String path) => File(path).exists();
|
||||
|
||||
@override
|
||||
Future<int> sizeOf(String path) async => (await File(path).stat()).size;
|
||||
|
||||
@override
|
||||
Future<void> truncate(String path) async =>
|
||||
File(path).writeAsBytes(const [], flush: true);
|
||||
|
||||
@override
|
||||
Future<void> createEmpty(String path) => File(path).create(recursive: true);
|
||||
|
||||
@override
|
||||
Future<StreamSink<List<int>>> openAppend(String path) async =>
|
||||
File(path).openWrite(mode: FileMode.append);
|
||||
|
||||
@override
|
||||
Future<bool> verifyFile(
|
||||
String path, {
|
||||
required String expected,
|
||||
required String algo,
|
||||
}) async {
|
||||
final file = File(path);
|
||||
if (!await file.exists()) return false;
|
||||
|
||||
final digestHex = await _computeDigestHex(file, algo);
|
||||
return _constantTimeEquals(digestHex.toLowerCase(), expected.toLowerCase());
|
||||
}
|
||||
|
||||
Future<String> _computeDigestHex(File file, String algo) async {
|
||||
final hash = switch (algo.toLowerCase()) {
|
||||
'md5' => c.md5,
|
||||
'sha256' => c.sha256,
|
||||
_ => throw ArgumentError('Unsupported checksum algo: $algo'),
|
||||
};
|
||||
|
||||
// Stream the file into the hash converter; no extra buffers.
|
||||
final digest = await hash.bind(file.openRead()).first; // -> c.Digest
|
||||
return digest.toString(); // lowercase hex
|
||||
}
|
||||
|
||||
bool _constantTimeEquals(String a, String b) {
|
||||
if (a.length != b.length) return false;
|
||||
var result = 0;
|
||||
for (var i = 0; i < a.length; i++) {
|
||||
result |= a.codeUnitAt(i) ^ b.codeUnitAt(i);
|
||||
}
|
||||
return result == 0;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
import 'package:injectable/injectable.dart';
|
||||
import 'package:minio/minio.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/remote_object_meta.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/repositories/storage_reader.dart';
|
||||
|
||||
/// MinIO storage reader implementation.
|
||||
@LazySingleton(as: StorageReader)
|
||||
class MinioStorageReader implements StorageReader {
|
||||
final Minio _client;
|
||||
|
||||
MinioStorageReader(this._client);
|
||||
|
||||
@override
|
||||
Future<Stream<List<int>>> readFromOffset({
|
||||
required String bucket,
|
||||
required String key,
|
||||
required int startOffset,
|
||||
}) {
|
||||
return _client.getPartialObject(bucket, key, startOffset);
|
||||
}
|
||||
|
||||
@override
|
||||
Future<RemoteObjectMeta> stat({
|
||||
required String bucket,
|
||||
required String key,
|
||||
}) async {
|
||||
final s = await _client.statObject(bucket, key);
|
||||
return RemoteObjectMeta(size: s.size ?? 0, eTag: s.etag ?? '');
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
sealed class DownloadException implements Exception {
|
||||
final String message;
|
||||
|
||||
const DownloadException(this.message);
|
||||
|
||||
@override
|
||||
String toString() => '$runtimeType: $message';
|
||||
}
|
||||
|
||||
final class NetworkException extends DownloadException {
|
||||
const NetworkException(super.message);
|
||||
}
|
||||
|
||||
final class RemoteException extends DownloadException {
|
||||
const RemoteException(super.message);
|
||||
}
|
||||
|
||||
final class IoException extends DownloadException {
|
||||
const IoException(super.message);
|
||||
}
|
||||
|
||||
final class CancelledException extends DownloadException {
|
||||
const CancelledException() : super('Cancelled');
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
/// Represents the state of a download.
|
||||
final class DownloadProgress {
|
||||
/// Download speed in bytes per second.
|
||||
final int speed;
|
||||
|
||||
/// Total bytes downloaded so far.
|
||||
final int downloaded;
|
||||
|
||||
/// Total bytes to download.
|
||||
final int total;
|
||||
|
||||
/// Estimated time of arrival (ETA) for the download to complete.
|
||||
final Duration eta;
|
||||
|
||||
const DownloadProgress({
|
||||
required this.speed,
|
||||
required this.downloaded,
|
||||
required this.total,
|
||||
required this.eta,
|
||||
});
|
||||
|
||||
const DownloadProgress.initial()
|
||||
: speed = 0,
|
||||
downloaded = 0,
|
||||
total = 0,
|
||||
eta = Duration.zero;
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
/// Represents a request to download an object from a storage bucket.
|
||||
class DownloadRequest {
|
||||
/// Local file path to save the downloaded object
|
||||
final String destinationPath;
|
||||
|
||||
const DownloadRequest({required this.destinationPath});
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
/// Represents the result of a download operation.
|
||||
class DownloadResult {
|
||||
/// Local file path where the downloaded object is stored.
|
||||
final String path;
|
||||
|
||||
/// Number of bytes downloaded.
|
||||
final int bytes;
|
||||
|
||||
/// Entity tag (ETag) of the downloaded object.
|
||||
final String eTag;
|
||||
|
||||
const DownloadResult({
|
||||
required this.path,
|
||||
required this.bytes,
|
||||
required this.eTag,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
/// Represents metadata for a remote object in a storage bucket.
|
||||
class RemoteObjectMeta {
|
||||
/// Size of the object in bytes.
|
||||
final int size;
|
||||
|
||||
/// Entity tag (ETag) of the object.
|
||||
final String eTag;
|
||||
|
||||
const RemoteObjectMeta({required this.size, required this.eTag});
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
import 'dart:async';
|
||||
|
||||
/// Interface for interacting with the file system.
|
||||
abstract interface class FileSystem {
|
||||
/// Ensures that the parent directory of the given path exists.
|
||||
Future<void> ensureParentExists(String path);
|
||||
|
||||
/// Checks if a file or directory exists at the given path.
|
||||
Future<bool> exists(String path);
|
||||
|
||||
/// Returns the size of the file at the given path.
|
||||
Future<int> sizeOf(String path);
|
||||
|
||||
/// Truncates the file at the given path to zero length.
|
||||
Future<void> truncate(String path);
|
||||
|
||||
/// Creates an empty file at the given path.
|
||||
Future<void> createEmpty(String path);
|
||||
|
||||
/// Opens a stream sink for appending data to the file at the given path.
|
||||
Future<StreamSink<List<int>>> openAppend(String path);
|
||||
|
||||
/// Verifies the integrity of a file at the given path.
|
||||
Future<bool> verifyFile(
|
||||
String path, {
|
||||
required String expected,
|
||||
required String algo,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/remote_object_meta.dart';
|
||||
|
||||
/// Interface for reading data from storage.
|
||||
abstract interface class StorageReader {
|
||||
/// Retrieves metadata about a remote object.
|
||||
Future<RemoteObjectMeta> stat({required String bucket, required String key});
|
||||
|
||||
/// Reads data from a remote object starting at the specified offset.
|
||||
Future<Stream<List<int>>> readFromOffset({
|
||||
required String bucket,
|
||||
required String key,
|
||||
required int startOffset,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,160 @@
|
||||
import 'dart:async';
|
||||
|
||||
import 'package:injectable/injectable.dart';
|
||||
import 'package:moonwell_launcher/config.dart';
|
||||
import 'package:moonwell_launcher/core/use_case.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/download_exceptions.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/download_progress.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/download_request.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/entities/download_result.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/repositories/file_system.dart';
|
||||
import 'package:moonwell_launcher/features/downloader/domain/repositories/storage_reader.dart';
|
||||
|
||||
final class DownloadObjectUseCaseInput {
|
||||
final DownloadRequest request;
|
||||
|
||||
final FutureOr<bool> Function()? isCancelled;
|
||||
|
||||
final void Function(DownloadResult result)? onComplete;
|
||||
|
||||
const DownloadObjectUseCaseInput({
|
||||
required this.request,
|
||||
this.isCancelled,
|
||||
this.onComplete,
|
||||
});
|
||||
}
|
||||
|
||||
@lazySingleton
|
||||
class DownloadObjectUseCase
|
||||
implements StreamUseCase<DownloadObjectUseCaseInput, DownloadProgress> {
|
||||
final StorageReader _storage;
|
||||
|
||||
final FileSystem _fileSystem;
|
||||
|
||||
const DownloadObjectUseCase({
|
||||
required StorageReader storage,
|
||||
required FileSystem fileSystem,
|
||||
}) : _storage = storage,
|
||||
_fileSystem = fileSystem;
|
||||
|
||||
@override
|
||||
Stream<DownloadProgress> call(DownloadObjectUseCaseInput input) async* {
|
||||
final request = input.request;
|
||||
|
||||
final meta = await _storage.stat(
|
||||
bucket: Config.bucketName,
|
||||
key: Config.objectName,
|
||||
);
|
||||
final total = meta.size;
|
||||
|
||||
final pathToFile = '${request.destinationPath}/${Config.objectName}';
|
||||
|
||||
await _fileSystem.ensureParentExists(pathToFile);
|
||||
|
||||
int offset = 0;
|
||||
if (await _fileSystem.exists(pathToFile)) {
|
||||
offset = await _fileSystem.sizeOf(pathToFile);
|
||||
if (offset > total) {
|
||||
await _fileSystem.truncate(pathToFile);
|
||||
offset = 0;
|
||||
}
|
||||
|
||||
if (offset == total) {
|
||||
input.onComplete?.call(
|
||||
DownloadResult(path: pathToFile, bytes: offset, eTag: meta.eTag),
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
await _fileSystem.createEmpty(pathToFile);
|
||||
}
|
||||
|
||||
final objectStream = await _storage.readFromOffset(
|
||||
bucket: Config.bucketName,
|
||||
key: Config.objectName,
|
||||
startOffset: offset,
|
||||
);
|
||||
final sink = await _fileSystem.openAppend(pathToFile);
|
||||
final controller = StreamController<DownloadProgress>();
|
||||
|
||||
final sw = Stopwatch()..start();
|
||||
int lastBytes = 0;
|
||||
int downloaded = offset;
|
||||
|
||||
Timer? ticker;
|
||||
|
||||
void tick() {
|
||||
final elapsedMs = sw.elapsedMilliseconds;
|
||||
final bps = elapsedMs == 0 ? 0 : (lastBytes * 1000) ~/ elapsedMs;
|
||||
final remaining = (total - downloaded).clamp(0, total);
|
||||
final eta = bps > 0 ? Duration(seconds: remaining ~/ bps) : Duration.zero;
|
||||
|
||||
controller.add(
|
||||
DownloadProgress(
|
||||
downloaded: downloaded,
|
||||
total: total,
|
||||
speed: bps,
|
||||
eta: eta,
|
||||
),
|
||||
);
|
||||
lastBytes = 0;
|
||||
sw
|
||||
..reset()
|
||||
..start();
|
||||
}
|
||||
|
||||
ticker = Timer.periodic(const Duration(seconds: 1), (_) => tick());
|
||||
|
||||
late StreamSubscription<List<int>> sub;
|
||||
controller.onCancel = () async {
|
||||
await sub.cancel();
|
||||
await sink.close();
|
||||
ticker?.cancel();
|
||||
};
|
||||
|
||||
sub = objectStream.listen(
|
||||
(chunk) async {
|
||||
if (input.isCancelled != null && await input.isCancelled!.call()) {
|
||||
ticker?.cancel();
|
||||
await sub.cancel();
|
||||
await sink.close();
|
||||
controller.addError(const CancelledException());
|
||||
await controller.close();
|
||||
return;
|
||||
}
|
||||
sink.add(chunk);
|
||||
final n = chunk.length;
|
||||
downloaded += n;
|
||||
lastBytes += n;
|
||||
},
|
||||
onError: (e, st) async {
|
||||
ticker?.cancel();
|
||||
await sink.close();
|
||||
await controller.close();
|
||||
throw _mapError(e);
|
||||
},
|
||||
onDone: () async {
|
||||
ticker?.cancel();
|
||||
tick(); // final snapshot
|
||||
await sink.close();
|
||||
await controller.close();
|
||||
input.onComplete?.call(
|
||||
DownloadResult(path: pathToFile, bytes: downloaded, eTag: meta.eTag),
|
||||
);
|
||||
},
|
||||
cancelOnError: true,
|
||||
);
|
||||
|
||||
yield* controller.stream;
|
||||
}
|
||||
|
||||
DownloadException _mapError(Object e) {
|
||||
final s = e.toString();
|
||||
if (s.contains('SocketException') || s.contains('HandshakeException')) {
|
||||
return NetworkException(s);
|
||||
}
|
||||
if (s.contains('Minio')) return RemoteException(s);
|
||||
return IoException(s);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
import 'package:injectable/injectable.dart';
|
||||
import 'package:moonwell_launcher/features/preferences/domain/repositories/preferences_repository.dart';
|
||||
import 'package:shared_preferences/shared_preferences.dart';
|
||||
|
||||
const _outputDirKey = 'output_dir';
|
||||
|
||||
@LazySingleton(as: PreferencesRepository, env: ['flutter'])
|
||||
class SharedPrefsPreferencesRepository implements PreferencesRepository {
|
||||
final SharedPreferences _preferences;
|
||||
|
||||
const SharedPrefsPreferencesRepository({
|
||||
required SharedPreferences preferences,
|
||||
}) : _preferences = preferences;
|
||||
|
||||
@override
|
||||
Future<Uri?> getOutputDir() {
|
||||
final rawPath = _preferences.getString(_outputDirKey);
|
||||
|
||||
if (rawPath == null) {
|
||||
return Future.value(null);
|
||||
}
|
||||
|
||||
return Future.value(Uri.parse(rawPath));
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> setOutputDir(Uri uri) {
|
||||
return _preferences.setString(_outputDirKey, uri.toString());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
abstract interface class PreferencesRepository {
|
||||
Future<Uri?> getOutputDir();
|
||||
|
||||
Future<void> setOutputDir(Uri uri);
|
||||
}
|
||||
Reference in New Issue
Block a user