add worker abstract

Author: iamalper

integration_test/app_test.dart

@@ -107,7 +107,7 @@ void main() {
       downloadedFiles = [];
     });
     tearDownAll(() async {
-      await receiver?.stopListening();
+      await receiver?.stop();
     });
   });
 
@@ -134,7 +134,7 @@ void main() {
         devices = await Discover.discover();
       }
       expect(devices.where((device) => device.code == code), isNotEmpty);
-      Future.delayed(const Duration(milliseconds: 500), sender.cancel);
+      Future.delayed(const Duration(milliseconds: 500), sender.stop);
       await sender.send(
           devices.first,
           [

integration_test/mobile_test.dart

@@ -76,6 +76,6 @@ void main() {
     downloadedFiles = [];
   });
   tearDownAll(() async {
-    await receiver?.stopListening();
+    await receiver?.stop();
   });
 }

lib/classes/receiver.dart

@@ -19,7 +19,7 @@ import '../constants.dart';
 
 ///Class for all Recieve jobs
 ///
-///Available methods are [listen] and [stopListening]
+///Available methods are [listen] and [stop]
 class Receiver {
   final _files = <DbFile>[];
   late final _ms = MediaStore();
@@ -251,7 +251,7 @@ class Receiver {
   ///Closes the listening server.
   ///
   ///Is is safe to call before [listen] or after [listen] .
-  Future<void> stopListening() async => await _server?.close();
+  Future<void> stop() async => await _server?.close();
 
   Map<String, dynamic> get map => {
         "useDb": useDb,

lib/classes/sender.dart

@@ -26,7 +26,7 @@ class Sender {
     return result?.files;
   }
 
-  void cancel() {
+  void stop() {
     log("Request cancelled", name: "Sender");
     _senderCancelToken.cancel();
   }

lib/classes/workers/base_worker.dart

@@ -0,0 +1,23 @@
+import 'dart:isolate';
+
+///Base class for implementing workers communication.
+///
+///Worker classes meant to manage workers.
+///It run in main isolate, but workers run separated isolates.
+abstract class BaseWorker {
+  ///Whether if worker is active.
+  Future<bool> isActive();
+
+  ///Stop worker and clean all resources.
+  Future<void> stop();
+
+  ///Creates [ReceivePort] for worker to main isolate communication and registers.
+  ///
+  ///Calling more than one, throws.
+  ReceivePort getReceivePort();
+
+  ///Looks up for the [SendPort] registered for main isolate to worker communication.
+  ///
+  ///Call it every time you need the [SendPort], instead storing as variable.
+  SendPort? getSendPort();
+}

lib/classes/workers/isolated_receiver.dart

@@ -9,11 +9,15 @@ import 'package:weepy/classes/workers/worker_interface.dart';
 import 'package:weepy/classes/workers/worker_messages.dart' as messages;
 import 'package:workmanager/workmanager.dart';
 
-class IsolatedReceiver extends Receiver {
+import 'base_worker.dart';
+
+class IsolatedReceiver extends Receiver implements BaseWorker {
   ///If [true] creates and manages progress notification.
   bool progressNotification;
 
-  ///Runs [Receiver] from a worker.
+  //TODO: Test completer
+  final aliveCheckCompleter = Completer<bool>();
+
   IsolatedReceiver(
       {super.onDownloadUpdatePercent,
       super.useDb,
@@ -26,6 +30,25 @@ class IsolatedReceiver extends Receiver {
       super.code,
       this.progressNotification = true});
 
+  @override
+  Future<bool> isActive() async {
+    final sendPort = getSendPort();
+    if (sendPort == null) {
+      return false;
+    }
+    sendPort.send(const messages.Alive().map);
+    const waitDuration = Duration(seconds: 5);
+    final aliveCanceller = Future.delayed(
+      waitDuration,
+      () => aliveCheckCompleter.complete(false),
+    );
+    final alive = await aliveCheckCompleter.future;
+    if (alive) {
+      aliveCanceller.ignore();
+    }
+    return alive;
+  }
+
   ///Starts worker and runs [Receiver.listen]
   ///
   ///If necessary, requests permission. Throws [NoStoragePermissionException]
@@ -42,21 +65,19 @@ class IsolatedReceiver extends Receiver {
         throw NoStoragePermissionException();
       }
     }
-    final port = ReceivePort();
-    IsolateNameServer.registerPortWithName(port.sendPort, MyTasks.receive.name);
-    port.listen(_portCallback);
-    await workManager.registerOneOffTask(
-        MyTasks.receive.name, MyTasks.receive.name,
+
+    getReceivePort().listen(_portCallback);
+    await workManager.registerOneOffTask(Tasks.receive.name, Tasks.receive.name,
         inputData: super.map, existingWorkPolicy: ExistingWorkPolicy.keep);
     return super.code;
   }
 
   @override
-  Future<void> stopListening() async {
+  Future<void> stop() async {
     if (progressNotification) {
       await notifications.cancelDownload();
     }
-    await workManager.cancelByUniqueName(MyTasks.receive.name);
+    await workManager.cancelByUniqueName(Tasks.receive.name);
   }
 
   Future<void> _portCallback(data) async {
@@ -92,12 +113,30 @@ class IsolatedReceiver extends Receiver {
         case messages.MessageType.downloadStarted:
           final _ = messages.DownloadStarted.fromMap(data);
           super.onDownloadStart?.call();
+        case messages.MessageType.alive:
+          aliveCheckCompleter.complete(true);
         default:
           throw Error();
       }
     } on Exception catch (e) {
-      log("Interface error", name: "IsolatedReceiver", error: e);
+      log("Error", name: "IsolatedReceiver", error: e);
       rethrow;
     }
   }
+
+  @override
+  ReceivePort getReceivePort() {
+    final receivePort = ReceivePort();
+    final isRegistered = IsolateNameServer.registerPortWithName(
+        receivePort.sendPort, PortNames.receiver2main.name);
+    assert(isRegistered);
+    return receivePort;
+  }
+
+  @override
+  SendPort? getSendPort() {
+    final sendPort =
+        IsolateNameServer.lookupPortByName(PortNames.main2receiver.name);
+    return sendPort;
+  }
 }

lib/classes/workers/isolated_sender.dart

@@ -1,25 +1,41 @@
 import 'dart:async';
+import 'dart:isolate';
+import 'dart:ui';
 import 'package:file_picker/file_picker.dart';
 import 'package:weepy/classes/notifications.dart' as notifications;
 import 'package:weepy/classes/sender.dart';
-import 'package:weepy/classes/workers/isolated_receiver.dart';
 import 'package:weepy/classes/workers/worker_interface.dart';
 import 'package:weepy/classes/workers/worker_messages.dart' as messages;
 import 'package:weepy/models.dart';
+import 'base_worker.dart';
 
-class IsolatedSender extends Sender {
+class IsolatedSender extends Sender implements BaseWorker {
   IsolatedSender({super.onUploadProgress});
-
-  ///Returns whether [IsolatedReceiver] active in a worker.
+  //TODO: Test completer
+  final aliveCheckCompleter = Completer<bool>();
+  @override
   Future<bool> isActive() async {
-    //TODO
-    throw UnimplementedError();
+    final sendPort = getSendPort();
+    if (sendPort == null) {
+      return false;
+    }
+    sendPort.send(const messages.Alive().map);
+    const waitDuration = Duration(seconds: 5);
+    final aliveCanceller = Future.delayed(
+      waitDuration,
+      () => aliveCheckCompleter.complete(false),
+    );
+    final alive = await aliveCheckCompleter.future;
+    if (alive) {
+      aliveCanceller.ignore();
+    }
+    return alive;
   }
 
   @override
   Future<void> send(Device device, Iterable<PlatformFile> files,
       {bool useDb = true, bool progressNotification = true}) async {
-    final port = await initialize();
+    await initialize();
     if (progressNotification) {
       progressNotification = await notifications.initialize();
     }
@@ -31,7 +47,8 @@ class IsolatedSender extends Sender {
     map["useDb"] = useDb;
 
     final exitBlock = Completer<void>();
-    port.listen((data) async {
+    getReceivePort().listen((data) async {
+      //Listen incoming messages
       switch (messages.MessageType.values[data["type"]]) {
         case messages.MessageType.updatePercent:
           final message = messages.UpdatePercent.fromMap(data);
@@ -43,11 +60,13 @@ class IsolatedSender extends Sender {
         case messages.MessageType.completed:
           final _ = messages.Completed.fromMap(data);
           exitBlock.complete();
+        case messages.MessageType.alive:
+          aliveCheckCompleter.complete(true);
         default:
           throw Error();
       }
     });
-    await workManager.registerOneOffTask(MyTasks.send.name, MyTasks.send.name,
+    await workManager.registerOneOffTask(Tasks.send.name, Tasks.send.name,
         inputData: map);
     if (progressNotification) {
       //Create notification
@@ -60,5 +79,21 @@ class IsolatedSender extends Sender {
   }
 
   @override
-  Future<void> cancel() => workManager.cancelByUniqueName(MyTasks.send.name);
+  Future<void> stop() => workManager.cancelByUniqueName(Tasks.send.name);
+
+  @override
+  ReceivePort getReceivePort() {
+    final receivePort = ReceivePort();
+    final isRegistered = IsolateNameServer.registerPortWithName(
+        receivePort.sendPort, PortNames.sender2main.name);
+    assert(isRegistered);
+    return receivePort;
+  }
+
+  @override
+  SendPort? getSendPort() {
+    final sendPort =
+        IsolateNameServer.lookupPortByName(PortNames.main2sender.name);
+    return sendPort;
+  }
 }

lib/classes/workers/worker_interface.dart

@@ -1,3 +1,8 @@
+///This library manages essentials for all `worker`'s
+///
+///We are using `worker`'s from workmanager plugin to continue jobs from background
+library;
+
 import 'dart:async';
 import 'dart:developer';
 import 'dart:io';
@@ -15,25 +20,25 @@ import 'package:workmanager/workmanager.dart';
 
 import '../receiver.dart';
 
-enum MyTasks { receive, send }
+enum PortNames { receiver2main, sender2main, main2receiver, main2sender }
 
-enum PortName { main2receiver, main2sender }
+enum Tasks { receive, send }
 
 final workManager = Workmanager();
 @pragma("vm:entry-point")
 void _callBack() {
   workManager.executeTask((taskName, inputData) async {
     try {
       final task =
-          MyTasks.values.singleWhere((element) => element.name == taskName);
+          Tasks.values.singleWhere((element) => element.name == taskName);
       switch (task) {
-        case MyTasks.receive:
+        case Tasks.receive:
           SendPort? getSendPort() =>
-              IsolateNameServer.lookupPortByName(MyTasks.receive.name);
+              IsolateNameServer.lookupPortByName(PortNames.receiver2main.name);
 
           final receiverPort = ReceivePort();
           IsolateNameServer.registerPortWithName(
-              receiverPort.sendPort, PortName.main2receiver.name);
+              receiverPort.sendPort, PortNames.main2receiver.name);
           receiverPort.listen((message) {
             //TODO: Test alive feature
             if (message["data"] == messages.MessageType.alive) {
@@ -67,13 +72,13 @@ void _callBack() {
           });
           await receiver.listen();
           return exitBlock.future;
-        case MyTasks.send:
+        case Tasks.send:
           final senderMap = inputData!;
           final receiverPort = ReceivePort();
           IsolateNameServer.registerPortWithName(
-              receiverPort.sendPort, PortName.main2sender.name);
+              receiverPort.sendPort, PortNames.main2sender.name);
           SendPort? getSendPort() =>
-              IsolateNameServer.lookupPortByName(MyTasks.send.name);
+              IsolateNameServer.lookupPortByName(PortNames.sender2main.name);
           receiverPort.listen((message) {
             if (message["data"] == messages.MessageType.alive) {
               //TODO: Test alive feature
@@ -104,13 +109,10 @@ void _callBack() {
       }
     } on Exception {
       rethrow;
-    
+    }
   });
 }
 
-Future<ReceivePort> initialize() async {
-  final port = ReceivePort();
-  IsolateNameServer.registerPortWithName(port.sendPort, MyTasks.send.name);
+Future<void> initialize() async {
   await workManager.initialize(_callBack, isInDebugMode: kDebugMode);
-  return port;
 }

lib/screens/receive_page.dart

@@ -98,7 +98,7 @@ class _ReceivePageInnerState extends ConsumerState<ReceivePageInner>
   @override
   void dispose() {
     _downloadAnimC.dispose();
-    _receiver.stopListening();
+    _receiver.stop();
     super.dispose();
   }
 

lib/screens/send_page.dart

@@ -96,7 +96,7 @@ class _SendPageInnerState extends ConsumerState<SendPageInner>
   @override
   void dispose() {
     _uploadAnimC.dispose();
-    _sender.cancel();
+    _sender.stop();
     super.dispose();
   }