Remove unnnecessary `show Future` from `dart:async` imports.
Based on https://github.com/dart-lang/isolate/pull/41
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
new file mode 100644
index 0000000..430a85e
--- /dev/null
+++ b/.github/dependabot.yml
@@ -0,0 +1,11 @@
+# Set update schedule for GitHub Actions
+# See https://docs.github.com/en/free-pro-team@latest/github/administering-a-repository/keeping-your-actions-up-to-date-with-dependabot
+
+version: 2
+updates:
+
+- package-ecosystem: "github-actions"
+ directory: "/"
+ schedule:
+ # Check for updates to GitHub Actions every weekday
+ interval: "daily"
diff --git a/.github/workflows/test-package.yml b/.github/workflows/test-package.yml
new file mode 100644
index 0000000..9e93f1b
--- /dev/null
+++ b/.github/workflows/test-package.yml
@@ -0,0 +1,61 @@
+name: Dart CI
+
+on:
+ # Run on PRs and pushes to the default branch.
+ push:
+ branches: [ master ]
+ pull_request:
+ branches: [ master ]
+ schedule:
+ - cron: "0 0 * * 0"
+
+env:
+ PUB_ENVIRONMENT: bot.github
+
+jobs:
+ # Check code formatting and static analysis on a single OS (linux)
+ # against Dart dev and stable.
+ analyze:
+ runs-on: ubuntu-latest
+ strategy:
+ fail-fast: false
+ matrix:
+ sdk: [dev]
+ steps:
+ - uses: actions/checkout@v2
+ - uses: dart-lang/setup-dart@v1
+ with:
+ sdk: ${{ matrix.sdk }}
+ - id: install
+ name: Install dependencies
+ run: dart pub get
+ - name: Check formatting
+ run: dart format --output=none --set-exit-if-changed .
+ if: always() && steps.install.outcome == 'success'
+ - name: Analyze code
+ run: dart analyze --fatal-infos
+ if: always() && steps.install.outcome == 'success'
+
+ # Run tests on a matrix consisting of two dimensions:
+ # 1. OS: ubuntu-latest, (macos-latest, windows-latest)
+ # 2. release channel: dev, stable
+ test:
+ needs: analyze
+ runs-on: ${{ matrix.os }}
+ strategy:
+ fail-fast: false
+ matrix:
+ # Add macos-latest and/or windows-latest if relevant for this package.
+ os: [ubuntu-latest]
+ sdk: [dev, 2.12.0]
+ steps:
+ - uses: actions/checkout@v2
+ - uses: dart-lang/setup-dart@v1
+ with:
+ sdk: ${{ matrix.sdk }}
+ - id: install
+ name: Install dependencies
+ run: dart pub get
+ - name: Run VM tests
+ run: dart test --platform vm
+ if: always() && steps.install.outcome == 'success'
diff --git a/.status b/.status
deleted file mode 100644
index a69ed9e..0000000
--- a/.status
+++ /dev/null
@@ -1,11 +0,0 @@
-# Copyright (c) 2015, the Dart project authors. Please see the AUTHORS file
-# for details. All rights reserved. Use of this source code is governed by a
-# BSD-style license that can be found in the LICENSE file.
-
-[ $runtime == vm ]
-test/isolaterunner_test: RuntimeError # addOnExitListener not implemented.
-
-[ $compiler == dart2js ]
-test/registry_test: CompileTimeError # Unimplemented: private symbol literals.
-# FunctionRef will be removed when the VM supports sending functions.
-test/isolaterunner_test: RuntimeError # Mirrors not working - FunctionRef broken.
\ No newline at end of file
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 90eeb66..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,22 +0,0 @@
-language: dart
-dart:
- - stable
- - dev
-
-dart_task:
- - test
- - dartanalyzer: --fatal-infos --fatal-warnings .
-
-matrix:
- include:
- # Only validate formatting using the dev release
- - dart: dev
- dart_task: dartfmt
-
-# Only building master means that we don't run two builds for each pull request.
-branches:
- only: [master]
-
-cache:
- directories:
- - $HOME/.pub-cache
diff --git a/AUTHORS b/AUTHORS
index e8063a8..eecd571 100644
--- a/AUTHORS
+++ b/AUTHORS
@@ -4,3 +4,5 @@
# Name/Organization <email address>
Google Inc.
+Alex Li <alexv.525.li@gmail.com>
+Bogdan Lukin <lukin.bogdan.a@gmail.com>
diff --git a/CHANGELOG.md b/CHANGELOG.md
index cd8b7e3..d9bd7c2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,10 @@
+## 2.1.0
+
+* Migrate to null safety.
+* Add `singleResponseFutureWithTimeout` and `singleCallbackPortWithTimeout`,
+ while deprecating the timeout functionality of
+ `singleResponseFuture` and `singleCallbackPort`.
+
## 2.0.3
* Update SDK requirements.
diff --git a/LICENSE b/LICENSE
index de31e1a..633672a 100644
--- a/LICENSE
+++ b/LICENSE
@@ -1,4 +1,5 @@
-Copyright 2015, the Dart project authors. All rights reserved.
+Copyright 2015, the Dart project authors.
+
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
@@ -9,7 +10,7 @@
copyright notice, this list of conditions and the following
disclaimer in the documentation and/or other materials provided
with the distribution.
- * Neither the name of Google Inc. nor the names of its
+ * Neither the name of Google LLC nor the names of its
contributors may be used to endorse or promote products derived
from this software without specific prior written permission.
diff --git a/README.md b/README.md
index c4d55d9..58a6eec 100644
--- a/README.md
+++ b/README.md
@@ -1,4 +1,11 @@
-[](https://travis-ci.org/dart-lang/isolate)
+[](https://github.com/dart-lang/isolate/actions/workflows/test-package.yml)
+
+# DISCONTINUED
+
+This package has been discontinued, and will no longer be maintained.
+
+------------
+
Helps with isolates and isolate communication in Dart.
Requires the `dart:isolate` library being available.
diff --git a/analysis_options.yaml b/analysis_options.yaml
index 3c4fd78..0b0a7b5 100644
--- a/analysis_options.yaml
+++ b/analysis_options.yaml
@@ -1,43 +1,13 @@
-include: package:pedantic/analysis_options.yaml
-#analyzer:
-# strong-mode:
-# implicit-casts: false
+include: package:lints/recommended.yaml
+analyzer:
+ errors:
+ deprecated_member_use_from_same_package: ignore # https://github.com/dart-lang/linter/issues/2703
+ void_checks: ignore # https://github.com/dart-lang/linter/issues/2685
linter:
rules:
- - avoid_empty_else
- - avoid_init_to_null
- - avoid_null_checks_in_equality_operators
- avoid_unused_constructor_parameters
- - await_only_futures
- - camel_case_types
- cancel_subscriptions
- - constant_identifier_names
- - control_flow_in_finally
- directives_ordering
- - empty_catches
- - empty_constructor_bodies
- - empty_statements
- - hash_and_equals
- - implementation_imports
- - iterable_contains_unrelated_type
- - library_names
- - library_prefixes
- - list_remove_unrelated_type
- - non_constant_identifier_names
- - overridden_fields
- package_api_docs
- - package_names
- - package_prefixed_library_names
- - prefer_equal_for_default_values
- - prefer_final_fields
- - prefer_generic_function_type_aliases
- - prefer_is_not_empty
- - slash_for_doc_comments
- test_types_in_equals
- throw_in_finally
- - type_init_formals
- - unnecessary_brace_in_string_interps
- - unnecessary_const
- - unnecessary_new
- - unrelated_type_equality_checks
- - valid_regexps
diff --git a/example/http_server.dart b/example/http_server.dart
index 115c49d..5e4857f 100644
--- a/example/http_server.dart
+++ b/example/http_server.dart
@@ -12,18 +12,19 @@
import 'package:isolate/ports.dart';
import 'package:isolate/runner.dart';
-Future<Future<Object> Function()> runHttpServer(
+Future<Future<Object?> Function()> runHttpServer(
Runner runner, int port, HttpListener listener) async {
var stopPort = await runner.run(_startHttpServer, [port, listener]);
return () => _sendStop(stopPort);
}
-Future _sendStop(SendPort stopPort) => singleResponseFuture(stopPort.send);
+Future<Object?> _sendStop(SendPort stopPort) =>
+ singleResponseFuture(stopPort.send);
-Future<SendPort> _startHttpServer(List args) async {
- int port = args[0];
- HttpListener listener = args[1];
+Future<SendPort> _startHttpServer(List<Object?> args) async {
+ var port = args[0] as int;
+ var listener = args[1] as HttpListener;
var server =
await HttpServer.bind(InternetAddress.anyIPv6, port, shared: true);
@@ -41,6 +42,7 @@
/// The object should be sendable to an equivalent isolate.
abstract class HttpListener {
Future start(HttpServer server);
+
Future stop();
}
@@ -53,7 +55,7 @@
static final _id = Isolate.current.hashCode;
final SendPort _counter;
- StreamSubscription _subscription;
+ StreamSubscription? _subscription;
EchoHttpListener(this._counter);
@@ -77,8 +79,7 @@
@override
Future stop() async {
print('Stopping isolate $_id');
- await _subscription.cancel();
- _subscription = null;
+ await _subscription?.cancel();
}
}
@@ -97,12 +98,13 @@
await ServerSocket.bind(InternetAddress.anyIPv6, port, shared: true);
port = socket.port;
- var isolates = await Future.wait(
+ var isolates = await Future.wait<IsolateRunner>(
Iterable.generate(5, (_) => IsolateRunner.spawn()), cleanUp: (isolate) {
isolate.close();
});
- var stoppers = await Future.wait(isolates.map((IsolateRunner isolate) {
+ var stoppers =
+ await Future.wait<Function>(isolates.map((IsolateRunner isolate) {
return runHttpServer(isolate, socket.port, listener);
}), cleanUp: (shutdownServer) {
shutdownServer();
diff --git a/example/runner_pool.dart b/example/runner_pool.dart
index 78a7756..1c36b84 100644
--- a/example/runner_pool.dart
+++ b/example/runner_pool.dart
@@ -4,8 +4,8 @@
library isolate.example.runner_pool;
-import 'package:isolate/load_balancer.dart';
import 'package:isolate/isolate_runner.dart';
+import 'package:isolate/load_balancer.dart';
void main() {
var N = 44;
@@ -30,10 +30,10 @@
Future<List<int>> parfib(int limit, int parallelity) {
return LoadBalancer.create(parallelity, IsolateRunner.spawn)
.then((LoadBalancer pool) {
- var fibs = List<Future<int>>(limit + 1);
+ var fibs = List<Future<int>?>.filled(limit + 1, null);
// Schedule all calls with exact load value and the heaviest task
// assigned first.
- void schedule(a, b, i) {
+ void schedule(int a, int b, int i) {
if (i < limit) {
schedule(a + b, a, i + 1);
}
@@ -42,15 +42,10 @@
schedule(0, 1, 0);
// And wait for them all to complete.
- return Future.wait(fibs).whenComplete(pool.close);
+ return Future.wait(fibs.cast<Future<int>>()).whenComplete(pool.close);
});
}
-int computeFib(int n) {
- var result = fib(n);
- return result;
-}
-
int fib(int n) {
if (n < 2) return n;
return fib(n - 1) + fib(n - 2);
diff --git a/lib/isolate_runner.dart b/lib/isolate_runner.dart
index 81aa1bb..15e0f65 100644
--- a/lib/isolate_runner.dart
+++ b/lib/isolate_runner.dart
@@ -30,7 +30,7 @@
final SendPort _commandPort;
/// Future returned by [onExit]. Set when [onExit] is first read.
- Future<void> _onExitFuture;
+ Future<void>? _onExitFuture;
/// Create an [IsolateRunner] wrapper for [isolate]
///
@@ -61,7 +61,7 @@
isolate.setErrorsFatal(false);
var pingChannel = SingleResponseChannel();
isolate.ping(pingChannel.port);
- var commandPort = await channel.result;
+ var commandPort = await channel.result as SendPort;
var result = IsolateRunner(isolate, commandPort);
// Guarantees that setErrorsFatal has completed.
await pingChannel.result;
@@ -101,7 +101,7 @@
/// .timeout(new Duration(...), onTimeout: () => print("No response"));
/// ```
Future<void> kill({Duration timeout = const Duration(seconds: 1)}) {
- var onExit = singleResponseFuture(isolate.addOnExitListener);
+ final onExit = singleResponseFuture(isolate.addOnExitListener);
if (Duration.zero == timeout) {
isolate.kill(priority: Isolate.immediate);
return onExit;
@@ -127,14 +127,10 @@
/// Guaranteed to only complete after all previous sent isolate commands
/// (like pause and resume) have been handled.
/// Paused isolates do respond to ping requests.
- Future<bool> ping({Duration timeout = const Duration(seconds: 1)}) {
- var channel = SingleResponseChannel(
- callback: _kTrue, timeout: timeout, timeoutValue: false);
- isolate.ping(channel.port);
- return channel.result;
- }
-
- static bool _kTrue(_) => true;
+ Future<bool> ping({Duration timeout = const Duration(seconds: 1)}) =>
+ singleResponseFutureWithTimeout((port) {
+ isolate.ping(port, response: true);
+ }, timeout, false);
/// Pauses the isolate.
///
@@ -146,12 +142,14 @@
///
/// If [resumeCapability] is omitted, it defaults to the [isolate]'s
/// [Isolate.pauseCapability].
+ /// If the isolate has no pause capability, nothing happens.
///
/// Calling pause more than once with the same `resumeCapability`
/// has no further effect. Only a single call to [resume] is needed
/// to resume the isolate.
- void pause([Capability resumeCapability]) {
+ void pause([Capability? resumeCapability]) {
resumeCapability ??= isolate.pauseCapability;
+ if (resumeCapability == null) return;
isolate.pause(resumeCapability);
}
@@ -159,11 +157,13 @@
///
/// If [resumeCapability] is omitted, it defaults to the isolate's
/// [Isolate.pauseCapability].
+ /// If the isolate has no pause capability, nothing happens.
///
/// Even if `pause` has been called more than once with the same
/// `resumeCapability`, a single resume call with stop the pause.
- void resume([Capability resumeCapability]) {
+ void resume([Capability? resumeCapability]) {
resumeCapability ??= isolate.pauseCapability;
+ if (resumeCapability == null) return;
isolate.resume(resumeCapability);
}
@@ -174,6 +174,12 @@
/// If the call returns a [Future], the final result of that future
/// will be returned.
///
+ /// If [timeout] is provided, and the returned future does not complete
+ /// before that duration has passed,
+ /// the [onTimeout] action is executed instead, and its result (whether it
+ /// returns or throws) is used as the result of the returned future.
+ /// If [onTimeout] is omitted, it defaults to throwing a[TimeoutException].
+ ///
/// This works similar to the arguments to [Isolate.spawn], except that
/// it runs in the existing isolate and the return value is returned to
/// the caller.
@@ -188,8 +194,8 @@
/// }
/// ```
@override
- Future<R> run<R, P>(FutureOr<R> Function(P argument) function, P argument,
- {Duration timeout, Function() onTimeout}) {
+ Future<R> run<R, P>(FutureOr<R>? Function(P argument) function, P argument,
+ {Duration? timeout, FutureOr<R> Function()? onTimeout}) {
return singleResultFuture<R>((SendPort port) {
_commandPort.send(list4(_run, function, argument, port));
}, timeout: timeout, onTimeout: onTimeout);
@@ -201,36 +207,35 @@
/// as errors in the stream. Be ready to handle the errors.
///
/// The stream closes when the isolate shuts down.
- Stream get errors {
- StreamController controller;
- RawReceivePort port;
- void handleError(message) {
- if (message == null) {
- // Isolate shutdown.
- port.close();
- controller.close();
- } else {
- // Uncaught error.
- String errorDescription = message[0];
- String stackDescription = message[1];
- var error = RemoteError(errorDescription, stackDescription);
- controller.addError(error, error.stackTrace);
- }
- }
-
- controller = StreamController.broadcast(
- sync: true,
- onListen: () {
- port = RawReceivePort(handleError);
- isolate.addErrorListener(port.sendPort);
- isolate.addOnExitListener(port.sendPort);
- },
- onCancel: () {
- isolate.removeErrorListener(port.sendPort);
- isolate.removeOnExitListener(port.sendPort);
+ ///
+ /// If the isolate shuts down while noone is listening on this stream,
+ /// the stream will not be closed, and listening to the stream again
+ /// after the isolate has shut down will not yield any events.
+ Stream<Never> get errors {
+ var controller = StreamController<Never>.broadcast(sync: true);
+ controller.onListen = () {
+ var port = RawReceivePort();
+ port.handler = (message) {
+ if (message == null) {
+ // Isolate shutdown.
port.close();
- port = null;
- });
+ controller.close();
+ } else {
+ // Uncaught error.
+ final errorDescription = message[0] as String;
+ final stackDescription = message[1] as String;
+ var error = RemoteError(errorDescription, stackDescription);
+ controller.addError(error, error.stackTrace);
+ }
+ };
+ isolate.addErrorListener(port.sendPort);
+ isolate.addOnExitListener(port.sendPort);
+ controller.onCancel = () {
+ port.close();
+ isolate.removeErrorListener(port.sendPort);
+ isolate.removeOnExitListener(port.sendPort);
+ };
+ };
return controller.stream;
}
@@ -241,15 +246,14 @@
/// If the isolate has already stopped responding to commands,
/// the returned future will be completed after one second,
/// using [ping] to check if the isolate is still alive.
- Future<void> get onExit {
- // TODO(lrn): Is there a way to see if an isolate is dead
- // so we can close the receive port for this future?
- // Using [ping] for now.
+ Future<void>? get onExit {
+ // Using [ping] to see if the isolate is dead.
+ // Can't distinguish that from a slow-to-answer isolate.
if (_onExitFuture == null) {
var channel = SingleResponseChannel<void>();
isolate.addOnExitListener(channel.port);
_onExitFuture = channel.result.then(ignore);
- ping().then((bool alive) {
+ ping().then<void>((bool alive) {
if (!alive) {
channel.interrupt();
_onExitFuture = null;
@@ -269,6 +273,7 @@
/// instead of relying on [IsolateRunner.spawn].
class IsolateRunnerRemote {
final RawReceivePort _commandPort = RawReceivePort();
+
IsolateRunnerRemote() {
_commandPort.handler = _handleCommand;
}
@@ -279,17 +284,17 @@
/// manually, otherwise it's handled by [IsolateRunner.spawn].
SendPort get commandPort => _commandPort.sendPort;
- static void _create(Object data) {
+ static void _create(Object? data) {
var initPort = data as SendPort;
var remote = IsolateRunnerRemote();
initPort.send(remote.commandPort);
}
- void _handleCommand(List<Object> command) {
+ void _handleCommand(List<Object?> command) {
switch (command[0]) {
case _shutdown:
_commandPort.close();
- (command[1] as SendPort)?.send(null);
+ (command[1] as SendPort?)?.send(null);
break;
case _run:
var function = command[1] as Function;
diff --git a/lib/load_balancer.dart b/lib/load_balancer.dart
index da5b7e0..97d7cad 100644
--- a/lib/load_balancer.dart
+++ b/lib/load_balancer.dart
@@ -5,7 +5,7 @@
/// A load-balancing runner pool.
library isolate.load_balancer;
-import 'dart:async' show FutureOr;
+import 'dart:async' show Completer, FutureOr;
import 'runner.dart';
import 'src/errors.dart';
@@ -15,27 +15,58 @@
///
/// Keeps a pool of runners,
/// and allows running function through the runner with the lowest current load.
+///
+/// The number of pool runner entries is fixed when the pool is created.
+/// When the pool is [close]d, all runners are closed as well.
+///
+/// The load balancer is not reentrant.
+/// Executing a [run] function should not *synchronously*
+/// call methods on the load balancer.
class LoadBalancer implements Runner {
- // A heap-based priority queue of entries, prioritized by `load`.
- // Each entry has its own entry in the queue, for faster update.
+ /// A stand-in future which can be used as a default value.
+ ///
+ /// The future never completes, so it should not be exposed to users.
+ static final _defaultFuture = Completer<Never>().future;
+
+ /// Reusable empty fixed-length list.
+ static final _emptyQueue = List<_LoadBalancerEntry>.empty(growable: false);
+
+ /// A heap-based priority queue of entries, prioritized by `load`.
+ ///
+ /// The entries of the list never change, only their positions.
+ /// Those with positions below `_length`
+ /// are considered to currently be in the queue.
+ /// All operations except [close] should end up with all entries
+ /// still in the pool. Some entries may be removed temporarily in order
+ /// to change their load and then add them back.
+ ///
+ /// Each [_LoadBalancerEntry] has its current position in the queue
+ /// as [_LoadBalancerEntry.queueIndex].
+ ///
+ /// Is set to an empty list on clear.
List<_LoadBalancerEntry> _queue;
- // The number of entries currently in the queue.
+ /// Current number of elements in [_queue].
+ ///
+ /// Always a number between zero and [_queue.length].
+ /// Elements with indices below this value are
+ /// in the queue, and maintain the heap invariant.
+ /// Elements with indices above this value are temporarily
+ /// removed from the queue and are ordered by when they
+ /// were removed.
int _length;
- // Whether [stop] has been called.
- Future<void> _stopFuture;
+ /// The future returned by [stop].
+ ///
+ /// Is `null` until [stop] is first called.
+ Future<void>? _stopFuture;
/// Create a load balancer backed by the [Runner]s of [runners].
LoadBalancer(Iterable<Runner> runners) : this._(_createEntries(runners));
LoadBalancer._(List<_LoadBalancerEntry> entries)
: _queue = entries,
- _length = entries.length {
- for (var i = 0; i < _length; i++) {
- _queue[i].queueIndex = i;
- }
- }
+ _length = entries.length;
/// The number of runners currently in the pool.
int get length => _length;
@@ -56,8 +87,10 @@
}
static List<_LoadBalancerEntry> _createEntries(Iterable<Runner> runners) {
- var entries = runners.map((runner) => _LoadBalancerEntry(runner));
- return List<_LoadBalancerEntry>.from(entries, growable: false);
+ var index = 0;
+ return runners
+ .map((runner) => _LoadBalancerEntry(runner, index++))
+ .toList(growable: false);
}
/// Execute the command in the currently least loaded isolate.
@@ -73,11 +106,20 @@
/// as normal. If the runners are running in other isolates, then
/// the [onTimeout] function must be a constant function.
@override
- Future<R> run<R, P>(FutureOr<R> Function(P argument) function, argument,
- {Duration timeout, FutureOr<R> Function() onTimeout, int load = 100}) {
+ Future<R> run<R, P>(FutureOr<R> Function(P argument) function, P argument,
+ {Duration? timeout, FutureOr<R> Function()? onTimeout, int load = 100}) {
RangeError.checkNotNegative(load, 'load');
- var entry = _first;
- _increaseLoad(entry, load);
+ if (_length == 0) {
+ // Can happen if created with zero runners,
+ // or after being closed.
+ if (_stopFuture != null) {
+ throw StateError("Load balancer has been closed");
+ }
+ throw StateError("No runners in pool");
+ }
+ var entry = _queue.first;
+ entry.load += load;
+ _bubbleDown(entry, 0);
return entry.run(this, load, function, argument, timeout, onTimeout);
}
@@ -97,18 +139,19 @@
/// as normal.
List<Future<R>> runMultiple<R, P>(
int count, FutureOr<R> Function(P argument) function, P argument,
- {Duration timeout, FutureOr<R> Function() onTimeout, int load = 100}) {
+ {Duration? timeout, FutureOr<R> Function()? onTimeout, int load = 100}) {
RangeError.checkValueInInterval(count, 1, _length, 'count');
RangeError.checkNotNegative(load, 'load');
if (count == 1) {
- return List<Future<R>>(1)
- ..[0] = run(function, argument,
- load: load, timeout: timeout, onTimeout: onTimeout);
+ return List<Future<R>>.filled(
+ 1,
+ run(function, argument,
+ load: load, timeout: timeout, onTimeout: onTimeout));
}
- var result = List<Future<R>>(count);
+ var result = List<Future<R>>.filled(count, _defaultFuture);
if (count == _length) {
// No need to change the order of entries in the queue.
- for (var i = 0; i < count; i++) {
+ for (var i = 0; i < _length; i++) {
var entry = _queue[i];
entry.load += load;
result[i] =
@@ -117,18 +160,15 @@
} else {
// Remove the [count] least loaded services and run the
// command on each, then add them back to the queue.
- // This avoids running the same command twice in the same
- // isolate.
- // We can't assume that the first [count] entries in the
- // heap list are the least loaded.
- var entries = List<_LoadBalancerEntry>(count);
for (var i = 0; i < count; i++) {
- entries[i] = _removeFirst();
+ _removeFirst();
}
+ // The removed entries are stored in `_queue` in positions from
+ // `_length` to `_length + count - 1`.
for (var i = 0; i < count; i++) {
- var entry = entries[i];
+ var entry = _queue[_length];
entry.load += load;
- _add(entry);
+ _addNext();
result[i] =
entry.run(this, load, function, argument, timeout, onTimeout);
}
@@ -138,17 +178,15 @@
@override
Future<void> close() {
- if (_stopFuture != null) return _stopFuture;
- _stopFuture =
- MultiError.waitUnordered(_queue.take(_length).map((e) => e.close()))
- .then(ignore);
- // Remove all entries.
- for (var i = 0; i < _length; i++) {
- _queue[i].queueIndex = -1;
- }
- _queue = null;
+ var stopFuture = _stopFuture;
+ if (stopFuture != null) return stopFuture;
+ var queue = _queue;
+ var length = _length;
+ _queue = _emptyQueue;
_length = 0;
- return _stopFuture;
+ return _stopFuture = MultiError.waitUnordered(
+ [for (var i = 0; i < length; i++) queue[i].close()],
+ ).then(ignore);
}
/// Place [element] in heap at [index] or above.
@@ -156,17 +194,19 @@
/// Put element into the empty cell at `index`.
/// While the `element` has higher priority than the
/// parent, swap it with the parent.
+ ///
+ /// Ignores [element]'s initial [_LoadBalancerEntry.queueIndex],
+ /// but sets it to the final position when the element has
+ /// been placed.
void _bubbleUp(_LoadBalancerEntry element, int index) {
while (index > 0) {
var parentIndex = (index - 1) ~/ 2;
var parent = _queue[parentIndex];
if (element.compareTo(parent) > 0) break;
- _queue[index] = parent;
- parent.queueIndex = index;
+ _queue[index] = parent..queueIndex = index;
index = parentIndex;
}
- _queue[index] = element;
- element.queueIndex = index;
+ _queue[index] = element..queueIndex = index;
}
/// Place [element] in heap at [index] or above.
@@ -174,6 +214,10 @@
/// Put element into the empty cell at `index`.
/// While the `element` has lower priority than either child,
/// swap it with the highest priority child.
+ ///
+ /// Ignores [element]'s initial [_LoadBalancerEntry.queueIndex],
+ /// but sets it to the final position when the element has
+ /// been placed.
void _bubbleDown(_LoadBalancerEntry element, int index) {
while (true) {
var childIndex = index * 2 + 1; // Left child index.
@@ -188,102 +232,79 @@
}
}
if (element.compareTo(child) <= 0) break;
- _queue[index] = child;
- child.queueIndex = index;
+ _queue[index] = child..queueIndex = index;
index = childIndex;
}
- _queue[index] = element;
- element.queueIndex = index;
+ _queue[index] = element..queueIndex = index;
}
- /// Removes the entry from the queue, but doesn't stop its service.
+ /// Removes the first entry from the queue, but doesn't stop its service.
///
/// The entry is expected to be either added back to the queue
/// immediately or have its stop method called.
- void _remove(_LoadBalancerEntry entry) {
- var index = entry.queueIndex;
- if (index < 0) return;
- entry.queueIndex = -1;
+ ///
+ /// After the remove, the entry is stored as `_queue[_length]`.
+ _LoadBalancerEntry _removeFirst() {
+ assert(_length > 0);
+ _LoadBalancerEntry entry = _queue.first;
_length--;
- var replacement = _queue[_length];
- _queue[_length] = null;
- if (index < _length) {
- if (entry.compareTo(replacement) < 0) {
- _bubbleDown(replacement, index);
- } else {
- _bubbleUp(replacement, index);
- }
+ if (_length > 0) {
+ var replacement = _queue[_length];
+ _queue[_length] = entry..queueIndex = _length;
+ _bubbleDown(replacement, 0);
}
+ return entry;
}
- /// Adds entry to the queue.
- void _add(_LoadBalancerEntry entry) {
- if (_stopFuture != null) throw StateError('LoadBalancer is stopped');
- assert(entry.queueIndex < 0);
- if (_queue.length == _length) {
- _grow();
- }
+ /// Adds next unused entry to the queue.
+ ///
+ /// Adds the entry at [_length] to the queue.
+ void _addNext() {
+ assert(_length < _queue.length);
var index = _length;
+ var entry = _queue[index];
_length = index + 1;
_bubbleUp(entry, index);
}
- void _increaseLoad(_LoadBalancerEntry entry, int load) {
- assert(load >= 0);
- entry.load += load;
- if (entry.inQueue) {
- _bubbleDown(entry, entry.queueIndex);
- }
- }
-
+ /// Decreases the load of an element which is currently in the queue.
+ ///
+ /// Elements outside the queue can just have their `load` modified directly.
void _decreaseLoad(_LoadBalancerEntry entry, int load) {
assert(load >= 0);
entry.load -= load;
- if (entry.inQueue) {
- _bubbleUp(entry, entry.queueIndex);
+ var index = entry.queueIndex;
+ // Should always be the case unless the load balancer
+ // has been closed, or events are happening out of their
+ // proper order.
+ if (index < _length) {
+ _bubbleUp(entry, index);
}
}
-
- void _grow() {
- var newQueue = List(_length * 2);
- newQueue.setRange(0, _length, _queue);
- _queue = newQueue;
- }
-
- _LoadBalancerEntry get _first {
- assert(_length > 0);
- return _queue[0];
- }
-
- _LoadBalancerEntry _removeFirst() {
- var result = _first;
- _remove(result);
- return result;
- }
}
class _LoadBalancerEntry implements Comparable<_LoadBalancerEntry> {
+ /// The position in the heap queue.
+ ///
+ /// Maintained when entries move around the queue.
+ /// Only needed for [LoadBalancer._decreaseLoad].
+ int queueIndex;
+
// The current load on the isolate.
int load = 0;
- // The current index in the heap-queue.
- // Negative when the entry is not part of the queue.
- int queueIndex = -1;
- // The service used to send commands to the other isolate.
+ // The service used to execute commands.
Runner runner;
- _LoadBalancerEntry(Runner runner) : runner = runner;
-
- /// Whether the entry is still in the queue.
- bool get inQueue => queueIndex >= 0;
+ _LoadBalancerEntry(this.runner, this.queueIndex);
Future<R> run<R, P>(
LoadBalancer balancer,
int load,
FutureOr<R> Function(P argument) function,
- argument,
- Duration timeout,
- FutureOr<R> Function() onTimeout) {
+ P argument,
+ Duration? timeout,
+ FutureOr<R> Function()? onTimeout) {
return runner
.run<R, P>(function, argument, timeout: timeout, onTimeout: onTimeout)
.whenComplete(() {
diff --git a/lib/ports.dart b/lib/ports.dart
index 478e984..0bc1eda 100644
--- a/lib/ports.dart
+++ b/lib/ports.dart
@@ -28,14 +28,21 @@
///
/// The [callback] function is called once, with the first message
/// received by the receive port.
-/// All further messages are ignored.
+/// All further messages are ignored and the port is closed.
///
/// If [timeout] is supplied, it is used as a limit on how
-/// long it can take before the message is received. If a
-/// message isn't received in time, the `callback` function
-/// is called once with the [timeoutValue] instead.
+/// long it can take before the message is received.
+/// If a message has not been received within the [timeout] duration,
+/// the callback is called with the [timeoutValue] instead, and
+/// the port is closed.
+/// If the message type, [P], does not allow `null` and [timeout] is
+/// non-`null`, then [timeoutValue] must be provided and non-`null`.
///
-/// If the received value is not a [T], it will cause an uncaught
+/// Use [singleCallbackPortWithTimeout] instead of the deprecated
+/// members. That will prevent run-time error arising from calling
+/// this method with a non-nullable [P] type and a null [timeoutValue].
+///
+/// If the received value is not a [P], it will cause an uncaught
/// asynchronous error in the current zone.
///
/// Returns the `SendPort` expecting the single message.
@@ -46,23 +53,73 @@
/// ..first.timeout(duration, () => timeoutValue).then(callback))
/// .sendPort
/// ```
+/// when [timeout] is provided.
SendPort singleCallbackPort<P>(void Function(P response) callback,
- {Duration timeout, P timeoutValue}) {
+ {@Deprecated("Use singleCallbackPortWithTimeout instead") Duration? timeout,
+ @Deprecated("Use singleCallbackPortWithTimeout instead") P? timeoutValue}) {
+ if (timeout == null) {
+ return _singleCallbackPort<P>(callback);
+ }
+ if (timeoutValue is! P) {
+ throw ArgumentError.value(
+ null, "timeoutValue", "The result type is non-null");
+ }
+ return singleCallbackPortWithTimeout<P>(callback, timeout, timeoutValue);
+}
+
+/// Helper function for [singleCallbackPort].
+///
+/// Replace [singleCallbackPort] with this
+/// when removing the deprecated parameters.
+SendPort _singleCallbackPort<P>(void Function(P) callback) {
var responsePort = RawReceivePort();
var zone = Zone.current;
callback = zone.registerUnaryCallback(callback);
- Timer timer;
+ responsePort.handler = (response) {
+ responsePort.close();
+ zone.runUnary(callback, response as P);
+ };
+ return responsePort.sendPort;
+}
+
+/// Create a [SendPort] that accepts only one message.
+///
+/// The [callback] function is called once, with the first message
+/// received by the receive port.
+/// All further messages are ignored and the port is closed.
+///
+/// If a message has not been received within the [timeout] duration,
+/// the callback is called with the [timeoutValue] instead, and
+/// the port is closed.
+///
+/// If the received value is not a [P], it will cause an uncaught
+/// asynchronous error in the current zone.
+///
+/// Returns the `SendPort` expecting the single message.
+///
+/// Equivalent to:
+/// ```dart
+/// (new ReceivePort()
+/// ..first.timeout(duration, () => timeoutValue).then(callback))
+/// .sendPort
+/// ```
+SendPort singleCallbackPortWithTimeout<P>(
+ void Function(P response) callback, Duration timeout, P timeoutValue) {
+ var responsePort = RawReceivePort();
+ var zone = Zone.current;
+ callback = zone.registerUnaryCallback(callback);
+ Timer? timer;
responsePort.handler = (response) {
responsePort.close();
timer?.cancel();
zone.runUnary(callback, response as P);
};
- if (timeout != null) {
- timer = Timer(timeout, () {
- responsePort.close();
- callback(timeoutValue);
- });
- }
+
+ timer = Timer(timeout, () {
+ responsePort.close();
+ callback(timeoutValue);
+ });
+
return responsePort.sendPort;
}
@@ -92,17 +149,19 @@
/// completed in response to another event, either a port message or a timer.
///
/// Returns the `SendPort` expecting the single message.
-SendPort singleCompletePort<R, P>(Completer<R> completer,
- {FutureOr<R> Function(P message) callback,
- Duration timeout,
- FutureOr<R> Function() onTimeout}) {
+SendPort singleCompletePort<R, P>(
+ Completer<R> completer, {
+ FutureOr<R> Function(P message)? callback,
+ Duration? timeout,
+ FutureOr<R> Function()? onTimeout,
+}) {
if (callback == null && timeout == null) {
- return singleCallbackPort<Object>((response) {
+ return _singleCallbackPort<Object>((response) {
_castComplete<R>(completer, response);
});
}
var responsePort = RawReceivePort();
- Timer timer;
+ Timer? timer;
if (callback == null) {
responsePort.handler = (response) {
responsePort.close();
@@ -129,7 +188,13 @@
timer = Timer(timeout, () {
responsePort.close();
if (onTimeout != null) {
- completer.complete(Future.sync(onTimeout));
+ /// workaround for incomplete generic parameters promotion.
+ /// example is available in 'TimeoutFirst with invalid null' test
+ try {
+ completer.complete(Future.sync(onTimeout));
+ } catch (e, st) {
+ completer.completeError(e, st);
+ }
} else {
completer
.completeError(TimeoutException('Future not completed', timeout));
@@ -155,34 +220,85 @@
/// long it can take before the message is received. If a
/// message isn't received in time, the [timeoutValue] used
/// as the returned future's value instead.
+/// If the result type, [R], does not allow `null`, and [timeout] is provided,
+/// then [timeoutValue] must also be non-`null`.
+/// Use [singleResponseFutureWithTimeout] instead of providing
+/// the optional parameters to this function. It prevents getting run-time
+/// errors from providing a [timeout] and no [timeoutValue] with a non-nullable
+/// result type.
///
-/// If you want a timeout on the returned future, it's recommended to
-/// use the [timeout] parameter, and not [Future.timeout] on the result.
-/// The `Future` method won't be able to close the underlying [ReceivePort].
-Future<R> singleResponseFuture<R>(void Function(SendPort responsePort) action,
- {Duration timeout, R timeoutValue}) {
+/// If you need a timeout on the operation, it's recommended to specify
+/// a timeout using [singleResponseFutureWithTimeout],
+/// and not use [Future.timeout] on the returned `Future`.
+/// The `Future` method won't be able to close the underlying [ReceivePort],
+/// and will keep waiting for the first response anyway.
+Future<R> singleResponseFuture<R>(
+ void Function(SendPort responsePort) action, {
+ @Deprecated("Use singleResponseFutureWithTimeout instead") Duration? timeout,
+ @Deprecated("Use singleResponseFutureWithTimeout instead") R? timeoutValue,
+}) {
+ if (timeout == null) {
+ return _singleResponseFuture<R>(action);
+ }
+ if (timeoutValue is! R) {
+ throw ArgumentError.value(
+ null, "timeoutValue", "The result type is non-null");
+ }
+ return singleResponseFutureWithTimeout(action, timeout, timeoutValue);
+}
+
+/// Helper function for [singleResponseFuture].
+///
+/// Use this as the implementation of [singleResponseFuture]
+/// when removing the deprecated parameters.
+Future<R> _singleResponseFuture<R>(
+ void Function(SendPort responsePort) action) {
var completer = Completer<R>.sync();
var responsePort = RawReceivePort();
- Timer timer;
var zone = Zone.current;
- responsePort.handler = (Object response) {
+ responsePort.handler = (response) {
responsePort.close();
- timer?.cancel();
zone.run(() {
_castComplete<R>(completer, response);
});
};
- if (timeout != null) {
- timer = Timer(timeout, () {
- responsePort.close();
- completer.complete(timeoutValue);
- });
- }
try {
action(responsePort.sendPort);
} catch (error, stack) {
responsePort.close();
- timer?.cancel();
+ // Delay completion because completer is sync.
+ scheduleMicrotask(() {
+ completer.completeError(error, stack);
+ });
+ }
+ return completer.future;
+}
+
+/// Same as [singleResponseFuture], but with required [timeoutValue],
+/// this allows us not to require a nullable return value
+Future<R> singleResponseFutureWithTimeout<R>(
+ void Function(SendPort responsePort) action,
+ Duration timeout,
+ R timeoutValue) {
+ var completer = Completer<R>.sync();
+ var responsePort = RawReceivePort();
+ var timer = Timer(timeout, () {
+ responsePort.close();
+ completer.complete(timeoutValue);
+ });
+ var zone = Zone.current;
+ responsePort.handler = (response) {
+ responsePort.close();
+ timer.cancel();
+ zone.run(() {
+ _castComplete<R>(completer, response);
+ });
+ };
+ try {
+ action(responsePort.sendPort);
+ } catch (error, stack) {
+ responsePort.close();
+ timer.cancel();
// Delay completion because completer is sync.
scheduleMicrotask(() {
completer.completeError(error, stack);
@@ -196,8 +312,8 @@
/// The result of [future] is sent on [resultPort] in a form expected by
/// either [receiveFutureResult], [completeFutureResult], or
/// by the port of [singleResultFuture].
-void sendFutureResult(Future<Object> future, SendPort resultPort) {
- future.then((value) {
+void sendFutureResult(Future<Object?> future, SendPort resultPort) {
+ future.then<void>((value) {
resultPort.send(list1(value));
}, onError: (error, stack) {
resultPort.send(list2('$error', '$stack'));
@@ -223,9 +339,9 @@
/// If `onTimeout` is omitted, it defaults to throwing
/// a [TimeoutException].
Future<R> singleResultFuture<R>(void Function(SendPort responsePort) action,
- {Duration timeout, FutureOr<R> Function() onTimeout}) {
+ {Duration? timeout, FutureOr<R> Function()? onTimeout}) {
var completer = Completer<R>.sync();
- var port = singleCompletePort<R, List<Object>>(completer,
+ var port = singleCompletePort<R, List<Object?>>(completer,
callback: receiveFutureResult, timeout: timeout, onTimeout: onTimeout);
try {
action(port);
@@ -239,12 +355,12 @@
/// Completes a completer with a message created by [sendFutureResult]
///
/// The [response] must be a message on the format sent by [sendFutureResult].
-void completeFutureResult<R>(List<Object> response, Completer<R> completer) {
+void completeFutureResult<R>(List<Object?> response, Completer<R> completer) {
if (response.length == 2) {
- var error = RemoteError(response[0], response[1]);
+ var error = RemoteError(response[0] as String, response[1] as String);
completer.completeError(error, error.stackTrace);
} else {
- R result = response[0];
+ var result = response[0] as R;
completer.complete(result);
}
}
@@ -253,12 +369,12 @@
/// result.
///
/// The [response] must be a message on the format sent by [sendFutureResult].
-Future<R> receiveFutureResult<R>(List<Object> response) {
+Future<R> receiveFutureResult<R>(List<Object?> response) {
if (response.length == 2) {
- var error = RemoteError(response[0], response[1]);
+ var error = RemoteError(response[0] as String, response[1] as String);
return Future.error(error, error.stackTrace);
}
- R result = response[0];
+ var result = response[0] as R;
return Future<R>.value(result);
}
@@ -270,8 +386,8 @@
final Zone _zone;
final RawReceivePort _receivePort;
final Completer<R> _completer;
- final Function _callback;
- Timer _timer;
+ final FutureOr<R> Function(dynamic)? _callback;
+ Timer? _timer;
/// Creates a response channel.
///
@@ -281,25 +397,38 @@
/// to `callback`, and the result of that is used to complete `result`.
///
/// If [timeout] is provided, the future is completed after that
- /// duration if it hasn't received a value from the port earlier.
+ /// duration if it hasn't received a value from the port earlier,
+ /// with a value determined as follows:
/// If [throwOnTimeout] is true, the the future is completed with a
- /// [TimeoutException] as an error if it times out.
+ /// [TimeoutException] as an error.
/// Otherwise, if [onTimeout] is provided,
/// the future is completed with the result of running `onTimeout()`.
- /// If `onTimeout` is not provided either,
- /// the future is completed with `timeoutValue`, which defaults to `null`.
+ /// If `onTimeout` is also not provided,
+ /// then the future is completed with the provided [timeoutValue],
+ /// which defaults to `null`.
+ /// If the result type, [R], is not nullable, and [timeoutValue]
+ /// is to be used as the result of the future,
+ /// then it must have a non-`null` value.
SingleResponseChannel(
- {FutureOr<R> Function(Null value) callback,
- Duration timeout,
+ {FutureOr<R> Function(dynamic value)? callback,
+ Duration? timeout,
bool throwOnTimeout = false,
- FutureOr<R> Function() onTimeout,
- R timeoutValue})
+ FutureOr<R> Function()? onTimeout,
+ R? timeoutValue})
: _receivePort = RawReceivePort(),
_completer = Completer<R>.sync(),
_callback = callback,
_zone = Zone.current {
_receivePort.handler = _handleResponse;
if (timeout != null) {
+ if (!throwOnTimeout &&
+ onTimeout == null &&
+ timeoutValue == null &&
+ timeoutValue is! R) {
+ _receivePort.close();
+ throw ArgumentError.value(null, "timeoutValue",
+ "The value is needed and the result must not be null");
+ }
_timer = Timer(timeout, () {
// Executed as a timer event.
_receivePort.close();
@@ -310,7 +439,7 @@
} else if (onTimeout != null) {
_completer.complete(Future.sync(onTimeout));
} else {
- _completer.complete(timeoutValue);
+ _completer.complete(timeoutValue as R);
}
}
});
@@ -328,7 +457,12 @@
/// If the channel hasn't received a value yet, or timed out, it is stopped
/// (like by a timeout) and the [SingleResponseChannel.result]
/// is completed with [result].
- void interrupt([R result]) {
+ /// If the result type is not nullable, the [result] must not be `null`.
+ void interrupt([R? result]) {
+ if (result is! R) {
+ throw ArgumentError.value(null, "result",
+ "The value is needed and the result must not be null");
+ }
_receivePort.close();
_cancelTimer();
if (!_completer.isCompleted) {
@@ -338,8 +472,9 @@
}
void _cancelTimer() {
- if (_timer != null) {
- _timer.cancel();
+ final timer = _timer;
+ if (timer != null) {
+ timer.cancel();
_timer = null;
}
}
@@ -348,7 +483,8 @@
// Executed as a port event.
_receivePort.close();
_cancelTimer();
- if (_callback == null) {
+ final callback = _callback;
+ if (callback == null) {
try {
_completer.complete(v as R);
} catch (e, s) {
@@ -363,7 +499,7 @@
// created in a different error zone, an error from the root zone
// would become uncaught.
_zone.run(() {
- _completer.complete(Future.sync(() => _callback(v)));
+ _completer.complete(Future<R>.sync(() => callback(v)));
});
}
}
@@ -371,7 +507,7 @@
// Helper function that casts an object to a type and completes a
// corresponding completer, or completes with the error if the cast fails.
-void _castComplete<R>(Completer<R> completer, Object value) {
+void _castComplete<R>(Completer<R> completer, Object? value) {
try {
completer.complete(value as R);
} catch (error, stack) {
diff --git a/lib/registry.dart b/lib/registry.dart
index de41553..d972e09 100644
--- a/lib/registry.dart
+++ b/lib/registry.dart
@@ -32,7 +32,7 @@
///
/// A [Registry] object caches objects found using the [lookup]
/// method, or added using [add], and returns the same object every time
-/// they are requested.
+/// it is requested.
/// A different [Registry] object that works on the same underlying registry,
/// will not preserve the identity of elements
///
@@ -44,7 +44,7 @@
/// See [SendPort] for details on the restrictions on objects which can be sent
/// between isolates.
///
-/// A registry can be sued to make a number of object available to separate
+/// A registry can be used to make a number of objects available to separate
/// workers in different isolates, for example ones created using
/// [IsolateRunner], without sending all the objects to all the isolates.
/// A worker can then request the data it needs, and it can add new data
@@ -98,9 +98,9 @@
// The cache is stored in an [Expando], not on the object.
// This allows sending the `Registry` object through a `SendPort` without
// also copying the cache.
- static final Expando _caches = Expando();
+ static final Expando<_RegistryCache> _caches = Expando<_RegistryCache>();
- /// Port for sending command to the central registry manager.
+ /// Port for sending commands to the central registry manager.
final SendPort _commandPort;
/// Create a registry linked to a [RegistryManager] through [commandPort].
@@ -121,13 +121,7 @@
: _commandPort = commandPort,
_timeout = timeout;
- _RegistryCache get _cache {
- _RegistryCache cache = _caches[this];
- if (cache != null) return cache;
- cache = _RegistryCache();
- _caches[this] = cache;
- return cache;
- }
+ _RegistryCache get _cache => _caches[this] ??= _RegistryCache();
/// Check and get the identity of an element.
///
@@ -153,7 +147,7 @@
/// from other elements. Any object can be used as a tag, as long as
/// it preserves equality when sent through a [SendPort].
/// This makes [Capability] objects a good choice for tags.
- Future<Capability> add(T element, {Iterable tags}) {
+ Future<Capability> add(T element, {Iterable? tags}) {
var cache = _cache;
if (cache.contains(element)) {
return Future<Capability>.sync(() {
@@ -163,10 +157,10 @@
}
var completer = Completer<Capability>();
var port = singleCompletePort(completer,
- callback: (List response) {
+ callback: (List<Object?> response) {
assert(cache.isAdding(element));
- int id = response[0];
- Capability removeCapability = response[1];
+ var id = response[0] as int;
+ var removeCapability = response[1] as Capability;
cache.register(id, element);
return removeCapability;
},
@@ -181,22 +175,24 @@
return completer.future;
}
- /// Remove the element from the registry.
+ /// Removes the [element] from the registry.
///
- /// Returns `true` if removing the element succeeded, or `false` if the
- /// elements wasn't in the registry, or if it couldn't be removed.
+ /// Returns `true` if removing the element succeeded, and `false` if the
+ /// elements either wasn't in the registry, or it couldn't be removed.
///
/// The [removeCapability] must be the same capability returned by [add]
/// when the object was added. If the capability is wrong, the
- /// object is not removed, and this function returns false.
+ /// object is not removed, and this function returns `false`.
Future<bool> remove(T element, Capability removeCapability) {
var id = _cache.id(element);
if (id == null) {
+ // If the element is not in the cache, then it was not a value
+ // that originally came from the registry.
return Future<bool>.value(false);
}
var completer = Completer<bool>();
var port = singleCompletePort(completer, callback: (bool result) {
- _cache.remove(id);
+ if (result) _cache.remove(id);
return result;
}, timeout: _timeout);
_commandPort.send(list4(_removeValue, id, removeCapability, port));
@@ -212,8 +208,8 @@
/// Tags are compared using [Object.==] equality.
///
/// Fails if any of the elements are not in the registry.
- Future addTags(Iterable<T> elements, Iterable<Object> tags) {
- List<Object> ids = elements.map(_getId).toList(growable: false);
+ Future addTags(Iterable<T> elements, Iterable<Object?> tags) {
+ var ids = elements.map(_getId).toList(growable: false);
return _addTags(ids, tags);
}
@@ -224,8 +220,8 @@
/// before or not.
///
/// Fails if any of the elements are not in the registry.
- Future<void> removeTags(Iterable<T> elements, Iterable tags) {
- List ids = elements.map(_getId).toList(growable: false);
+ Future<void> removeTags(Iterable<T> elements, Iterable<Object?> tags) {
+ var ids = elements.map(_getId).toList(growable: false);
tags = tags.toList(growable: false);
var completer = Completer<void>();
var port = singleCompletePort(completer, timeout: _timeout);
@@ -233,7 +229,7 @@
return completer.future;
}
- Future<void> _addTags(List<int> ids, Iterable tags) {
+ Future<void> _addTags(List<int> ids, Iterable<Object?> tags) {
tags = tags.toList(growable: false);
var completer = Completer<void>();
var port = singleCompletePort(completer, timeout: _timeout);
@@ -250,53 +246,58 @@
/// In that case, at most the first `max` results are returned,
/// in whatever order the registry finds its results.
/// Otherwise all matching elements are returned.
- Future<List<T>> lookup({Iterable<Object> tags, int max}) {
+ Future<List<T>> lookup({Iterable<Object?>? tags, int? max}) async {
if (max != null && max < 1) {
throw RangeError.range(max, 1, null, 'max');
}
if (tags != null) tags = tags.toList(growable: false);
var completer = Completer<List<T>>();
- var port = singleCompletePort(completer, callback: (List response) {
+ var port = singleCompletePort(completer, callback: (List<T> response) {
// Response is even-length list of (id, element) pairs.
var cache = _cache;
var count = response.length ~/ 2;
- var result = List<T>(count);
- for (var i = 0; i < count; i++) {
- var id = response[i * 2] as int;
- var element = response[i * 2 + 1] as T;
- element = cache.register(id, element);
- result[i] = element;
- }
+ var result = List<T>.generate(
+ count,
+ (i) =>
+ cache.register(response[i * 2] as int, response[i * 2 + 1]) as T,
+ growable: false);
return result;
}, timeout: _timeout);
_commandPort.send(list4(_findValue, tags, max, port));
- return completer.future;
+ return await completer.future;
}
}
/// Isolate-local cache used by a [Registry].
///
-/// Maps between id-numbers and elements.
-/// An object is considered an element of the registry if it
+/// Maps between id numbers and elements.
+///
+/// Each instance of [Registry] has its own cache,
+/// and only considers elements part of the registry
+/// if they are registered in its cache.
+/// An object becomes registered either when calling
+/// [add] on that particular [Registry] instance,
+/// or when fetched using [lookup] through that
+/// registry instance.
class _RegistryCache {
// Temporary marker until an object gets an id.
static const int _beingAdded = -1;
- final Map<int, Object> id2object = HashMap();
- final Map<Object, int> object2id = HashMap.identity();
+ final Map<int, Object?> id2object = HashMap();
+ final Map<Object?, int> object2id = HashMap.identity();
- int id(Object object) {
+ int? id(Object? object) {
var result = object2id[object];
if (result == _beingAdded) return null;
return result;
}
- Object operator [](int id) => id2object[id];
+ Object? operator [](int id) => id2object[id];
// Register a pair of id/object in the cache.
// if the id is already in the cache, just return the existing
// object.
- Object register(int id, Object object) {
+ Object? register(int id, Object? object) {
object = id2object.putIfAbsent(id, () {
object2id[object] = id;
return object;
@@ -335,7 +336,7 @@
/// Maps id to entry. Each entry contains the id, the element, its tags,
/// and a capability required to remove it again.
final _entries = HashMap<int, _RegistryEntry>();
- final _tag2id = HashMap<Object, Set<int>>();
+ final _tag2id = HashMap<Object?, Set<int>>();
/// Create a new registry managed by the created [RegistryManager].
///
@@ -364,32 +365,44 @@
// Used as argument to putIfAbsent.
static Set<int> _createSet() => HashSet<int>();
- void _handleCommand(List command) {
+ void _handleCommand(List<dynamic> command) {
switch (command[0]) {
case _addValue:
- _add(command[1], command[2] as List, command[3] as SendPort);
+ _add(command[1], command[2] as List<Object?>?, command[3] as SendPort);
return;
case _removeValue:
- _remove(command[1], command[2] as Capability, command[3] as SendPort);
+ _remove(
+ command[1] as int,
+ command[2] as Capability,
+ command[3] as SendPort,
+ );
return;
case _addTagsValue:
- _addTags(command[1], command[2] as List, command[3] as SendPort);
+ _addTags(
+ command[1] as List<int>,
+ command[2] as List,
+ command[3] as SendPort,
+ );
return;
case _removeTagsValue:
- _removeTags(command[1], command[2] as List, command[3] as SendPort);
+ _removeTags(
+ command[1] as List<int>,
+ command[2] as List,
+ command[3] as SendPort,
+ );
return;
case _getTagsValue:
- _getTags(command[1], command[2] as SendPort);
+ _getTags(command[1] as int, command[2] as SendPort);
return;
case _findValue:
- _find(command[1] as List, command[2] as int, command[3] as SendPort);
+ _find(command[1] as List?, command[2] as int?, command[3] as SendPort);
return;
default:
throw UnsupportedError('Unknown command: ${command[0]}');
}
}
- void _add(Object object, List tags, SendPort replyPort) {
+ void _add(object, List? tags, SendPort replyPort) {
var id = ++_nextId;
var entry = _RegistryEntry(id, object);
_entries[id] = entry;
@@ -410,13 +423,12 @@
}
_entries.remove(id);
for (var tag in entry.tags) {
- _tag2id[tag].remove(id);
+ _tag2id[tag]!.remove(id);
}
replyPort.send(true);
}
- void _addTags(List<int> ids, List tags, SendPort replyPort) {
- assert(tags != null);
+ void _addTags(List<int> ids, List<Object?> tags, SendPort replyPort) {
assert(tags.isNotEmpty);
for (var id in ids) {
var entry = _entries[id];
@@ -431,7 +443,6 @@
}
void _removeTags(List<int> ids, List tags, SendPort replyPort) {
- assert(tags != null);
assert(tags.isNotEmpty);
for (var id in ids) {
var entry = _entries[id];
@@ -439,7 +450,7 @@
entry.tags.removeAll(tags);
}
for (var tag in tags) {
- Set tagIds = _tag2id[tag];
+ Set? tagIds = _tag2id[tag];
if (tagIds == null) continue;
tagIds.removeAll(ids);
}
@@ -475,7 +486,7 @@
return matchingIds;
}
- void _find(List tags, int max, SendPort replyPort) {
+ void _find(List? tags, int? max, SendPort replyPort) {
assert(max == null || max > 0);
var result = [];
if (tags == null || tags.isEmpty) {
@@ -489,12 +500,13 @@
return;
}
var matchingIds = _findTaggedIds(tags);
- max ??= matchingIds.length; // All results.
+
+ var actualMax = max ?? matchingIds.length; // All results.
for (var id in matchingIds) {
result.add(id);
- result.add(_entries[id].element);
- max--;
- if (max == 0) break;
+ result.add(_entries[id]!.element);
+ actualMax -= 1;
+ if (actualMax == 0) break;
}
replyPort.send(result);
}
@@ -510,8 +522,9 @@
/// Entry in [RegistryManager].
class _RegistryEntry {
final int id;
- final Object element;
+ final Object? element;
final Set tags = HashSet();
final Capability removeCapability = Capability();
+
_RegistryEntry(this.id, this.element);
}
diff --git a/lib/runner.dart b/lib/runner.dart
index e69b3ad..c902f86 100644
--- a/lib/runner.dart
+++ b/lib/runner.dart
@@ -27,16 +27,15 @@
/// Waits for the result of the call, and completes the returned future
/// with the result, whether it's a value or an error.
///
- /// If the returned future does not complete before `timeLimit` has passed,
+ /// If [timeout] is provided, and the returned future does not complete
+ /// before that duration has passed,
/// the [onTimeout] action is executed instead, and its result (whether it
/// returns or throws) is used as the result of the returned future.
- ///
- /// If `onTimeout` is omitted, a timeout will cause the returned future to
- /// complete with a [TimeoutException].
+ /// If [onTimeout] is omitted, it defaults to throwing a[TimeoutException].
///
/// The default implementation runs the function in the current isolate.
Future<R> run<R, P>(FutureOr<R> Function(P argument) function, P argument,
- {Duration timeout, FutureOr<R> Function() onTimeout}) {
+ {Duration? timeout, FutureOr<R> Function()? onTimeout}) {
var result = Future.sync(() => function(argument));
if (timeout != null) {
result = result.timeout(timeout, onTimeout: onTimeout);
diff --git a/lib/src/errors.dart b/lib/src/errors.dart
index 38316e7..4c76642 100644
--- a/lib/src/errors.dart
+++ b/lib/src/errors.dart
@@ -2,7 +2,6 @@
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
-// TODO(lrn): This should be in package:async?
/// Helper functions for working with errors.
///
/// The [MultiError] class combines multiple errors into one object,
@@ -18,6 +17,7 @@
// in the output.
// If there are too many errors, they will all get at least one line.
static const int _maxLines = 55;
+
// Minimum number of lines in the toString for each error.
static const int _minLinesPerError = 1;
@@ -40,16 +40,17 @@
///
/// The order of values is not preserved (if that is needed, use
/// [wait]).
- static Future<List<Object>> waitUnordered<T>(Iterable<Future<T>> futures,
- {void Function(T successResult) cleanUp}) {
- Completer<List<Object>> completer;
+ static Future<List<Object?>> waitUnordered<T>(Iterable<Future<T>> futures,
+ {void Function(T successResult)? cleanUp}) {
+ var completer = Completer<List<Object?>>();
var count = 0;
var errors = 0;
var values = 0;
// Initialized to `new List(count)` when count is known.
// Filled up with values on the left, errors on the right.
// Order is not preserved.
- List<Object> results;
+ List<Object?> results = const <Never>[];
+
void checkDone() {
if (errors + values < count) return;
if (errors == 0) {
@@ -60,7 +61,7 @@
completer.completeError(MultiError(errorList));
}
- var handleValue = (T v) {
+ void handleValue(T v) {
// If this fails because [results] is null, there is a future
// which breaks the Future API by completing immediately when
// calling Future.then, probably by misusing a synchronous completer.
@@ -69,23 +70,25 @@
Future.sync(() => cleanUp(v));
}
checkDone();
- };
- var handleError = (e, s) {
+ }
+
+ void handleError(Object e, StackTrace s) {
if (errors == 0 && cleanUp != null) {
for (var i = 0; i < values; i++) {
var value = results[i];
- if (value != null) Future.sync(() => cleanUp(value));
+ if (value != null) Future.sync(() => cleanUp(value as T));
}
}
results[results.length - ++errors] = e;
checkDone();
- };
+ }
+
for (var future in futures) {
count++;
- future.then(handleValue, onError: handleError);
+ future.then<void>(handleValue, onError: handleError);
}
- if (count == 0) return Future.value(List(0));
- results = List(count);
+ if (count == 0) return Future.value(List.filled(0, null));
+ results = List.filled(count, null);
completer = Completer();
return completer.future;
}
@@ -99,16 +102,17 @@
/// The order of values is preserved, and if any error occurs, the
/// [MultiError.errors] list will have errors in the corresponding slots,
/// and `null` for non-errors.
- Future<List<Object>> wait<T>(Iterable<Future<T>> futures,
- {void Function(T successResult) cleanUp}) {
- Completer<List<Object>> completer;
+ Future<List<Object?>> wait<T>(Iterable<Future<T>> futures,
+ {void Function(T successResult)? cleanUp}) {
+ var completer = Completer<List<Object?>>();
var count = 0;
var hasError = false;
var completed = 0;
// Initialized to `new List(count)` when count is known.
// Filled with values until the first error, then cleared
// and filled with errors.
- List<Object> results;
+ List<Object?> results = const <Never>[];
+
void checkDone() {
completed++;
if (completed < count) return;
@@ -122,7 +126,7 @@
for (var future in futures) {
var i = count;
count++;
- future.then((v) {
+ future.then<void>((v) {
if (!hasError) {
results[i] = v;
} else if (cleanUp != null) {
@@ -134,18 +138,18 @@
if (cleanUp != null) {
for (var i = 0; i < results.length; i++) {
var result = results[i];
- if (result != null) Future.sync(() => cleanUp(result));
+ if (result != null) Future.sync(() => cleanUp(result as T));
}
}
- results = List<Object>(count);
+ results = List<Object?>.filled(count, null);
hasError = true;
}
results[i] = e;
checkDone();
});
}
- if (count == 0) return Future.value(List(0));
- results = List<T>(count);
+ if (count == 0) return Future.value(List.filled(0, null));
+ results = List<T?>.filled(count, null);
completer = Completer();
return completer.future;
}
diff --git a/lib/src/raw_receive_port_multiplexer.dart b/lib/src/raw_receive_port_multiplexer.dart
index 3db8b01..bccebf8 100644
--- a/lib/src/raw_receive_port_multiplexer.dart
+++ b/lib/src/raw_receive_port_multiplexer.dart
@@ -31,12 +31,12 @@
class _MultiplexRawReceivePort implements RawReceivePort {
final RawReceivePortMultiplexer _multiplexer;
final int _id;
- Function _handler;
+ Function? _handler;
_MultiplexRawReceivePort(this._multiplexer, this._id, this._handler);
@override
- set handler(Function handler) {
+ set handler(Function? handler) {
_handler = handler;
}
@@ -49,13 +49,14 @@
SendPort get sendPort => _multiplexer._createSendPort(_id);
void _invokeHandler(message) {
- _handler(message);
+ _handler?.call(message);
}
}
class _MultiplexSendPort implements SendPort {
final SendPort _sendPort;
final int _id;
+
_MultiplexSendPort(this._id, this._sendPort);
@override
@@ -75,7 +76,7 @@
_port.handler = _multiplexResponse;
}
- RawReceivePort createRawReceivePort([void Function(dynamic) handler]) {
+ RawReceivePort createRawReceivePort([void Function(dynamic)? handler]) {
var id = _nextId++;
var result = _MultiplexRawReceivePort(this, id, handler);
_map[id] = result;
@@ -87,7 +88,7 @@
}
void _multiplexResponse(list) {
- int id = list[0];
+ var id = list[0];
var message = list[1];
var receivePort = _map[id];
// If the receive port is closed, messages are dropped, just as for
diff --git a/lib/src/util.dart b/lib/src/util.dart
index 989dd28..cc4ba25 100644
--- a/lib/src/util.dart
+++ b/lib/src/util.dart
@@ -4,36 +4,43 @@
/// Utility functions.
+/// A [Comparator] that asserts that its first argument is comparable.
+///
+/// The function behaves just like [List.sort]'s
+/// default comparison function. It is entirely dynamic in its testing.
+///
+/// Should be used when optimistically comparing object that are assumed
+/// to be comparable.
+/// If the elements are known to be comparable, use [compareComparable].
+int defaultCompare(Object? value1, Object? value2) =>
+ (value1 as Comparable<Object?>).compareTo(value2);
+
/// Ignore an argument.
///
/// Can be used to drop the result of a future like `future.then(ignore)`.
void ignore(_) {}
/// Create a single-element fixed-length list.
-List<Object> list1(Object v1) => List(1)..[0] = v1;
+List<T> list1<T extends Object?>(T v1) => List<T>.filled(1, v1);
/// Create a two-element fixed-length list.
-List<Object> list2(Object v1, Object v2) => List(2)
- ..[0] = v1
- ..[1] = v2;
+List<T> list2<T extends Object?>(T v1, T v2) => List<T>.filled(2, v1)..[1] = v2;
/// Create a three-element fixed-length list.
-List<Object> list3(Object v1, Object v2, Object v3) => List(3)
- ..[0] = v1
+List<T> list3<T extends Object?>(T v1, T v2, T v3) => List<T>.filled(3, v1)
..[1] = v2
..[2] = v3;
/// Create a four-element fixed-length list.
-List<Object> list4(Object v1, Object v2, Object v3, Object v4) => List(4)
- ..[0] = v1
- ..[1] = v2
- ..[2] = v3
- ..[3] = v4;
+List<T> list4<T extends Object?>(T v1, T v2, T v3, T v4) =>
+ List<T>.filled(4, v1)
+ ..[1] = v2
+ ..[2] = v3
+ ..[3] = v4;
/// Create a five-element fixed-length list.
-List<Object> list5(Object v1, Object v2, Object v3, Object v4, Object v5) =>
- List(5)
- ..[0] = v1
+List<T> list5<T extends Object?>(T v1, T v2, T v3, T v4, T v5) =>
+ List<T>.filled(5, v1)
..[1] = v2
..[2] = v3
..[3] = v4
diff --git a/pubspec.yaml b/pubspec.yaml
index c5c4e08..a1ca11f 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,13 +1,13 @@
name: isolate
-version: 2.0.3
+version: 2.1.0
author: Dart Team <misc@dartlang.org>
description: >-
Utility functions and classes related to the 'dart:isolate' library.
homepage: https://github.com/dart-lang/isolate
environment:
- sdk: '>=2.3.0 <3.0.0'
+ sdk: '>=2.12.0 <3.0.0'
dev_dependencies:
- pedantic: ^1.0.0
+ lints: ^1.0.0
test: ^1.0.0
diff --git a/test/isolaterunner_test.dart b/test/isolaterunner_test.dart
index e205bc2..a292eb5 100644
--- a/test/isolaterunner_test.dart
+++ b/test/isolaterunner_test.dart
@@ -2,8 +2,6 @@
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
-library isolate.test.isolaterunner_test;
-
import 'dart:isolate' show Capability;
import 'package:isolate/isolate_runner.dart';
@@ -37,8 +35,8 @@
// Check that each isolate has its own _global variable.
return Future.wait(Iterable.generate(2, (_) => IsolateRunner.spawn()))
.then((runners) {
- Future runAll(Function(IsolateRunner runner, int index) action) {
- var indices = Iterable.generate(runners.length);
+ Future runAll(Future Function(IsolateRunner runner, int index) action) {
+ final indices = Iterable<int>.generate(runners.length);
return Future.wait(indices.map((i) => action(runners[i], i)));
}
@@ -107,6 +105,8 @@
dynamic id(x) => x;
-var _global;
+Object? _global;
+
dynamic getGlobal(_) => _global;
+
void setGlobal(v) => _global = v;
diff --git a/test/load_balancer_test.dart b/test/load_balancer_test.dart
new file mode 100644
index 0000000..4b5a3d8
--- /dev/null
+++ b/test/load_balancer_test.dart
@@ -0,0 +1,158 @@
+// Copyright (c) 2021, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:isolate/isolate.dart';
+import 'package:test/test.dart';
+
+void main() {
+ test('Simple use', () async {
+ // The simplest possible case. Should not throw anywhere.
+ // Load balancer with at least one isolate-runner.
+ var lb = await LoadBalancer.create(1, () => IsolateRunner.spawn());
+ // Run at least one task.
+ await lb.run(runTask, null);
+ // Close it.
+ await lb.close();
+ });
+
+ /// Run multiple tasks one at a time.
+ test("Run multiple indivudual jobs", () async {
+ var lb = await createIsolateRunners(4);
+ List<List<int>> results =
+ await Future.wait([for (var i = 0; i < 10; i++) lb.run(getId, i)]);
+ // All request numbers should be accounted for.
+ var first = {for (var result in results) result.first};
+ expect(first, {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+ // All isolate runners should have been used.
+ var last = {for (var result in results) result.last};
+ expect(last, {1, 2, 3, 4});
+ await lb.close();
+ });
+ test("Run multiple - zero", () async {
+ var lb = createLocalRunners(4);
+ expect(() => lb.runMultiple(0, runTask, 0), throwsArgumentError);
+ });
+ test("Run multiple - more than count", () async {
+ var lb = createLocalRunners(4);
+ expect(() => lb.runMultiple(5, runTask, 0), throwsArgumentError);
+ });
+ test("Run multiple - 1", () async {
+ var lb = createLocalRunners(4);
+ var results = await Future.wait(lb.runMultiple(1, LocalRunner.getId, 0));
+ expect(results, hasLength(1));
+ });
+ test("Run multiple - more", () async {
+ var lb = createLocalRunners(4);
+ var results = await Future.wait(lb.runMultiple(2, LocalRunner.getId, 0));
+ expect(results, hasLength(2));
+ // Different runners.
+ expect({for (var result in results) result.last}, hasLength(2));
+ });
+ test("Run multiple - all", () async {
+ var lb = createLocalRunners(4);
+ var results = await Future.wait(lb.runMultiple(4, LocalRunner.getId, 0));
+ expect(results, hasLength(4));
+ // Different runners.
+ expect({for (var result in results) result.last}, hasLength(4));
+ });
+ test("Always lowest load runner", () async {
+ var lb = createLocalRunners(4);
+ // Run tasks with input numbers cooresponding to the other tasks they
+ // expect to share runner with.
+ // Loads: 0, 0, 0, 0.
+ var r1 = lb.run(LocalRunner.getId, 0, load: 100);
+ // Loads: 0, 0, 0, 100(1).
+ var r2 = lb.run(LocalRunner.getId, 1, load: 50);
+ // Loads: 0, 0, 50(2), 100(1).
+ var r3 = lb.run(LocalRunner.getId, 2, load: 25);
+ // Loads: 0, 25(3), 50(2), 100(1).
+ var r4 = lb.run(LocalRunner.getId, 3, load: 10);
+ // Loads: 10(4), 25(3), 50(2), 100(1).
+ var r5 = lb.run(LocalRunner.getId, 3, load: 10);
+ // Loads: 20(4, 4), 25(3), 50(2), 100(1).
+ var r6 = lb.run(LocalRunner.getId, 3, load: 90);
+ // Loads: 25(3), 50(2), 100(1), 110(4, 4, 4).
+ var r7 = lb.run(LocalRunner.getId, 2, load: 90);
+ // Loads: 50(2), 100(1), 110(4, 4, 4), 115(3, 3).
+ var r8 = lb.run(LocalRunner.getId, 1, load: 64);
+ // Loads: 100(1), 110(4, 4, 4), 114(2, 2), 115(3, 3).
+ var r9 = lb.run(LocalRunner.getId, 0, load: 100);
+ // Loads: 110(4, 4, 4), 114(2, 2), 115(3, 3), 200(1, 1).
+ var results = await Future.wait([r1, r2, r3, r4, r5, r6, r7, r8, r9]);
+
+ // Check that tasks with the same input numbers ran on the same runner.
+ var runnerIds = [-1, -1, -1, -1];
+ for (var result in results) {
+ var expectedRunner = result.first;
+ var actualRunnerId = result.last;
+ var seenId = runnerIds[expectedRunner];
+ if (seenId == -1) {
+ runnerIds[expectedRunner] = actualRunnerId;
+ } else if (seenId != actualRunnerId) {
+ fail("Task did not run on lowest loaded runner\n$result");
+ }
+ }
+ // Check that all four runners were used.
+ var uniqueRunnerIds = {...runnerIds};
+ expect(uniqueRunnerIds, {1, 2, 3, 4});
+ await lb.close();
+ });
+}
+
+void runTask(_) {
+ // Trivial task.
+}
+
+// An isolate local ID.
+int localId = -1;
+void setId(int value) {
+ localId = value;
+}
+
+// Request a response including input and local ID.
+// Allows detecting which isolate the request is run on.
+List<int> getId(int input) => [input, localId];
+
+/// Creates isolate runners with `localId` in the range 1..count.
+Future<LoadBalancer> createIsolateRunners(int count) async {
+ var runners = <Runner>[];
+ for (var i = 1; i <= count; i++) {
+ var runner = await IsolateRunner.spawn();
+ await runner.run(setId, i);
+ runners.add(runner);
+ }
+ return LoadBalancer(runners);
+}
+
+LoadBalancer createLocalRunners(int count) =>
+ LoadBalancer([for (var i = 1; i <= count; i++) LocalRunner(i)]);
+
+class LocalRunner implements Runner {
+ final int id;
+ LocalRunner(this.id);
+
+ static Future<List<int>> getId(int input) async =>
+ [input, Zone.current[#runner].id as int];
+
+ @override
+ Future<void> close() async {}
+
+ @override
+ Future<R> run<R, P>(FutureOr<R> Function(P argument) function, P argument,
+ {Duration? timeout, FutureOr<R> Function()? onTimeout}) {
+ return runZoned(() {
+ var result = Future.sync(() => function(argument));
+ if (timeout != null) {
+ result = result.timeout(timeout, onTimeout: onTimeout ?? _throwTimeout);
+ }
+ return result;
+ }, zoneValues: {#runner: this});
+ }
+
+ static Never _throwTimeout() {
+ throw TimeoutException("timeout");
+ }
+}
diff --git a/test/ports_test.dart b/test/ports_test.dart
index bc133c4..9032c76 100644
--- a/test/ports_test.dart
+++ b/test/ports_test.dart
@@ -2,8 +2,6 @@
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
-library isolate.test.ports_test;
-
import 'dart:async';
import 'dart:isolate';
@@ -25,25 +23,35 @@
var completer = Completer.sync();
var p = singleCallbackPort(completer.complete);
p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 42);
});
});
+ test('Value without timeout non-nullable', () {
+ var completer = Completer<int>.sync();
+ var p = singleCallbackPort(completer.complete);
+ p.send(42);
+ return completer.future.then<void>((int v) {
+ expect(v, 42);
+ });
+ });
+
+ test('Value without timeout nullable', () {
+ var completer = Completer<int?>.sync();
+ var p = singleCallbackPort(completer.complete);
+ p.send(null);
+ return completer.future.then<void>((int? v) {
+ expect(v, null);
+ });
+ });
+
test('FirstValue', () {
var completer = Completer.sync();
var p = singleCallbackPort(completer.complete);
p.send(42);
p.send(37);
- return completer.future.then((v) {
- expect(v, 42);
- });
- });
- test('Value', () {
- var completer = Completer.sync();
- var p = singleCallbackPort(completer.complete);
- p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 42);
});
});
@@ -52,7 +60,7 @@
var completer = Completer.sync();
var p = singleCallbackPort(completer.complete, timeout: _ms * 500);
p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 42);
});
});
@@ -61,7 +69,7 @@
var completer = Completer.sync();
singleCallbackPort(completer.complete,
timeout: _ms * 100, timeoutValue: 37);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 37);
});
});
@@ -71,10 +79,31 @@
var p = singleCallbackPort(completer.complete,
timeout: _ms * 100, timeoutValue: 37);
Timer(_ms * 500, () => p.send(42));
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 37);
});
});
+
+ /// invalid null is a compile time error
+ test('TimeoutFirst with valid null', () {
+ var completer = Completer.sync();
+ var p = singleCallbackPort(completer.complete,
+ timeout: _ms * 100, timeoutValue: null);
+ Timer(_ms * 500, () => p.send(42));
+ return completer.future.then<void>((v) {
+ expect(v, null);
+ });
+ });
+
+ /// invalid null is a compile time error
+ test('TimeoutFirstWithTimeout with valid null', () {
+ var completer = Completer.sync();
+ var p = singleCallbackPortWithTimeout(completer.complete, _ms * 100, null);
+ Timer(_ms * 500, () => p.send(42));
+ return completer.future.then<void>((v) {
+ expect(v, null);
+ });
+ });
}
void testSingleCompletePort() {
@@ -82,7 +111,7 @@
var completer = Completer.sync();
var p = singleCompletePort(completer);
p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 42);
});
});
@@ -94,7 +123,7 @@
return 87;
});
p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 87);
});
});
@@ -106,7 +135,7 @@
return Future.delayed(_ms * 500, () => 88);
});
p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 88);
});
});
@@ -118,7 +147,7 @@
throw 89;
});
p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) async {
fail('unreachable');
}, onError: (e, s) {
expect(e, 89);
@@ -132,7 +161,7 @@
return Future.error(90);
});
p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((_) {
fail('unreachable');
}, onError: (e, s) {
expect(e, 90);
@@ -144,7 +173,7 @@
var p = singleCompletePort(completer);
p.send(42);
p.send(37);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 42);
});
});
@@ -157,7 +186,7 @@
});
p.send(42);
p.send(37);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 87);
});
});
@@ -166,7 +195,7 @@
var completer = Completer.sync();
var p = singleCompletePort(completer, timeout: _ms * 500);
p.send(42);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 42);
});
});
@@ -174,7 +203,7 @@
test('Timeout', () {
var completer = Completer.sync();
singleCompletePort(completer, timeout: _ms * 100);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e is TimeoutException, isTrue);
@@ -184,7 +213,7 @@
test('TimeoutCallback', () {
var completer = Completer.sync();
singleCompletePort(completer, timeout: _ms * 100, onTimeout: () => 87);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 87);
});
});
@@ -193,7 +222,7 @@
var completer = Completer.sync();
singleCompletePort(completer,
timeout: _ms * 100, onTimeout: () => throw 91);
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e, 91);
@@ -204,7 +233,7 @@
var completer = Completer.sync();
singleCompletePort(completer,
timeout: _ms * 100, onTimeout: () => Future.value(87));
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 87);
});
});
@@ -213,7 +242,7 @@
var completer = Completer.sync();
singleCompletePort(completer,
timeout: _ms * 100, onTimeout: () => Future.error(92));
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e, 92);
@@ -225,7 +254,7 @@
singleCompletePort(completer,
timeout: _ms * 100,
onTimeout: () => Future.delayed(_ms * 500, () => 87));
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 87);
});
});
@@ -235,7 +264,7 @@
singleCompletePort(completer,
timeout: _ms * 100,
onTimeout: () => Future.delayed(_ms * 500, () => throw 87));
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e, 87);
@@ -247,26 +276,70 @@
var p =
singleCompletePort(completer, timeout: _ms * 100, onTimeout: () => 37);
Timer(_ms * 500, () => p.send(42));
- return completer.future.then((v) {
+ return completer.future.then<void>((v) {
expect(v, 37);
});
});
+
+ test('TimeoutFirst with valid null', () {
+ var completer = Completer<int?>.sync();
+ var p = singleCompletePort(completer,
+ timeout: _ms * 100, onTimeout: () => null);
+ Timer(_ms * 500, () => p.send(42));
+ return expectLater(completer.future, completion(null));
+ });
+
+ test('TimeoutFirst with invalid null', () {
+ var completer = Completer<int>.sync();
+
+ // Example of incomplete generic parameters promotion.
+ // Same code with [singleCompletePort<int, dynamic>] is a compile time error.
+ var p = singleCompletePort(
+ completer,
+ timeout: _ms * 100,
+ onTimeout: () => null,
+ );
+ Timer(_ms * 500, () => p.send(42));
+ return expectLater(completer.future, throwsA(isA<TypeError>()));
+ });
}
void testSingleResponseFuture() {
test('FutureValue', () {
return singleResponseFuture((SendPort p) {
p.send(42);
- }).then((v) {
+ }).then<void>((v) {
expect(v, 42);
});
});
+ test('FutureValue without timeout', () {
+ return singleResponseFuture<int>((SendPort p) {
+ p.send(42);
+ }).then<void>((v) {
+ expect(v, 42);
+ });
+ });
+
+ test('FutureValue without timeout valid null', () {
+ return singleResponseFuture<int?>((SendPort p) {
+ p.send(null);
+ }).then<void>((v) {
+ expect(v, null);
+ });
+ });
+
+ test('FutureValue without timeout invalid null', () {
+ return expectLater(singleResponseFuture<int>((SendPort p) {
+ p.send(null);
+ }), throwsA(isA<TypeError>()));
+ });
+
test('FutureValueFirst', () {
return singleResponseFuture((SendPort p) {
p.send(42);
p.send(37);
- }).then((v) {
+ }).then<void>((v) {
expect(v, 42);
});
});
@@ -274,7 +347,7 @@
test('FutureError', () {
return singleResponseFuture((SendPort p) {
throw 93;
- }).then((v) {
+ }).then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e, 93);
@@ -285,7 +358,7 @@
return singleResponseFuture((SendPort p) {
// no-op.
}, timeout: _ms * 100)
- .then((v) {
+ .then<void>((v) {
expect(v, null);
});
});
@@ -294,7 +367,25 @@
return singleResponseFuture((SendPort p) {
// no-op.
}, timeout: _ms * 100, timeoutValue: 42)
- .then((v) {
+ .then<void>((int? v) {
+ expect(v, 42);
+ });
+ });
+
+ test('FutureTimeoutValue with valid null timeoutValue', () {
+ return singleResponseFuture<int?>((SendPort p) {
+ // no-op.
+ }, timeout: _ms * 100, timeoutValue: null)
+ .then<void>((int? v) {
+ expect(v, null);
+ });
+ });
+
+ test('FutureTimeoutValue with non-null timeoutValue', () {
+ return singleResponseFuture<int>((SendPort p) {
+ // no-op.
+ }, timeout: _ms * 100, timeoutValue: 42)
+ .then<void>((int v) {
expect(v, 42);
});
});
@@ -304,7 +395,7 @@
test('Value', () {
return singleResultFuture((SendPort p) {
sendFutureResult(Future.value(42), p);
- }).then((v) {
+ }).then<void>((v) {
expect(v, 42);
});
});
@@ -313,7 +404,7 @@
return singleResultFuture((SendPort p) {
sendFutureResult(Future.value(42), p);
sendFutureResult(Future.value(37), p);
- }).then((v) {
+ }).then<void>((v) {
expect(v, 42);
});
});
@@ -321,7 +412,7 @@
test('Error', () {
return singleResultFuture((SendPort p) {
sendFutureResult(Future.error(94), p);
- }).then((v) {
+ }).then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e is RemoteError, isTrue);
@@ -332,7 +423,7 @@
return singleResultFuture((SendPort p) {
sendFutureResult(Future.error(95), p);
sendFutureResult(Future.error(96), p);
- }).then((v) {
+ }).then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e is RemoteError, isTrue);
@@ -342,7 +433,7 @@
test('Error', () {
return singleResultFuture((SendPort p) {
throw 93;
- }).then((v) {
+ }).then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e is RemoteError, isTrue);
@@ -353,7 +444,7 @@
return singleResultFuture((SendPort p) {
// no-op.
}, timeout: _ms * 100)
- .then((v) {
+ .then<void>((v) {
fail('unreachable');
}, onError: (e, s) {
expect(e is TimeoutException, isTrue);
@@ -363,15 +454,14 @@
test('TimeoutValue', () {
return singleResultFuture((SendPort p) {
// no-op.
- }, timeout: _ms * 100, onTimeout: () => 42).then((v) {
+ }, timeout: _ms * 100, onTimeout: () => 42).then<void>((v) {
expect(v, 42);
});
});
test('TimeoutError', () {
- return singleResultFuture((SendPort p) {
- // no-op.
- }, timeout: _ms * 100, onTimeout: () => throw 97).then((v) {
+ return singleResultFuture((SendPort p) {},
+ timeout: _ms * 100, onTimeout: () => throw 97).then<void>((v) {
expect(v, 42);
}, onError: (e, s) {
expect(e, 97);
@@ -381,34 +471,34 @@
void testSingleResponseChannel() {
test('Value', () {
- var channel = SingleResponseChannel();
+ final channel = SingleResponseChannel();
channel.port.send(42);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
expect(v, 42);
});
});
test('ValueFirst', () {
- var channel = SingleResponseChannel();
+ final channel = SingleResponseChannel();
channel.port.send(42);
channel.port.send(37);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
expect(v, 42);
});
});
test('ValueCallback', () {
- var channel = SingleResponseChannel(callback: (v) => 2 * v);
+ final channel = SingleResponseChannel(callback: ((v) => 2 * (v as num)));
channel.port.send(42);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
expect(v, 84);
});
});
test('ErrorCallback', () {
- var channel = SingleResponseChannel(callback: (v) => throw 42);
+ final channel = SingleResponseChannel(callback: ((v) => throw 42));
channel.port.send(37);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
fail('unreachable');
}, onError: (v, s) {
expect(v, 42);
@@ -416,17 +506,18 @@
});
test('AsyncValueCallback', () {
- var channel = SingleResponseChannel(callback: (v) => Future.value(2 * v));
+ final channel =
+ SingleResponseChannel(callback: ((v) => Future.value(2 * (v as num))));
channel.port.send(42);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
expect(v, 84);
});
});
test('AsyncErrorCallback', () {
- var channel = SingleResponseChannel(callback: (v) => Future.error(42));
+ final channel = SingleResponseChannel(callback: ((v) => Future.error(42)));
channel.port.send(37);
- return channel.result.then((v) {
+ return channel.result.then<void>((_) {
fail('unreachable');
}, onError: (v, s) {
expect(v, 42);
@@ -434,16 +525,16 @@
});
test('Timeout', () {
- var channel = SingleResponseChannel(timeout: _ms * 100);
- return channel.result.then((v) {
+ final channel = SingleResponseChannel(timeout: _ms * 100);
+ return channel.result.then<void>((v) {
expect(v, null);
});
});
test('TimeoutThrow', () {
- var channel =
+ final channel =
SingleResponseChannel(timeout: _ms * 100, throwOnTimeout: true);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
fail('unreachable');
}, onError: (v, s) {
expect(v is TimeoutException, isTrue);
@@ -451,12 +542,12 @@
});
test('TimeoutThrowOnTimeoutAndValue', () {
- var channel = SingleResponseChannel(
+ final channel = SingleResponseChannel(
timeout: _ms * 100,
throwOnTimeout: true,
onTimeout: () => 42,
timeoutValue: 42);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
fail('unreachable');
}, onError: (v, s) {
expect(v is TimeoutException, isTrue);
@@ -464,32 +555,32 @@
});
test('TimeoutOnTimeout', () {
- var channel =
+ final channel =
SingleResponseChannel(timeout: _ms * 100, onTimeout: () => 42);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
expect(v, 42);
});
});
test('TimeoutOnTimeoutAndValue', () {
- var channel = SingleResponseChannel(
+ final channel = SingleResponseChannel(
timeout: _ms * 100, onTimeout: () => 42, timeoutValue: 37);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
expect(v, 42);
});
});
test('TimeoutValue', () {
- var channel = SingleResponseChannel(timeout: _ms * 100, timeoutValue: 42);
- return channel.result.then((v) {
+ final channel = SingleResponseChannel(timeout: _ms * 100, timeoutValue: 42);
+ return channel.result.then<void>((v) {
expect(v, 42);
});
});
test('TimeoutOnTimeoutError', () {
- var channel =
+ final channel =
SingleResponseChannel(timeout: _ms * 100, onTimeout: () => throw 42);
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
fail('unreachable');
}, onError: (v, s) {
expect(v, 42);
@@ -497,17 +588,17 @@
});
test('TimeoutOnTimeoutAsync', () {
- var channel = SingleResponseChannel(
+ final channel = SingleResponseChannel(
timeout: _ms * 100, onTimeout: () => Future.value(42));
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
expect(v, 42);
});
});
test('TimeoutOnTimeoutAsyncError', () {
- var channel = SingleResponseChannel(
+ final channel = SingleResponseChannel(
timeout: _ms * 100, onTimeout: () => Future.error(42));
- return channel.result.then((v) {
+ return channel.result.then<void>((v) {
fail('unreachable');
}, onError: (v, s) {
expect(v, 42);
diff --git a/test/registry_test.dart b/test/registry_test.dart
index 9e3cdec..763bc9c 100644
--- a/test/registry_test.dart
+++ b/test/registry_test.dart
@@ -2,14 +2,11 @@
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
-library isolate.test.registry_test;
-
import 'dart:async';
import 'dart:isolate';
import 'package:isolate/isolate_runner.dart';
import 'package:isolate/registry.dart';
-
import 'package:test/test.dart';
const _ms = Duration(milliseconds: 1);
@@ -44,7 +41,7 @@
return registry.add(element, tags: [tag]);
}).then((_) {
return registry.lookup();
- }).then((all) {
+ }).then<void>((all) {
expect(all.length, 10);
expect(all.map((v) => v.id).toList()..sort(),
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
@@ -60,7 +57,7 @@
return registry.add(element, tags: [tag]);
}).then((_) {
return registry.lookup(tags: [Oddity.odd]);
- }).then((all) {
+ }).then<void>((all) {
expect(all.length, 5);
expect(all.map((v) => v.id).toList()..sort(), [1, 3, 5, 7, 9]);
}).whenComplete(regman.close);
@@ -75,7 +72,7 @@
return registry.add(element, tags: [tag]);
}).then((_) {
return registry.lookup(max: 5);
- }).then((all) {
+ }).then<void>((all) {
expect(all.length, 5);
}).whenComplete(regman.close);
});
@@ -93,7 +90,7 @@
return registry.add(element, tags: tags);
}).then((_) {
return registry.lookup(tags: [2, 3]);
- }).then((all) {
+ }).then<void>((all) {
expect(all.length, 5);
expect(all.map((v) => v.id).toList()..sort(), [0, 6, 12, 18, 24]);
}).whenComplete(regman.close);
@@ -112,7 +109,7 @@
return registry.add(element, tags: tags);
}).then((_) {
return registry.lookup(tags: [2, 3], max: 3);
- }).then((all) {
+ }).then<void>((all) {
expect(all.length, 3);
expect(all.every((v) => (v.id % 6) == 0), isTrue);
}).whenComplete(regman.close);
@@ -126,7 +123,7 @@
var object = Object();
return registry.add(object).then((_) {
return registry.lookup();
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, hasLength(1));
expect(entries.first, same(object));
}).whenComplete(regman.close);
@@ -141,7 +138,7 @@
var objects = [object1, object2, object3];
return Future.wait(objects.map(registry.add)).then((_) {
return registry.lookup();
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, hasLength(3));
for (var entry in entries) {
expect(entry, isIn(objects));
@@ -155,7 +152,7 @@
var object = Object();
return registry.add(object).then((_) {
return registry.add(object);
- }).then((_) {
+ }).then<void>((_) {
fail('Unreachable');
}, onError: (e, s) {
expect(e, isStateError);
@@ -175,7 +172,7 @@
return registry.add(object2);
}).then((_) {
return registry.lookup();
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, hasLength(2));
var entry1 = entries.first;
var entry2 = entries.last;
@@ -196,7 +193,7 @@
return registry.add(object);
}).then((_) {
return registry.lookup();
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, hasLength(1));
expect(entries.first, same(object));
}).whenComplete(regman.close);
@@ -214,18 +211,18 @@
return registry.add(object3, tags: [4, 5, 6, 7]);
}).then((_) {
return registry.lookup(tags: [3]);
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, hasLength(2));
expect(entries.first == object1 || entries.last == object1, isTrue);
expect(entries.first == object2 || entries.last == object2, isTrue);
}).then((_) {
return registry.lookup(tags: [2]);
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, hasLength(1));
expect(entries.first, same(object2));
}).then((_) {
return registry.lookup(tags: [3, 6]);
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, hasLength(1));
expect(entries.first, same(object2));
}).whenComplete(regman.close);
@@ -246,7 +243,7 @@
}).then((removeSuccess) {
expect(removeSuccess, isTrue);
return registry.lookup();
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, isEmpty);
}).whenComplete(regman.close);
});
@@ -261,7 +258,7 @@
expect(entries.first, same(object));
return registry.remove(object, Capability());
});
- }).then((removeSuccess) {
+ }).then<void>((removeSuccess) {
expect(removeSuccess, isFalse);
}).whenComplete(regman.close);
});
@@ -285,7 +282,7 @@
return registry.removeTags([object], ['x']);
}).then((_) {
return registry.lookup(tags: ['x']);
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, isEmpty);
}).whenComplete(regman.close);
});
@@ -349,6 +346,7 @@
}
var _regmen = {};
+
Registry createRegMan(id) {
var regman = RegistryManager();
_regmen[id] = regman;
@@ -370,7 +368,7 @@
expect(entries, hasLength(1));
expect(entries.first, same(object));
return registry.remove(entries.first, removeCapability);
- }).then((removeSuccess) {
+ }).then<void>((removeSuccess) {
expect(removeSuccess, isTrue);
});
});
@@ -388,7 +386,7 @@
var regman = RegistryManager(timeout: _ms * 500);
var registry = regman.registry;
regman.close();
- return registry.add(Object()).then((_) {
+ return registry.add(Object()).then<void>((_) {
fail('unreachable');
}, onError: (e, s) {
expect(e is TimeoutException, isTrue);
@@ -401,7 +399,7 @@
var object = Object();
return registry.add(object).then((rc) {
regman.close();
- return registry.remove(object, rc).then((_) {
+ return registry.remove(object, rc).then<void>((_) {
fail('unreachable');
}, onError: (e, s) {
expect(e is TimeoutException, isTrue);
@@ -415,7 +413,7 @@
var object = Object();
return registry.add(object).then((rc) {
regman.close();
- return registry.addTags([object], ['x']).then((_) {
+ return registry.addTags([object], ['x']).then<void>((_) {
fail('unreachable');
}, onError: (e, s) {
expect(e is TimeoutException, isTrue);
@@ -429,7 +427,7 @@
var object = Object();
return registry.add(object).then((rc) {
regman.close();
- return registry.removeTags([object], ['x']).then((_) {
+ return registry.removeTags([object], ['x']).then<void>((_) {
fail('unreachable');
}, onError: (e, s) {
expect(e is TimeoutException, isTrue);
@@ -441,7 +439,7 @@
var regman = RegistryManager(timeout: _ms * 500);
var registry = regman.registry;
regman.close();
- registry.lookup().then((_) {
+ registry.lookup().then<void>((_) {
fail('unreachable');
}, onError: (e, s) {
expect(e is TimeoutException, isTrue);
@@ -455,7 +453,7 @@
var registry1 = regman.registry;
var registry2 = regman.registry;
var l1 = ['x'];
- var l2;
+ dynamic l2;
return registry1.add(l1, tags: ['y']).then((removeCapability) {
return registry2.lookup().then((entries) {
expect(entries, hasLength(1));
@@ -471,7 +469,7 @@
}).then((removeSuccess) {
expect(removeSuccess, isTrue);
return registry1.lookup();
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, isEmpty);
});
}).whenComplete(regman.close);
@@ -503,7 +501,7 @@
}).then((removeSuccess) {
expect(removeSuccess, isTrue);
return registry2.lookup();
- }).then((entries) {
+ }).then<void>((entries) {
expect(entries, isEmpty);
});
}).whenComplete(regman.close);
@@ -526,9 +524,12 @@
class Element {
final int id;
+
Element(this.id);
+
@override
int get hashCode => id;
+
@override
bool operator ==(Object other) => other is Element && id == other.id;
}
