Alveus.JobRunner

A lightweight, AOT-compatible job runner for executing tasks sequentially with support for dynamic job enqueueing, custom contexts, and comprehensive execution callbacks.

Features

• Sequential Execution: Jobs are executed in FIFO order
• Dynamic Job Enqueueing: Jobs can enqueue additional jobs during execution
• Custom Context: Pass custom data to all jobs in the sequence
• Exception Handling: Configure whether to halt or continue on exceptions
• Execution Callbacks: Hook into job lifecycle events
• AOT Compatible: Fully supports Native AOT compilation
• Flexible Job Definition: Define jobs via classes or inline delegates

Installation

dotnet add package Alveus.JobRunner

Quick Start

Using DelegateJob

using Alveus.JobRunner;

var runner = new JobRunner<MyContext>();

// Add jobs using delegates
runner.Enqueue(new DelegateJob<MyContext>(
    async (queue, context, ct) =>
    {
        Console.WriteLine("Job 1 executing");

        // Enqueue additional jobs dynamically
        queue.Enqueue(new DelegateJob<MyContext>(
            async (q, ctx, c) => Console.WriteLine("Job 3 executing"),
            "Dynamic Job"
        ));
    },
    "Job 1"
));

runner.Enqueue(new DelegateJob<MyContext>(
    async (queue, context, ct) => Console.WriteLine("Job 2 executing"),
    "Job 2"
));

// Execute all jobs
var context = new MyContext();
await runner.ExecuteAsync(context);

Using Custom Job Classes

using Alveus.JobRunner;

[JobName("Send Email")]
public class SendEmailJob : IJob<EmailContext>
{
    public async Task ExecuteAsync(IJobQueue<EmailContext> queue, EmailContext context, CancellationToken cancellationToken)
    {
        await SendEmailAsync(context.Recipient, context.Message);

        // Optionally enqueue follow-up jobs
        queue.Enqueue(new LogEmailSentJob());
    }
}

var runner = new JobRunner<EmailContext>();
runner.Enqueue(new SendEmailJob());
await runner.ExecuteAsync(new EmailContext
{
    Recipient = "user@example.com",
    Message = "Hello!"
});

Exception Handling

Halt on Exception (Default)

var runner = new JobRunner<string>();
runner.Enqueue(new DelegateJob<string>(async (q, ctx, ct) => throw new Exception()));

// Throws exception and stops execution
await runner.ExecuteAsync("context", haltOnException: true);

Continue on Exception

var runner = new JobRunner<string>();
runner.Enqueue(new DelegateJob<string>(async (q, ctx, ct) => throw new Exception()));
runner.Enqueue(new DelegateJob<string>(async (q, ctx, ct) => Console.WriteLine("Still executing")));

// Continues executing remaining jobs after exception
await runner.ExecuteAsync("context", haltOnException: false);

Execution Callbacks

OnJobExecuting

var runner = new JobRunner<string>();

runner.OnJobExecuting += context =>
{
    Console.WriteLine($"Executing job {context.JobIndex + 1}/{context.TotalJobs}");
    if (context.JobName != null)
    {
        Console.WriteLine($"Job name: {context.JobName}");
    }
};

runner.Enqueue(new DelegateJob<string>(async (q, ctx, ct) => { }, "My Job"));
await runner.ExecuteAsync("context");

OnJobException

var runner = new JobRunner<string>();

runner.OnJobException += (context, exception) =>
{
    Console.WriteLine($"Job {context.JobIndex} failed: {exception.Message}");
    // Log to monitoring service, send alerts, etc.
};

runner.Enqueue(new DelegateJob<string>(async (q, ctx, ct) => throw new Exception("Oops")));
await runner.ExecuteAsync("context", haltOnException: false);

Job Naming

Using JobNameAttribute

[JobName("Process Payment")]
public class ProcessPaymentJob : IJob<PaymentContext>
{
    public async Task ExecuteAsync(IJobQueue<PaymentContext> queue, PaymentContext context, CancellationToken cancellationToken)
    {
        // Process payment
    }
}

Using DelegateJob Constructor

runner.Enqueue(new DelegateJob<string>(
    async (q, ctx, ct) => { /* work */ },
    "My Named Job"
));

Advanced Usage

Dynamic Job Chaining

var runner = new JobRunner<ProcessContext>();

runner.Enqueue(new DelegateJob<ProcessContext>(
    async (queue, context, ct) =>
    {
        var items = await FetchItemsAsync();

        // Dynamically enqueue a job for each item
        foreach (var item in items)
        {
            queue.Enqueue(new ProcessItemJob(item));
        }
    },
    "Fetch and Queue Items"
));

await runner.ExecuteAsync(new ProcessContext());

Progress Tracking

var runner = new JobRunner<string>();
var progress = new Progress();

runner.OnJobExecuting += context =>
{
    progress.Report(context.JobIndex, context.TotalJobs);
};

// Add jobs...
await runner.ExecuteAsync("context");

Cancellation Support

var runner = new JobRunner<string>();
var cts = new CancellationTokenSource();

// Add jobs...

// Cancel after 5 seconds
cts.CancelAfter(TimeSpan.FromSeconds(5));

try
{
    await runner.ExecuteAsync("context", cancellationToken: cts.Token);
}
catch (OperationCanceledException)
{
    Console.WriteLine("Execution was cancelled");
}

API Reference

IJobRunner<TContext>

• void Enqueue(IJob<TContext> job) - Enqueues a job for execution
• Task ExecuteAsync(TContext context, bool haltOnException = true, CancellationToken cancellationToken = default) - Executes all enqueued jobs
• event JobExecutingCallback<TContext>? OnJobExecuting - Fires before each job executes
• event JobExceptionCallback<TContext>? OnJobException - Fires when a job throws an exception

IJob<TContext>

• Task ExecuteAsync(IJobQueue<TContext> queue, TContext context, CancellationToken cancellationToken) - Executes the job

JobExecutionContext<TContext>

• IJob<TContext> Job - The job being executed
• string? JobName - The name of the job (from attribute or constructor)
• int JobIndex - Zero-based index of the job
• int TotalJobs - Total number of jobs in the queue

License

Refer to LICENSE document for license information.

Contributing

Copyright (c) Alveus Dev (https://alveus.dev). All rights reserved.