Subtask.Storage.MongoDB 1.0.1

Subtask.Storage.MongoDB

Subtask.Storage.MongoDB is the server side of the Subtask library, used to manage and schedule the running of processes.

Installation

The Subtask.Storage.MongoDB Nuget package can be found at https://nuget.org/packages/Subtask.Storage.MongoDB/.

To install it via the .NET CLI, simply enter:

dotnet add package Subtask.Storage.MongoDB

Examples

Initialization - DefaultStorage

var client = new MongoClient(/*YOUR CONNECTION STRING*/);
var database = client.GetDatabase(databaseName); // Create a database if databaseName doesn't exist yet
database.CreateCollection(collectionName); // Create a collection
var collection = database.GetCollection<MDProcess>(collectionName);
var storage = new DefaultStorage(collection, collectionName);

Get a Process

bool result = storage.TryGetProcess(
    tenantId, 
    idToGet,
    out IProcess process, 
    new CancellationToken()
);

Get a Process in a Specific Application

bool result = storage.TryGetProcessByApplication(
    applicationName, 
    idToGet,
    out IProcess process, 
    new CancellationToken()
);

Get All Processes in a Tenant ID

IAsyncCursor<IProcess> cursor = storage.GetProcesses("TestDefaultStorage Tenant 1");

Get All Processes

IAsyncCursor<IProcess> cursor = storage.GetProcesses();

Create a Process

var process = await storage.GetOrCreateProcess(
    applicationName,
    tenantId,
    user,
    new SubtaskMeta(
        name, 
        typeof(MDProcess), 
        new ThreadRelieverSettingsFactory(),
        uniqueKey
    ),
    new TestProcessInput(5),
    new CancellationToken()
);

Get the Next Process for Execution

var result = await storage.TryGetNextProcessForExecution(applicationName, new CancellationToken());

Updating Process Status

await storage.UpdateProcessPreparing(idToUpdate, new CancellationToken());
await storage.UpdateProcessRunning(idToUpdate, new CancellationToken());
await storage.UpdateProcessStopping(idToUpdate, new CancellationToken());
await storage.UpdateProcessStopped(idToUpdate, new CancellationToken());
await storage.UpdateProcessFailed(idToUpdate, failMessage, /* bool */ isComingFromRuntime, new CancellationToken());
await storage.UpdateProcessError(idToUpdate, errorMessage, new CancellationToken());

Remove Completed Processes in a Tenant ID

await storage.RemoveCompletedProcesses(tenantId, new CancellationToken());

Remove Completed Processes from All Available Processes

await storage.RemoveCompletedProcesses(new CancellationToken());

Remove Single Completed Process in a Tenant ID

await _storage.RemoveCompletedProcess(tenantId, idToRemove, new CancellationToken());

Delete a Process

await storage.DeleteProcess(idToDelete, new CancellationToken());

Reset a Failed Process

TryResults<IProcess> result = await storage.TryResetProcess(idToReset, new CancellationToken());

Ping a Process

await storage.Ping(idToPing, new CancellationToken());

Cleaning up Orphan Runners in an Application

Fails processes that have not been pinged (inactive) in the last 60 seconds.

await storage.CleanupOrphanRunners(applicationName, new CancellationToken())