Sequential task queues in Vault Application Framework applications

[DataContract]
public class Configuration
{
	/// <summary>
	/// If true, running data is logged to the Windows Event Log.
	/// </summary>
	[DataMember( Order = 0 )]
	[JsonConfEditor(
		Label = "Enable Logging",
		DefaultValue = false
	)]
	public bool LoggingEnabled { get; set; } = false;

	/// <summary>
	/// The max number of seconds that can pass between polling intervals.
	/// </summary>
	[DataMember( Order = 1 )]
	[JsonConfIntegerEditor(
		DefaultValue = 15,
		Label = "Maximum Polling Interval",
		HelpText = "The max number of seconds that can pass between polling intervals.",
		Min = 5
	)]
	public int MaxPollingIntervalInSeconds { get; set; } = 10;
}

/// <summary>
/// The entry point for this Vault Application Framework application.
/// </summary>
/// <remarks>Examples and further information available on the developer portal: http://developer.m-files.com/. </remarks>
public class VaultApplication
	: MFiles.VAF.Core.ConfigurableVaultApplicationBase<Configuration>, IUsesTaskQueue
{
	/// <summary>
	/// The token source for task processing cancellation.
	/// </summary>
	private CancellationTokenSource TokenSource { get; set; }

	/// <summary>
	/// The sequential task processor.
	/// </summary>
	public SequentialTaskProcessor TaskProcessor { get; set; }

	/// <summary>
	/// The server that this application is running on.
	/// </summary>
	internal static VaultServerAttachment CurrentServer { get; private set; }

	/// <summary>
	/// The task queue ID. This should be unique for your application
	/// so that you don't interfere with other task queues.
	/// </summary>
	public const string SequentialTaskQueueId = "MFiles.Samples.SequentialTaskQueue.Application.TaskQueueId";
	
	/// <summary>
	/// The task type.
	/// This can be used to have different handlers for different types of task in your queue.
	/// </summary>
	public const string TaskTypeSequentialTask = "TaskType.SequentialTask";

	/// <summary>
	/// Override the start operations so we can register our task processor.
	/// </summary>
	/// <param name="vaultPersistent">Non-transactional Vault object.</param>
	public override void StartOperations(Vault vaultPersistent)
	{
		// Set a reference to the current server reference.
		VaultApplication.CurrentServer = vaultPersistent
			.GetVaultServerAttachments()
			.GetCurrent();

		// Allow the base to process the start operations.
		base.StartOperations(vaultPersistent);

		// Enable polling/processing of the queue.
		this.TaskQueueManager.EnableTaskPolling(true);
	}

	/// <summary>
	/// Registers the task queue used by this application or module.
	/// </summary>
	public void RegisterTaskQueues()
	{
		// TODO: Register the task queue.
	}
}

/// <summary>
/// Registers the task queue used by this application or module.
/// </summary>
public void RegisterTaskQueues()
{
	// Create the cancellation token source.
	if (this.TokenSource == null)
		this.TokenSource = new CancellationTokenSource();

	// Create the task processor.
	if (this.TaskProcessor == null)
	{
		// Initialize the task processor.
		this.TaskProcessor = new SequentialTaskProcessor(
			new AppTaskProcessorSettings
			{
				PollTasksOnJobCompletion = true,
				MaxConcurrentJobs = 1, // Always 1 for a sequential task processor.
				PermanentVault = this.PermanentVault,
				EnableAutomaticTaskUpdates = true,
				QueueDef = new TaskQueueDef
				{
					TaskType = TaskQueueManager.TaskType.ApplicationTasks,
					Id = VaultApplication.SequentialTaskQueueId,
					ProcessingBehavior = MFTaskQueueProcessingBehavior.MFProcessingBehaviorSequential,
					MaximumPollingIntervalInSeconds = this.Configuration.MaxPollingIntervalInSeconds,
					LastBroadcastId = ""
				},
				TaskHandlers = new Dictionary<string, TaskProcessorJobHandler>
				{
					/* One task type is handled here, by the ProcessSequentialTask handler. */
					{ VaultApplication.TaskTypeSequentialTask, ProcessSequentialTask }
				},
				TaskQueueManager = this.TaskQueueManager
			},
			this.TokenSource.Token
		);


	}

	// Register the task queue.
	this.TaskProcessor.RegisterTaskQueues();
}

/// <summary>
/// Processes a single job.
/// </summary>
/// <param name="job">Task processor job.</param>
private void ProcessSequentialTask( TaskProcessorJob job )
{
	// TODO: Process the job itself.
}

/// <summary>
/// Processes a single job.
/// </summary>
/// <param name="job">Task processor job.</param>
private void ProcessSequentialTask( TaskProcessorJob job )
{
	// Debug Logging.
	if( this.Configuration.LoggingEnabled )
		SysUtils.ReportInfoToEventLog( $"Sequential task processing with task id => {job.Data?.Value.Id}." );

	// Ensure cancellation has not been requested.
	job.ThrowIfCancellationRequested();

	// Update the progress of the task in the task queue.
	this.TaskProcessor.UpdateTaskAsAssignedToProcessor( job );

	// Sanity.
	if (null == job.Data?.Value)
	{
		return;
	}

	// Deserialize the directive.
	var dir = TaskQueueDirective.Parse<TaskQueueDirective>( job.Data?.Value );

	// TODO: Perform the processing.
	// Note that the job.Vault reference is not running within a transaction.
	// If transactionality is required then consider calling a vault extension method here.

	// Report back progress.
	this.TaskProcessor.UpdateTaskInfo
	(
		job,
		MFTaskState.MFTaskStateInProgress,
		$"Job processed.",
		false
	);
}

public class ObjVerExTaskQueueDirective
	: TaskQueueDirective
{
	/// <summary>
	/// Parse-able ObjVerEx string.
	/// </summary>
	public string ObjVerEx { get; set; }
}

/// <summary>
/// Processes a single job.
/// </summary>
/// <param name="job">Task processor job.</param>
private void ProcessSequentialTask( TaskProcessorJob job )
{
	// ... method body left out for clarity

	// Deserialize the directive.
	var dir = TaskQueueDirective.Parse<ObjVerExTaskQueueDirective>( job.Data?.Value );

	// Sanity.
	if(string.IsNullOrWhiteSpace(dir?.ObjVerEx))
		return;

	// Load the object from the directive information.
	var objVerEx = ObjVerEx.Parse(job.Vault, dir.ObjVerEx);

	// TODO: Perform the processing.

}

// Create a task in the task queue.
var itemId = this.TaskProcessor.CreateApplicationTaskSafe
(
	true, // allow retries
	VaultApplication.SequentialTaskQueueId, // The queue to add to.
	VaultApplication.TaskTypeSequentialTask // The task type.
);

// Create a task in the task queue.
var itemId = this.TaskProcessor.CreateApplicationTaskSafe
(
	true, // allow retries
	VaultApplication.SequentialTaskQueueId, // The queue to add to.
	VaultApplication.TaskTypeSequentialTask, // The task type.
	new ObjVerExTaskQueueDirective { ObjVerEx = item.ToString() }.ToBytes()
);