-
Notifications
You must be signed in to change notification settings - Fork 38
/
worker.dart
78 lines (69 loc) · 2.05 KB
/
worker.dart
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
import 'dart:async';
import 'dart:isolate';
import 'package:ffi/ffi.dart';
import 'package:flutter/foundation.dart';
/// Background worker based on Dart [Isolate].
class BackgroundWorker {
BackgroundWorker._(this._receivePort, this._sendPort);
final ReceivePort _receivePort;
final SendPort _sendPort;
bool _isDisposed = false;
static Future<BackgroundWorker> create({String? debugName}) async {
final receivePort = ReceivePort();
await Isolate.spawn(
_workerEntry,
receivePort.sendPort,
debugName: debugName,
);
return BackgroundWorker._(receivePort, await receivePort.first as SendPort);
}
static void _workerEntry(SendPort sendPort) {
final receivePort = ReceivePort();
sendPort.send(receivePort.sendPort);
late final StreamSubscription sub;
sub = receivePort.listen((message) {
if (message is _ComputeParams) {
message.execute();
} else {
sub.cancel();
receivePort.close();
return;
}
});
}
Future<R> compute<M, R>(ComputeCallback<M, R> callback, M message) async {
if (_isDisposed) {
throw StateError('Worker is already disposed');
}
final sendPort = ReceivePort();
_sendPort.send(_ComputeParams(sendPort.sendPort, callback, message));
return await sendPort.first as R;
}
/// [compute] wrapper that also provides [Arena] for temporary memory allocation.
Future<R> computeWithArena<M, R>(
R Function(Arena arena, M message) callback,
M message,
) =>
compute(
(message) => using(
(arena) => callback(arena, message),
),
message,
);
void dispose() {
try {
_isDisposed = true;
_sendPort.send(null);
_receivePort.close();
} catch (e) {
debugPrint('Failed to dispose worker (possible double-dispose?): $e');
}
}
}
class _ComputeParams<M, R> {
_ComputeParams(this.sendPort, this.callback, this.message);
final SendPort sendPort;
final ComputeCallback<M, R> callback;
final M message;
void execute() => sendPort.send(callback(message));
}