run method

@override Future<WorkerResponse> run(Worker worker)

Run the task using the Worker context.

Implementation

@override
Future<WorkerResponse> run(Worker worker) async {
  Stream<List<Conversation>> pipeline = Stream.value(batch);
  for (StepDescriptor step in steps) {
    pipeline = pipeline.transform(worker.registry.transformers[step.type]!
        .call(step.config, worker.progress, cache));
  }

  List<Conversation> transformed = [];
  await for (List<Conversation> conversation in pipeline) {
    transformed.addAll(conversation);
  }
  return TransformResponse(transformed);
}