Tutorials

Build a Dart task processor

Create a runnable task processor with typed commands, grouping, admission policies, rejection reporting, and observable state.

v0.1.0-dev.2Beginnerusecase_forge

Open source reference ↗

Build a Dart task processor

We will build a small worker coordinator with two commands. SubmitTask accepts at most two tasks per queue in one second. ReserveSlot waits for a short debounce window and rejects a conflicting reservation for the same worker queue.

What you will learn

Model intent with typed commands.
Group related executions with executionKey.
Keep rate limiting and debounce in the admission stage.
Publish immutable state through the execution context.
Observe terminal snapshots before closing the UseCase.

Create the project

Terminal

shell

dart create -t console usecase_forge_tasks
cd usecase_forge_tasks
dart pub add usecase_forge

PROJECT STRUCTURE

text

usecase_forge_tasks/
├── bin/
│   └── usecase_forge_tasks.dart
├── lib/
│   └── task_processing.dart
└── pubspec.yaml

Define observable state

Create an immutable value that records completed work.

lib/task_processing.dart

import 'package:usecase_forge/usecase_forge.dart';

final class TaskProcessingState {
  const TaskProcessingState({
    required this.processedCount,
    required this.lastTask,
  });

  const TaskProcessingState.empty() : processedCount = 0, lastTask = '';

  final int processedCount;
  final String lastTask;

  @override
  bool operator ==(Object other) =>
      other is TaskProcessingState &&
      other.processedCount == processedCount &&
      other.lastTask == lastTask;

  @override
  int get hashCode => Object.hash(processedCount, lastTask);
}

Value equality lets the output boundary distinguish a meaningful state change from the same state being published again.

Model two kinds of intent

lib/task_processing.dart

final class SubmitTask extends UseCaseCommand {
  const SubmitTask({required this.queue, required this.taskId});
  final String queue;
  final String taskId;
  @override
  Object get executionKey => queue;
}

final class ReserveSlot extends UseCaseCommand {
  const ReserveSlot({required this.queue, required this.taskId});
  final String queue;
  final String taskId;
  @override
  Object get executionKey => queue;
}

The key groups operations by queue. A busy email queue does not accidentally block a workers queue. Command type remains part of the runtime lane, so equal keys do not merge unrelated registrations.

Register handlers and admission policies

lib/task_processing.dart

final class TaskProcessingUseCase extends UseCase<TaskProcessingState> {
  TaskProcessingUseCase()
      : super(initialState: const TaskProcessingState.empty()) {
    registerCommand<SubmitTask>(
      _process,
      instructions: UseCaseInstructionOverrides(
        input: UseCaseInputInstructionOverrides(
          rateLimit: UseCaseRateLimitInstructions(
            maxExecutions: 2,
            duration: const Duration(seconds: 1),
          ),
        ),
      ),
    );
    registerCommand<ReserveSlot>(
      _reserve,
      instructions: UseCaseInstructionOverrides(
        input: UseCaseInputInstructionOverrides(
          debounce: UseCaseDebounceInstructions(
            duration: const Duration(milliseconds: 30),
          ),
          conflictPolicy: UseCaseInputConflictPolicy.rejectNew,
        ),
      ),
    );
  }

  final rejections = <(String, UseCaseCommandRejectionReason)>[];
}

Rate limit and debounce are admission policies. They decide when intent may advance; they do not change handler code.

Publish state from both handlers

lib/task_processing.dart

Future<void> _process(
  SubmitTask command,
  UseCaseExecutionContext<TaskProcessingState> context,
) async => _publishProcessed(context, command.taskId);

Future<void> _reserve(
  ReserveSlot command,
  UseCaseExecutionContext<TaskProcessingState> context,
) async => _publishProcessed(context, command.taskId);

void _publishProcessed(
  UseCaseExecutionContext<TaskProcessingState> context,
  String taskId,
) {
  final latest = context.snapshot.state;
  context.publish(TaskProcessingState(
    processedCount: latest.processedCount + 1,
    lastTask: taskId,
  ));
}

The handler reads the execution-visible snapshot rather than a captured value from construction time.

Record typed rejection

lib/task_processing.dart

@override
Future<void> onCommandRejected(
  UseCaseExecutionEntry<UseCaseCommand> entry,
  UseCaseCommandRejectionReason reason,
) async {
  final taskId = switch (entry.command) {
    SubmitTask command => command.taskId,
    ReserveSlot command => command.taskId,
    _ => 'unknown',
  };
  rejections.add((taskId, reason));
}

Admission rejection is observable and typed. It is not routed through the handler error path because the handler never started.

Run the scenario

bin/usecase_forge_tasks.dart

Future<void> main() async {
  final tasks = TaskProcessingUseCase();
  final threeFinished = tasks.stream
      .where((snapshot) => snapshot.phase == UseCaseExecutionPhase.finished)
      .take(3)
      .drain<void>();

  tasks.add(const SubmitTask(queue: 'email', taskId: 'email-1'));
  tasks.add(const SubmitTask(queue: 'email', taskId: 'email-2'));
  tasks.add(const SubmitTask(queue: 'email', taskId: 'email-3'));
  tasks.add(const ReserveSlot(queue: 'workers', taskId: 'slot-1'));
  tasks.add(const ReserveSlot(queue: 'workers', taskId: 'slot-2'));

  await threeFinished;
  print('Processed: ${tasks.state.state.processedCount}');
  for (final (taskId, reason) in tasks.rejections) {
    print('Rejected $taskId: ${reason.name}');
  }
  await tasks.close();
}

Complete source

The package contains the complete runnable source in example/task_processing.dart, and its documentation examples are executed by the package test suite. Use that file as the canonical implementation when copying this tutorial into an application.