airo_job_scheduler

pub package CI License: MIT

High-performance cooperative CPU resource scheduler, reactive task service, DAG workflow executor, and isolate pool manager for Flutter applications.

⚡ Key Features

  • Cooperative Resource Scheduling: Manage CPU budgets, battery-saver constraints, and thermal pressure.
  • Reactive Job Scheduler (AiroJobSchedulerService): Stream-based lifecycle events (AiroJobScheduled, AiroJobCompleted, AiroJobFailed, AiroJobRetried).
  • Observable Metrics (AiroJobMetrics): Real-time tracking of active jobs, completed counts, failure rates, and execution time averages.
  • Exponential Backoff & Full Jitter (AiroRetryPolicy): Built-in automatic retry handling for transient network and I/O failures.
  • DAG Job Workflows (AiroJobWorkflow & AiroJobWorkflowExecutor): Define and run job dependency trees with parallel execution of ready tasks.
  • Process Persistence (AiroJobPersistence): Store and recover pending jobs across application process restarts using memory or JSON adapters.

📦 Installation

Add airo_job_scheduler to your pubspec.yaml:

dependencies:
  airo_job_scheduler: ^1.1.0

🚀 Quick Start

1. Basic Reactive Job Scheduling

import 'package:airo_job_scheduler/airo_job_scheduler.dart';

void main() async {
  final scheduler = AiroJobSchedulerService();

  // Listen to live scheduler events
  scheduler.events.listen((event) {
    if (event is AiroJobCompleted) {
      print('Job ${event.jobId} completed in ${event.duration.inMilliseconds}ms');
    }
  });

  // Schedule a computation
  final result = await scheduler.scheduleJob<int>(
    jobId: 'parse-payload-1',
    kind: AiroWorkerJobKind.playlistImport,
    computation: () => 42,
    retryPolicy: const AiroRetryPolicy(maxRetries: 3),
  );

  print('Result: $result');
}

2. Observable Metrics Dashboard Integration

final scheduler = AiroJobSchedulerService();

// AiroJobMetrics extends ChangeNotifier for Flutter UI binding
ListenableBuilder(
  listenable: scheduler.metrics,
  builder: (context, child) {
    return Text('Active Jobs: ${scheduler.metrics.currentActiveJobs}');
  },
);

3. Executing Multi-Step DAG Workflows

final scheduler = AiroJobSchedulerService();

final now = DateTime.now();
final expires = now.add(const Duration(minutes: 15));

final jobA = AiroWorkerJobDescriptor(
  jobId: AiroWorkerStableValue.stable('fetch-api'),
  kind: AiroWorkerJobKind.deviceSync,
  createdAt: now,
  expiresAt: expires,
);

final jobB = AiroWorkerJobDescriptor(
  jobId: AiroWorkerStableValue.stable('process-json'),
  kind: AiroWorkerJobKind.playlistImport,
  createdAt: now,
  expiresAt: expires,
);

final workflow = AiroJobWorkflow(
  workflowId: 'sync-pipeline',
  jobs: [jobA, jobB],
);

// Job B runs only after Job A completes
workflow.addDependency('fetch-api', 'process-json');

final executor = AiroJobWorkflowExecutor(
  scheduler: scheduler,
  workflow: workflow,
);

final results = await executor.executeWorkflow(
  jobCallbacks: {
    'fetch-api': () => 'raw_data',
    'process-json': () => {'status': 'processed'},
  },
);

📄 License

MIT License - see LICENSE for details.

Libraries

airo_job_scheduler