Processors

A processor is the fundamental unit of work in distenka. You extend the abstract Processor<TConfig, TItem> class, implement two methods, and distenka handles parallelism, retries, logging, and cluster coordination for you.

User story: Automating image data extraction

Imagine you're working at a company that handles large volumes of images — photos or scanned documents uploaded by users. These images often contain text, such as receipts, notes, or labels, that must be extracted automatically. This is where Optical Character Recognition (OCR) comes in.

In this guide we'll build a process using distenka that takes image URLs, extracts the text with OCR, sends it to a webhook, and saves it in MongoDB. The result is scalable (parallel image processing), resilient (automatic retries on failure), and easy to run from the command line.

Configuration: defaults and custom inputs

distenka makes your process configurable through a JSON file or CLI arguments. There are two types of config: distenka's built-in execution settings (parallelism, retries) and your own custom inputs specific to the business process.

Execution settings

These control how the process runs. Sensible defaults are provided so you can start immediately:

parallelTaskCount: 1 — Runs one item at a time. Increase to process multiple images simultaneously.
itemFailureRetryCount: 1 — Retry a failed item once before marking it as failed.
itemFailureCountToStopProcess: 1 — Stop the entire process after this many item failures.
exceptionRetryCount: 0 — No retries for unhandled exceptions by default.
unsuccessfulRetryCount: 0 — No retries for non-exception failures (e.g. Result.Failure).
handleExceptions: true — Automatically catches and logs exceptions; prevents crashes.
launchDebugger: false — Set to true to attach a debugger on startup.

Example execution override in your config file:

appsettings-style execution block (json)

{
  "execution": {
    "parallelTaskCount": 4,
    "itemFailureRetryCount": 3,
    "itemFailureCountToStopProcess": 10
  }
}

Custom config inputs

Extend Config to add the properties your process needs. Defaults are provided so tests and quick runs work out of the box:

ImageUrls — List of image URLs to process.
WebhookUrl — Endpoint to POST extracted text to.
MongoDbConnectionString — MongoDB connection string.
DatabaseName — Target database name.
CollectionName — Target collection name.
TesseractDataPath — Path to Tesseract language data folder.

ImageOCRConfig.cs

using Distenka;

namespace ImageOCRApp
{
    public class ImageOCRConfig : Config
    {
        public List<string> ImageUrls { get; set; }
        public string WebhookUrl { get; set; }
        public string MongoDbConnectionString { get; set; }
        public string DatabaseName { get; set; }
        public string CollectionName { get; set; }
        public string TesseractDataPath { get; set; }
    }
}

Process setup: from constructor to execution

Let's break down the ImageOCRProcess class. Inherit from distenka's base processor, wire up shared resources in the constructor, then implement GetItemsAsync and ProcessAsync.

Constructor and abstract class

The process extends Processor<TConfig, TItem> where TConfig is your settings class and TItem is the type you're processing (here, string for image URLs). The constructor initialises shared resources — keep it lightweight:

ImageOCRProcess.cs — class declaration & constructor

public class ImageOCRProcess : Processor<ImageOCRConfig, string>
{
    private static readonly ILogger<ImageOCRProcess> Logger =
        LoggerFactory.Create(b => b.AddConsole())
                     .CreateLogger<ImageOCRProcess>();

    private readonly HttpClient httpClient;
    private readonly IMongoCollection<BsonDocument> collection;

    public ImageOCRProcess(ImageOCRConfig config) : base(config)
    {
        this.httpClient = new HttpClient();

        var mongoClient = new MongoClient(config.MongoDbConnectionString);
        this.collection = mongoClient
            .GetDatabase(config.DatabaseName)
            .GetCollection<BsonDocument>(config.CollectionName);
    }
}

Thread safety

Mark shared resources as readonly. When parallelTaskCount > 1, ProcessAsync runs concurrently on multiple threads — avoid mutable shared state.

GetItemsAsync — what to process

Override GetItemsAsync to return the stream of items distenka will iterate over. Using IAsyncEnumerable enables efficient, lazy streaming of large datasets:

ImageOCRProcess.cs — GetItemsAsync

public override IAsyncEnumerable<string> GetItemsAsync()
{
    return Config.ImageUrls.ToAsyncEnumerable();
}

In production, you could replace this with a database query, message queue drain, or any other async source — distenka doesn't care where the items come from.

ProcessAsync — the per-item work

This is the heart of your process. It's called once per item, potentially in parallel. Return Result.Success() or Result.Failure(ex):

ImageOCRProcess.cs — ProcessAsync (skeleton)

public override async Task<Result> ProcessAsync(string imageUrl)
{
    try
    {
        // 1. Download the image
        // 2. Perform OCR
        // 3. POST to webhook
        // 4. Save to MongoDB
        return Result.Success();
    }
    catch (Exception ex)
    {
        Logger.LogError(ex, "Failed to process image: {ImageUrl}", imageUrl);
        return Result.Failure(ex);
    }
}

Full implementation

