LCOV - lcov.info worker.dart

LCOV - code coverage report

Current view

top level - src/worker - worker.dart

Test

lcov.info

Date

2026-02-21

Legend

Lines: hit; not hit
Branches: taken; not taken; # not executed

	Hit	Total	Coverage
Lines	72	81	88.9%
Functions	0	0	-
Branches	0	0	-

Each row represents a line of source code
Line	Hits	Source code
1		`import 'dart:async';`
2
3		`import 'package:cancelation_token/cancelation_token.dart';`
4		`import 'package:logger/web.dart';`
5		`import 'package:using/using.dart';`
6
7		`import '../_impl/xplat/_forward_completer.dart';`
8		`import '../_impl/xplat/_forward_stream_controller.dart';`
9		`import '../_impl/xplat/_time_stamp.dart';`
10		`import '../channel.dart';`
11		`import '../exceptions/exception_manager.dart';`
12		`import '../exceptions/squadron_exception.dart';`
13		`import '../exceptions/task_terminated_exception.dart';`
14		`import '../exceptions/worker_exception.dart';`
15		`import '../invoker.dart';`
16		`import '../iworker.dart';`
17		`import '../stats/worker_stat.dart';`
18		`import '../tokens/_squadron_cancelation_token.dart';`
19		`import '../typedefs.dart';`
20		`import '../worker/worker_request.dart';`
21		`import '../worker_service.dart';`
22
23		`part 'worker.stats.dart';`
24
25		`/// Base worker class.`
26		`///`
27		`/// This base class takes care of creating the [Channel] and firing up the`
28		`/// worker. Typically, derived classes should add proxy methods sending`
29		`/// [WorkerRequest]s to the worker.`
30		`abstract class Worker`
31		`with Releasable`
32		`implements WorkerService, IWorker, Invoker {`
33		`/// Creates a [Worker] with the specified entrypoint.`
34	11	`Worker(this._entryPoint,`
35		`{PlatformThreadHook? threadHook, ExceptionManager? exceptionManager})`
36		`: _threadHook = threadHook,`
37		`_exceptionManager = exceptionManager {`
38	22	`_stats = _Stats(this);`
39		`}`
40
41	11	`@override`
42		`void release() {`
43	11	`stop();`
44	11	`super.release();`
45		`}`
46
47		`/// The [Worker]'s entry point; typically, a top-level function in native`
48		`/// world or the Uri to a JavaScript file in browser world.`
49		`final EntryPoint _entryPoint;`
50
51		`@override`
52		`Logger? channelLogger;`
53
54	11	`@override`
55		`ExceptionManager get exceptionManager =>`
56	21	`(_exceptionManager ??= ExceptionManager());`
57		`ExceptionManager? _exceptionManager;`
58
59		`final PlatformThreadHook? _threadHook;`
60
61		`/// The [Worker]'s start arguments.`
62		`List? getStartArgs();`
63
64		`late final _Stats _stats;`
65
66	33	`bool get isStopped => _stats.isStopped;`
67
68		`/// [Worker] statistics.`
69	0	`@Deprecated('Use getStats()')`
70	0	`WorkerStat get stats => _stats.snapshot;`
71
72		`/// [Worker] statistics.`
73	18	`WorkerStat getStats() => _stats.snapshot;`
74
75		`/// Returns true if the [Worker] is connected i.e., it has a valid [Channel].`
76		`/// Returns false otherwise.`
77	2	`bool get isConnected => _channel != null;`
78
79		`/// Shared [Channel] that can be used to communicate with the worker.`
80	6	`Channel? getSharedChannel() => _channel?.share();`
81
82		`Channel? _channel;`
83		`Future<Channel>? _openChannel;`
84
85		`/// Sends a workload to the worker.`
86	10	`@override`
87		`Future<dynamic> send(`
88		`int command, {`
89		`List args = const [],`
90		`CancelationToken? token,`
91		`bool inspectRequest = false,`
92		`bool inspectResponse = false,`
93		`}) async {`
94	0	`token?.throwIfCanceled();`
95
96		`// get the channel, start the worker if necessary`
97	18	`final channel = _channel ?? await start();`
98
99	10	`final completer = ForwardCompleter();`
100
101	0	`final squadronToken = token?.wrap();`
102	0	`squadronToken?.onCanceled.then((ex) {`
103	0	`_channel?.cancelToken(squadronToken);`
104	0	`completer.failure(SquadronException.from(ex, null, command));`
105		`});`
106
107	20	`_stats.beginWork();`
108	50	`completer.future.whenComplete(_stats.endWork).ignore();`
109
110		`try {`
111	10	`final res = await channel.sendRequest(`
112		`command,`
113		`args,`
114		`token: squadronToken,`
115		`inspectRequest: inspectRequest,`
116		`inspectResponse: inspectResponse,`
117		`);`
118	10	`completer.success(res);`
119		`} catch (ex, st) {`
120	6	`_stats.failed();`
121	6	`completer.failure(SquadronException.from(ex, st, command));`
122		`}`
123
124	10	`return completer.future;`
125		`}`
126
127		`/// Sends a streaming workload to the worker.`
128	5	`@override`
129		`Stream<dynamic> stream(`
130		`int command, {`
131		`List args = const [],`
132		`CancelationToken? token,`
133		`bool inspectRequest = false,`
134		`bool inspectResponse = false,`
135		`}) {`
136	2	`final squadronToken = token?.wrap();`
137
138		`late final ForwardStreamController controller;`
139
140	6	`squadronToken?.onCanceled.then((ex) {`
141	2	`if (!controller.isClosed) {`
142	4	`controller.subscription?.cancel();`
143	4	`controller.safeAddError(SquadronException.from(ex, null, command));`
144	2	`controller.close();`
145		`}`
146	4	`_channel?.cancelToken(squadronToken);`
147		`});`
148
149	10	`controller = ForwardStreamController(onListen: () async {`
150		`try {`
151	5	`if (controller.isClosed) return;`
152	2	`squadronToken?.throwIfCanceled();`
153	8	`final channel = _channel ?? await start();`
154	5	`if (controller.isClosed) return;`
155
156	10	`_stats.beginWork();`
157	25	`controller.done.whenComplete(_stats.endWork).ignore();`
158
159	5	`controller.attachSubscription(channel`
160	5	`.sendStreamingRequest(`
161		`command,`
162		`args,`
163		`token: squadronToken,`
164		`inspectRequest: inspectRequest,`
165		`inspectResponse: inspectResponse,`
166		`)`
167	5	`.listen(`
168	5	`controller.safeAdd,`
169	3	`onError: (ex, st) => controller`
170	6	`.safeAddError(SquadronException.from(ex, st, command)),`
171	5	`onDone: controller.close,`
172		`cancelOnError: false,`
173		`));`
174		`} catch (ex, st) {`
175	2	`_stats.failed();`
176	1	`controller.subscription?.cancel();`
177	2	`controller.safeAddError(SquadronException.from(ex, st, command));`
178	1	`controller.close();`
179		`}`
180		`});`
181
182	5	`return controller.stream;`
183		`}`
184
185		`/// Creates a [Channel] and starts the worker using the [_entryPoint].`
186	11	`@override`
187		`Future<Channel> start() {`
188	11	`if (isStopped) {`
189	1	`throw WorkerException('Invalid state: worker is stopped');`
190		`}`
191
192	11	`if (_channel != null) {`
193	0	`return Future.value(_channel!);`
194		`}`
195
196	22	`return _openChannel ??= Channel.open(`
197	11	`exceptionManager,`
198	11	`channelLogger,`
199	11	`_entryPoint,`
200	11	`getStartArgs() ?? const [],`
201	11	`_threadHook,`
202	21	`).then((channel) {`
203	10	`_channel = channel;`
204	20	`_stats.start();`
205		`return channel;`
206		`});`
207		`}`
208
209		`/// Stops this worker.`
210	11	`@override`
211		`void stop() {`
212	11	`if (!isStopped) {`
213	14	`channelLogger?.d('Stop worker');`
214	22	`_stats.stop();`
215	11	`_openChannel = null;`
216	21	`_channel?.close();`
217	11	`_channel = null;`
218		`}`
219		`}`
220
221		`/// Terminates this worker.`
222	2	`@override`
223		`void terminate([TaskTerminatedException? ex]) {`
224		`// terminate channel and stop worker`
225	2	`ex ??= TaskTerminatedException('Worker has been terminated');`
226	4	`_channel?.terminate(ex);`
227	2	`stop();`
228		`}`
229
230		`/// Workers do not need an [operations] map.`
231	0	`@override`
232		`OperationsMap get operations => WorkerService.noOperations;`
233		`}`

