Skip to main content

Programmatic Workers & Enqueuers

Use Stem's Dart APIs to embed task production and processing inside your application services. This guide focuses on the two core roles: producer (enqueuer) and worker.

This page is intentionally about the lower-level embedding surface. If you want the default happy path, prefer StemClient, StemApp, or StemWorkflowApp and come here only when you need direct runtime composition.

Producer (Enqueuer)​

lib/producer.dart
Future<void> minimalProducer() async {
  final app = await StemApp.inMemory(
    tasks: [
      FunctionTaskHandler<void>(
        name: 'email.send',
        entrypoint: (context, args) async {
          final to = args['to'] as String? ?? 'friend';
          print('Queued email to $to');
          return null;
        },
      ),
    ],
  );

  final taskId = await app.enqueue(
    'email.send',
    args: {'to': 'hello@example.com', 'subject': 'Welcome'},
  );

  print('Enqueued $taskId');
  await app.waitForTask<void>(taskId);
  await app.close();
}

Tips​

  • Always reuse the producer runtime (StemClient, StemApp, or raw Stem) rather than constructing one per request.
  • Use TaskOptions to set queue, retries, timeouts, and isolation.
  • Add custom metadata via the meta argument for observability or downstream processing.

Worker​

bin/worker.dart
class EmailTask extends TaskHandler<void> {
  @override
  String get name => 'email.send';

  @override
  TaskOptions get options => const TaskOptions(maxRetries: 2);

  @override
  Future<void> call(TaskContext context, Map<String, Object?> args) async {
    final to = args['to'] as String;
    print('Sending to $to (attempt ${context.attempt})');
  }
}

Future<void> minimalWorker() async {
  final broker = InMemoryBroker();
  final backend = InMemoryResultBackend();

  final worker = Worker(
    broker: broker,
    backend: backend,
    tasks: [EmailTask()],
    queue: 'default',
  );

  await worker.start();
}

Lifecycle Tips​

  • Call worker.shutdown() on SIGINT/SIGTERM to drain in-flight tasks and emit workerStopping/workerShutdown signals.
  • Monitor heartbeats via StemSignals.workerHeartbeat or the heartbeat backend for liveness checks.
  • Use WorkerLifecycleConfig to install signal handlers, configure soft/hard shutdown timeouts, and recycle isolates after N tasks or memory thresholds.

Putting It Together​

A lightweight service wires the producer and worker into your application startup:

lib/bootstrap.dart
class StemRuntime {
  StemRuntime({required this.tasks, required this.brokerUrl});

  final List<TaskHandler<Object?>> tasks;
  final String brokerUrl;

  final InMemoryBroker _stemBroker = InMemoryBroker();
  final InMemoryResultBackend _stemBackend = InMemoryResultBackend();
  final InMemoryBroker _workerBroker = InMemoryBroker();
  final InMemoryResultBackend _workerBackend = InMemoryResultBackend();

  late final Stem stem = Stem(
    broker: _stemBroker,
    backend: _stemBackend,
    tasks: tasks,
  );

  late final Worker worker = Worker(
    broker: _workerBroker,
    backend: _workerBackend,
    tasks: tasks,
  );

  Future<void> start() async {
    await worker.start();
  }

  Future<void> stop() async {
    await worker.shutdown();
    await _workerBackend.close();
    await _workerBroker.close();
    await _stemBackend.close();
    await _stemBroker.close();
  }
}

Swap the in-memory adapters for Redis/Postgres when you deploy, keeping the API surface the same.

If your service wants a higher-level bootstrap that owns broker/backend/tasks in one place, use StemApp or StemClient instead of wiring raw Stem and Worker instances by hand. This page stays focused on the lower-level embedding path.

Checklist​

  • Reuse producer and worker objects—avoid per-request construction.
  • Keep a shared tasks list/module so producers and workers stay in sync.
  • Reach for a custom TaskRegistry only when you need advanced dynamic registration behavior.
  • Capture task IDs returned by Stem.enqueue when you need to poll results or correlate with your own auditing.
  • Emit lifecycle signals (StemSignals) and wire logs/metrics early so production instrumentation is already in place.
  • For HTTP/GraphQL handlers, wrap enqueues in try/catch to surface validation errors before tasks hit the queue.

Next, continue with the Worker Control CLI or explore Signals for advanced instrumentation.