Fix: worker tasks parallel execution

Previously on native platforms `CombineWorker` didn't payed attantion to the `tasksPerIsolate` parameter and executed them sequentially

.vscode/settings.json

@@ -6,5 +6,5 @@
   },
   "dart.lineLength": 80,
   "markdown.extension.toc.levels": "2..4",
-  "dart.flutterSdkPath": "/Users/maksimzemlyanikin/fvm/versions/3.7.0"
+  "dart.flutterSdkPath": "/Users/max/fvm/versions/3.10.4"
 }

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 0.5.3
+- Fix `CombineWorker` parallel tasks execution.
+
 ## 0.5.2
 - Fix Isolate spawning from non UI Isolate.
 

analysis_options.yaml

@@ -21,6 +21,7 @@ linter:
     omit_local_variable_types: true
     always_declare_return_types: true
     # Custom disabled rules
+    avoid_redundant_argument_values: false
     prefer_asserts_with_message: false
     unnecessary_lambdas: false
     discarded_futures: false

example/macos/Podfile.lock

@@ -6,18 +6,18 @@ PODS:
 
 DEPENDENCIES:
   - FlutterMacOS (from `Flutter/ephemeral`)
-  - shared_preferences_foundation (from `Flutter/ephemeral/.symlinks/plugins/shared_preferences_foundation/macos`)
+  - shared_preferences_foundation (from `Flutter/ephemeral/.symlinks/plugins/shared_preferences_foundation/darwin`)
 
 EXTERNAL SOURCES:
   FlutterMacOS:
     :path: Flutter/ephemeral
   shared_preferences_foundation:
-    :path: Flutter/ephemeral/.symlinks/plugins/shared_preferences_foundation/macos
+    :path: Flutter/ephemeral/.symlinks/plugins/shared_preferences_foundation/darwin
 
 SPEC CHECKSUMS:
   FlutterMacOS: 8f6f14fa908a6fb3fba0cd85dbd81ec4b251fb24
   shared_preferences_foundation: 297b3ebca31b34ec92be11acd7fb0ba932c822ca
 
 PODFILE CHECKSUM: 353c8bcc5d5b0994e508d035b5431cfe18c1dea7
 
-COCOAPODS: 1.11.3
+COCOAPODS: 1.12.1

example/pubspec.lock

@@ -5,10 +5,10 @@ packages:
     dependency: transitive
     description:
       name: async
-      sha256: bfe67ef28df125b7dddcea62755991f807aa39a2492a23e1550161692950bbe0
+      sha256: "947bfcf187f74dbc5e146c9eb9c0f10c9f8b30743e341481c1e2ed3ecc18c20c"
       url: "https://pub.dev"
     source: hosted
-    version: "2.10.0"
+    version: "2.11.0"
   boolean_selector:
     dependency: transitive
     description:
@@ -21,10 +21,10 @@ packages:
     dependency: transitive
     description:
       name: characters
-      sha256: e6a326c8af69605aec75ed6c187d06b349707a27fbff8222ca9cc2cff167975c
+      sha256: "04a925763edad70e8443c99234dc3328f442e811f1d8fd1a72f1c8ad0f69a605"
       url: "https://pub.dev"
     source: hosted
-    version: "1.2.1"
+    version: "1.3.0"
   clock:
     dependency: transitive
     description:
@@ -37,17 +37,17 @@ packages:
     dependency: transitive
     description:
       name: collection
-      sha256: cfc915e6923fe5ce6e153b0723c753045de46de1b4d63771530504004a45fae0
+      sha256: "4a07be6cb69c84d677a6c3096fcf960cc3285a8330b4603e0d463d15d9bd934c"
       url: "https://pub.dev"
     source: hosted
-    version: "1.17.0"
+    version: "1.17.1"
   combine:
     dependency: "direct main"
     description:
       path: ".."
       relative: true
     source: path
-    version: "0.5.1"
+    version: "0.5.3"
   cupertino_icons:
     dependency: "direct main"
     description:
@@ -107,10 +107,10 @@ packages:
     dependency: transitive
     description:
       name: js
-      sha256: "5528c2f391ededb7775ec1daa69e65a2d61276f7552de2b5f7b8d34ee9fd4ab7"
+      sha256: f2c445dce49627136094980615a031419f7f3eb393237e4ecd97ac15dea343f3
       url: "https://pub.dev"
     source: hosted
-    version: "0.6.5"
+    version: "0.6.7"
   lints:
     dependency: transitive
     description:
@@ -123,10 +123,10 @@ packages:
     dependency: transitive
     description:
       name: matcher
-      sha256: "16db949ceee371e9b99d22f88fa3a73c4e59fd0afed0bd25fc336eb76c198b72"
+      sha256: "6501fbd55da300384b768785b83e5ce66991266cec21af89ab9ae7f5ce1c4cbb"
       url: "https://pub.dev"
     source: hosted
-    version: "0.12.13"
+    version: "0.12.15"
   material_color_utilities:
     dependency: transitive
     description:
@@ -139,18 +139,18 @@ packages:
     dependency: transitive
     description:
       name: meta
-      sha256: "6c268b42ed578a53088d834796959e4a1814b5e9e164f147f580a386e5decf42"
+      sha256: "3c74dbf8763d36539f114c799d8a2d87343b5067e9d796ca22b5eb8437090ee3"
       url: "https://pub.dev"
     source: hosted
-    version: "1.8.0"
+    version: "1.9.1"
   path:
     dependency: transitive
     description:
       name: path
-      sha256: db9d4f58c908a4ba5953fcee2ae317c94889433e5024c27ce74a37f94267945b
+      sha256: "8829d8a55c13fc0e37127c29fedf290c102f4e40ae94ada574091fe0ff96c917"
       url: "https://pub.dev"
     source: hosted