Each row represents a line of source code

Line

Branch

Hits

Source code

import 'dart:async';

import 'package:cancelation_token/cancelation_token.dart';

import 'package:logger/web.dart';

import 'package:using/using.dart';

import '../_impl/xplat/_forward_completer.dart';

import '../_impl/xplat/_forward_stream_controller.dart';

import '../_impl/xplat/_time_stamp.dart';

import '../channel.dart';

import '../exceptions/exception_manager.dart';

import '../exceptions/squadron_exception.dart';

import '../exceptions/task_terminated_exception.dart';

import '../exceptions/worker_exception.dart';

import '../invoker.dart';

import '../iworker.dart';

import '../stats/worker_stat.dart';

import '../tokens/_squadron_cancelation_token.dart';

import '../typedefs.dart';

import '../worker/worker_request.dart';

import '../worker_service.dart';

part 'worker.stats.dart';

/// Base worker class.

///

/// This base class takes care of creating the [Channel] and firing up the

/// worker. Typically, derived classes should add proxy methods sending

/// [WorkerRequest]s to the worker.

abstract class Worker

with Releasable

implements WorkerService, IWorker, Invoker {

/// Creates a [Worker] with the specified entrypoint.

Worker(this._entryPoint,

{PlatformThreadHook? threadHook, ExceptionManager? exceptionManager})

: _threadHook = threadHook,

_exceptionManager = exceptionManager {

_stats = _Stats(this);

}

@override

void release() {

stop();

super.release();

}

/// The [Worker]'s entry point; typically, a top-level function in native

/// world or the Uri to a JavaScript file in browser world.

final EntryPoint _entryPoint;

@override

Logger? channelLogger;

@override

ExceptionManager get exceptionManager =>

(_exceptionManager ??= ExceptionManager());

ExceptionManager? _exceptionManager;

final PlatformThreadHook? _threadHook;

/// The [Worker]'s start arguments.

List? getStartArgs();

late final _Stats _stats;

bool get isStopped => _stats.isStopped;

/// [Worker] statistics.

@Deprecated('Use getStats()')

WorkerStat get stats => _stats.snapshot;

/// [Worker] statistics.

WorkerStat getStats() => _stats.snapshot;

/// Returns true if the [Worker] is connected i.e., it has a valid [Channel].

/// Returns false otherwise.

bool get isConnected => _channel != null;

/// Shared [Channel] that can be used to communicate with the worker.

Channel? getSharedChannel() => _channel?.share();

Channel? _channel;

Future<Channel>? _openChannel;

/// Sends a workload to the worker.

@override

Future<dynamic> send(

int command, {

List args = const [],

CancelationToken? token,

bool inspectRequest = false,

bool inspectResponse = false,

}) async {

token?.throwIfCanceled();

// get the channel, start the worker if necessary

final channel = _channel ?? await start();

final completer = ForwardCompleter();

100

101

final squadronToken = token?.wrap();

102

squadronToken?.onCanceled.then((ex) {

103

_channel?.cancelToken(squadronToken);

104

completer.failure(SquadronException.from(ex, null, command));

105

});

106

107

_stats.beginWork();

108

completer.future.whenComplete(_stats.endWork).ignore();

109

110

try {

111

final res = await channel.sendRequest(

112

command,

113

args,

114

token: squadronToken,

115

inspectRequest: inspectRequest,

116

inspectResponse: inspectResponse,

117

);

118

completer.success(res);

119

} catch (ex, st) {

120

_stats.failed();

121

completer.failure(SquadronException.from(ex, st, command));

122

}

123

124

return completer.future;

125

}

126

127

/// Sends a streaming workload to the worker.

128

@override

129

Stream<dynamic> stream(

130

int command, {

131

List args = const [],

132

CancelationToken? token,

133

bool inspectRequest = false,

134

bool inspectResponse = false,

135

}) {

136

final squadronToken = token?.wrap();

137

138

late final ForwardStreamController controller;

139

140

squadronToken?.onCanceled.then((ex) {

141

if (!controller.isClosed) {

142

controller.subscription?.cancel();

143

controller.safeAddError(SquadronException.from(ex, null, command));

144

controller.close();

145

}

146

_channel?.cancelToken(squadronToken);

147

});

148

149

controller = ForwardStreamController(onListen: () async {

150

try {

151

if (controller.isClosed) return;

152

squadronToken?.throwIfCanceled();

153

final channel = _channel ?? await start();

154

if (controller.isClosed) return;

155

156

_stats.beginWork();

157

controller.done.whenComplete(_stats.endWork).ignore();

158

159

controller.attachSubscription(channel

160

.sendStreamingRequest(

161

command,

162

args,

163

token: squadronToken,

164

inspectRequest: inspectRequest,

165

inspectResponse: inspectResponse,

166

)

167

.listen(

168

controller.safeAdd,

169

onError: (ex, st) => controller

170

.safeAddError(SquadronException.from(ex, st, command)),

171

onDone: controller.close,

172

cancelOnError: false,

173

));

174

} catch (ex, st) {

175

_stats.failed();

176

controller.subscription?.cancel();

177

controller.safeAddError(SquadronException.from(ex, st, command));

178

controller.close();

179

}

180

});

181

182

return controller.stream;

183

}

184

185

/// Creates a [Channel] and starts the worker using the [_entryPoint].

186

@override

187

Future<Channel> start() {

188

if (isStopped) {

189

throw WorkerException('Invalid state: worker is stopped');

190

}

191

192

if (_channel != null) {

193

return Future.value(_channel!);

194

}

195

196

return _openChannel ??= Channel.open(

197

exceptionManager,

198

channelLogger,

199

_entryPoint,

200

getStartArgs() ?? const [],

201

_threadHook,

202

).then((channel) {

203

_channel = channel;

204

_stats.start();

205

return channel;

206

});

207

}

208

209

/// Stops this worker.

210

@override

211

void stop() {

212

if (!isStopped) {

213

channelLogger?.d('Stop worker');

214

_stats.stop();

215

_openChannel = null;

216

_channel?.close();

217

_channel = null;

218

}

219

}

220

221

/// Terminates this worker.

222

@override

223

void terminate([TaskTerminatedException? ex]) {

224

// terminate channel and stop worker

225

ex ??= TaskTerminatedException('Worker has been terminated');

226

_channel?.terminate(ex);

227

stop();

228

}

229

230

/// Workers do not need an [operations] map.

231

@override

232

OperationsMap get operations => WorkerService.noOperations;

233

}

Choose FeaturesFunction coverageBranch coverageLegend