Here is the complete, production-ready ImageOCRProcess class. It downloads each image, runs Tesseract OCR, POSTs the result to your webhook, and persists a record in MongoDB:

ImageOCRProcess.cs — full class

using System.Text;
using Distenka;
using Microsoft.Extensions.Logging;
using MongoDB.Bson;
using MongoDB.Driver;
using Newtonsoft.Json;
using Tesseract;

namespace ImageOCRApp
{
    public class ImageOCRProcess : Processor<ImageOCRConfig, string>
    {
        private static readonly ILogger<ImageOCRProcess> Logger =
            LoggerFactory.Create(b => b.AddConsole())
                         .CreateLogger<ImageOCRProcess>();

        private readonly HttpClient httpClient;
        private readonly IMongoCollection<BsonDocument> collection;

        public ImageOCRProcess(ImageOCRConfig config) : base(config)
        {
            this.httpClient = new HttpClient();
            var mongoClient = new MongoClient(config.MongoDbConnectionString);
            this.collection = mongoClient
                .GetDatabase(config.DatabaseName)
                .GetCollection<BsonDocument>(config.CollectionName);
        }

        public override IAsyncEnumerable<string> GetItemsAsync()
            => Config.ImageUrls.ToAsyncEnumerable();

        public override async Task<Result> ProcessAsync(string imageUrl)
        {
            try
            {
                Logger.LogInformation("Processing image: {ImageUrl}", imageUrl);

                // 1. Download the image
                var imageBytes = await httpClient.GetByteArrayAsync(imageUrl);

                // 2. Perform OCR via Tesseract
                using var engine = new TesseractEngine(
                    Config.TesseractDataPath, "eng", EngineMode.Default);
                using var img  = Pix.LoadFromMemory(imageBytes);
                using var page = engine.Process(img);
                string text = page.GetText();

                Logger.LogInformation("Extracted text: {Text}", text);

                // 3. POST to webhook
                var payload = new { ImageUrl = imageUrl, ExtractedText = text };
                var content = new StringContent(
                    JsonConvert.SerializeObject(payload),
                    Encoding.UTF8, "application/json");
                await httpClient.PostAsync(Config.WebhookUrl, content);

                // 4. Persist to MongoDB
                var document = new BsonDocument
                {
                    { "imageUrl",      imageUrl },
                    { "extractedText", text     },
                    { "timestamp",     DateTime.UtcNow }
                };
                await collection.InsertOneAsync(document);

                return Result.Success();
            }
            catch (Exception ex)
            {
                Logger.LogError(ex, "Failed to process image: {ImageUrl}", imageUrl);
                return Result.Failure(ex);
            }
        }
    }
}

Register in Program.cs

Wire up the distenka host in your entry point. This adds full CLI support automatically:

Program.cs

using Distenka;

class Program
{
    static async Task Main(string[] args)
    {
        await ProcessorHost
            .CreateDefaultBuilder(args)
            .Build()
            .RunProcessorAsync();
    }
}

Running the process locally

Step 1 — Generate a default config file

Use the built-in get command to scaffold a config JSON with all defaults populated:

Generate default config

dotnet run -- get ImageOCRProcess .

This creates ImageOCRProcess.json in your project directory:

ImageOCRProcess.json (generated defaults)

{
  "imageUrls": null,
  "webhookUrl": null,
  "mongoDbConnectionString": null,
  "databaseName": null,
  "collectionName": null,
  "tesseractDataPath": null,
  "process": "ImageOCRApp.ImageOCRProcess"
}

Step 2 — Edit the config

Fill in real values and tune the execution settings for your workload:

ImageOCRProcess.json (configured)

{
  "imageUrls": [
    "https://example.com/image1.jpg",
    "https://example.com/image2.jpg"
  ],
  "webhookUrl": "https://your-webhook-url.com",
  "mongoDbConnectionString": "mongodb://localhost:27017",
  "databaseName": "imageDb",
  "collectionName": "processedImages",
  "tesseractDataPath": "C:/tessdata",
  "process": "ImageOCRApp.ImageOCRProcess",
  "execution": {
    "parallelTaskCount": 2,
    "itemFailureRetryCount": 3,
    "itemFailureCountToStopProcess": 5
  }
}

parallelTaskCount uses multi-threading without any code changes in your process.
Override any value at runtime: dotnet run -- run --imageUrls "https://test.jpg"

Step 3 — Run

Run the process

dotnet run -- run ImageOCRProcess.json

The output streams live progress to the console: each item as it's processed, any retries, failure details with stack traces, and a final summary: PROCESS SUCCESSFUL — 12 items, 0 failures, 4.2s.

Debugging

IDE debugging — In Visual Studio or VS Code, set the launch args to run ImageOCRProcess.json and use breakpoints as normal.
CLI attach — Add "launchDebugger": true to the execution block; distenka will pause and prompt you to attach before running.
Add breakpoints inside ProcessAsync to inspect OCR output, webhook responses, or MongoDB documents in real time.

← Previous Installation Next → Platform