Milvasoft.Milvaion.Sdk 10.0.2

Milvaion Worker SDK

Milvaion Worker SDK is a .NET library that enables your applications to execute scheduled jobs dispatched by the Milvaion Scheduler API. It provides automatic job discovery, RabbitMQ integration, offline resilience, health monitoring, and comprehensive retry mechanisms.

Features

• Automatic Job Discovery - Scans your assembly for IJob implementations and registers them automatically
• RabbitMQ Integration - Consumes jobs from RabbitMQ queues with configurable parallelism
• Redis Support - Uses Redis for job cancellation signals and worker heartbeats
• Offline Resilience - Local outbox pattern for storing logs/status updates when disconnected
• Configurable Retries - Per-job retry policies with exponential backoff
• Health Monitoring - Automatic worker heartbeats and zombie detection
• Parallel Execution - Control concurrency per worker and per job type
• Graceful Shutdown - Proper cleanup and in-flight job handling on termination

Installation

Install the NuGet package:

dotnet add package Milvasoft.Milvaion.Sdk.Worker

Or via Package Manager Console:

Install-Package Milvasoft.Milvaion.Sdk.Worker

Quick Start

1. Create a Job Class

Implement the IJob interface:

using Milvaion.Sdk.Worker.Abstractions;

public class SendEmailJob : IJob
{
    public async Task<JobExecutionResult> ExecuteAsync(IJobContext context)
    {
        var email = context.GetData<string>("email");
        
        context.LogInformation($"Sending email to {email}");
        
        // Your business logic here
        await Task.Delay(1000);
        
        return JobExecutionResult.Success("Email sent successfully");
    }
}

2. Configure appsettings.json

{
  "Worker": {
    "WorkerId": "email-worker",
    "MaxParallelJobs": 5,
    "RabbitMQ": {
      "Host": "localhost",
      "Port": 5672,
      "Username": "guest",
      "Password": "guest",
      "VirtualHost": "/"
    },
    "Redis": {
      "ConnectionString": "localhost:6379",
      "Database": 0
    },
    "Heartbeat": {
      "Enabled": true,
      "IntervalSeconds": 30
    },
    "OfflineResilience": {
      "Enabled": true,
      "LocalStoragePath": "./worker_data"
    }
  },
  "JobConsumers": {
    "SendEmailJob": {
      "ConsumerId": "sendemail-consumer",
      "MaxParallelJobs": 3,
      "ExecutionTimeoutSeconds": 300,
      "MaxRetries": 3,
      "BaseRetryDelaySeconds": 60,
      "RoutingPatterns": ["sendemail.*"]
    }
  }
}

3. Register Worker Services in Program.cs

using Microsoft.Extensions.Hosting;
using Milvaion.Sdk.Worker;

var builder = Host.CreateApplicationBuilder(args);

// Auto-discover jobs and register worker services
builder.Services.AddMilvaionWorkerWithJobs(builder.Configuration);

var host = builder.Build();
await host.RunAsync();

Configuration Reference

Worker Options

Job Consumer Options

Each job can have dedicated configuration under JobConsumers:

Job Lifecycle

1. Discovery - SDK scans assembly for IJob implementations on startup
2. Validation - Ensures each job has corresponding configuration
3. Consumer Registration - Creates dedicated RabbitMQ consumer per job type
4. Execution - Receives message, deserializes context, executes job logic
5. Status Updates - Publishes Running, Completed, or Failed status to API
6. Retry Logic - On failure, republishes message with incremented retry count
7. DLQ Handling - Moves to Dead Letter Queue after max retries exhausted

Offline Resilience

When enabled, the SDK stores logs and status updates locally when the central API is unreachable:

{
  "OfflineResilience": {
    "Enabled": true,
    "LocalStoragePath": "./worker_data",
    "SyncIntervalSeconds": 60
  }
}

• Outbox Pattern: Persists updates to SQLite database
• Auto-Sync: Periodically retries sending stored entries
• At-Least-Once Guarantee: Ensures no data loss during network failures

Job Cancellation

Jobs can be cancelled via Redis pub/sub:

public class LongRunningJob : IJob
{
    public async Task<JobExecutionResult> ExecuteAsync(IJobContext context)
    {
        for (int i = 0; i < 100; i++)
        {
            // Check cancellation token
            context.CancellationToken.ThrowIfCancellationRequested();
            
            await Task.Delay(1000, context.CancellationToken);
        }
        
        return JobExecutionResult.Success();
    }
}

Best Practices

• Set Job-Specific Timeouts - Different jobs have different execution profiles
• Use Structured Logging - Leverage context.LogInformation() for traceable logs
• Handle Transient Failures - Design idempotent jobs for safe retries
• Monitor Worker Health - Use heartbeat and health endpoints
• Scale Horizontally - Deploy multiple worker instances with same WorkerId
• Enable Offline Resilience - For unreliable network environments

Examples

Complete Worker Project

See the SampleWorker project for a reference implementation.

Job with Retries

public class UnreliableApiJob : IJob
{
    private readonly HttpClient _httpClient;
    
    public UnreliableApiJob(HttpClient httpClient)
    {
        _httpClient = httpClient;
    }
    
    public async Task<JobExecutionResult> ExecuteAsync(IJobContext context)
    {
        try
        {
            var response = await _httpClient.GetAsync(context.GetData<string>("url"));
            response.EnsureSuccessStatusCode();
            
            return JobExecutionResult.Success();
        }
        catch (HttpRequestException ex)
        {
            // Will automatically retry based on MaxRetries config
            return JobExecutionResult.Failure($"API call failed: {ex.Message}");
        }
    }
}

Configuration:

{
  "JobConsumers": {
    "UnreliableApiJob": {
      "ConsumerId": "api-consumer",
      "MaxRetries": 5,
      "BaseRetryDelaySeconds": 30
    }
  }
}

Documentation

• Full Documentation
• Quick Start Guide
• Configuration Reference
• API Reference

Support

• Report Issues
• Discussions
• Email: support@milvasoft.com

License

Licensed under the MIT License.

Built with love by Milvasoft