-    version: "1.8.2"
+    version: "1.8.3"
   path_provider_linux:
     dependency: transitive
     description:
@@ -304,10 +304,10 @@ packages:
     dependency: transitive
     description:
       name: test_api
-      sha256: ad540f65f92caa91bf21dfc8ffb8c589d6e4dc0c2267818b4cc2792857706206
+      sha256: eb6ac1540b26de412b3403a163d919ba86f6a973fe6cc50ae3541b80092fdcfb
       url: "https://pub.dev"
     source: hosted
-    version: "0.4.16"
+    version: "0.5.1"
   vector_math:
     dependency: transitive
     description:
@@ -333,5 +333,5 @@ packages:
     source: hosted
     version: "0.2.0+3"
 sdks:
-  dart: ">=2.18.0 <3.0.0"
+  dart: ">=3.0.0-0 <4.0.0"
   flutter: ">=3.7.0"

lib/src/combine_worker/combine_task_executor.dart

@@ -85,7 +85,7 @@ class CombineTaskExecutor {
     if (initializer is WorkerInitializer) {
       await initializer();
     }
-    await for (final request in messenger.messages) {
+    messenger.messages.listen((request) async {
       if (request is _ExecutableTaskRequest) {
         late TaskResponse taskResponse;
         try {
@@ -104,7 +104,7 @@ class CombineTaskExecutor {
           );
         }
       }
-    }
+    });
   }
 }
 

pubspec.yaml

@@ -1,7 +1,7 @@
 name: combine
 description: A Flutter package which allows you to work with
   MethodChannels in Isolate and provides simplified Isolate and Thread Pool API.
-version: 0.5.2
+version: 0.5.3
 homepage: https://github.com/Maksimka101/combine
 repository: https://github.com/Maksimka101/combine
 
@@ -18,6 +18,6 @@ dev_dependencies:
     sdk: flutter
   all_lint_rules_community: ^0.0.9
   mocktail: ^0.3.0
-  meta: ^1.7.0
+  meta: ^1.9.1
 
 flutter:

test/combine_spawners/counter_combine_spawners.dart

@@ -34,7 +34,11 @@ Future<CombineInfo> spawnSimpleCounterIsolateInsideAnotherIsolate() {
 
 Future<CombineInfo> spawnMethodChannelCounterIsolate() {
   var counter = 0;
-  _counterMethodChannel.setMockMethodCallHandler((call) async => ++counter);
+  TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+      .setMockMethodCallHandler(
+    _counterMethodChannel,
+    (call) async => ++counter,
+  );
 
   return Combine().spawn((context) {
     context.messenger.messages.listen((event) async {
@@ -60,8 +64,10 @@ Future<CombineInfo> checkMethodChannelInIsolateIsInitialized() {
 
 Future<CombineInfo> spawnComplexMethodChannelCounterIsolate() {
   var counter = 0;
-  _counterMethodChannel.setMockMethodCallHandler((call) {
-    _counterMethodChannel.binaryMessenger.handlePlatformMessage(
+  TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+      .setMockMethodCallHandler(_counterMethodChannel, (call) {
+    TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
+        .handlePlatformMessage(
       "counter",
       const StandardMethodCodec().encodeMethodCall(
         MethodCall("counter", ++counter),

test/worker_tasks/parallel_tasks.dart

@@ -0,0 +1,9 @@
+bool? _flag;
+const setFlagDelay = Duration(milliseconds: 100);
+Future<void> setWaitAndRemoveFlag() async {
+  _flag = true;
+  await Future.delayed(setFlagDelay);
+  _flag = false;
+}
+
+Future<bool?> getFlagValue() async => _flag;

test/worker_test.dart

@@ -9,13 +9,14 @@ import 'utils/test_async_widgets.dart';
 import 'worker_tasks/initializer_tasks.dart';
 import 'worker_tasks/no_args_tasks.dart';
 import 'worker_tasks/one_arg_tasks.dart';
+import 'worker_tasks/parallel_tasks.dart';
 import 'worker_tasks/two_args_tasks.dart';
 
 void main() {
   late CombineWorker combineWorker;
 
   setUp(() {
-    combineWorker = CombineWorkerImpl();
+    combineWorker = CombineWorker.newInstance();
   });
   tearDown(() => combineWorker.close());
 
@@ -144,6 +145,32 @@ void main() {
     testAsyncWidgets('test isolate prefix', (_) async {
       await combineWorker.initialize(isolatesPrefix: 'prefix-test');
     });
+
+    group("Test 'tasksPerIsolate'", () {
+      test("tasks are executed in parallel", () async {
+        await combineWorker.initialize(
+          isolatesCount: 1,
+          tasksPerIsolate: 2,
+        );
+        final updateFlagFuture = combineWorker.execute(setWaitAndRemoveFlag);
+        await Future.delayed(setFlagDelay ~/ 2);
+        final flagValue = await combineWorker.execute(getFlagValue);
+        await updateFlagFuture;
+        expect(flagValue, isTrue);
+      });
+
+      test("tasks are executed in sequence", () async {
+        await combineWorker.initialize(
+          isolatesCount: 1,
+          tasksPerIsolate: 1,
+        );
+        final updateFlagFuture = combineWorker.execute(setWaitAndRemoveFlag);
+        await Future.delayed(setFlagDelay ~/ 2);
+        final flagValue = await combineWorker.execute(getFlagValue);
+        await updateFlagFuture;
+        expect(flagValue, isFalse);
+      });
+    });
   });
 
   group("Test with web worker factory", () {