Broadcast 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 max number of batches that can be running in the task processor at once.
	/// </summary>
	[DataMember( Order = 2 )]
	[JsonConfIntegerEditor(
		HelpText = "The max number of batches that can be running in the task processor at once.",
		Label = "Max Concurrent Batches",
		Min = 1,
		Max = 100,
		DefaultValue = 5
	)]
	public int MaxConcurrentBatches { get; set; } = 5;

	/// <summary>
	/// The max number of jobs that can be running within a batch at once.
	/// </summary>
	[DataMember( Order = 3 )]
	[JsonConfIntegerEditor(
		HelpText = "The max number of jobs that can be running within a batch at once.",
		Label = "Maximum Concurrent Jobs",
		Min = 5,
		Max = 100,
		DefaultValue = 5
	)]
	public int MaxConcurrentJobs { get; set; } = 5;
}

/// <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 broadcast task processor.
	/// </summary>
	public AppTaskBatchProcessor 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 BroadcastTaskQueueId = "MFiles.Samples.BroadcastTaskQueue.Application.BroadcastQueueId";
	
	/// <summary>
	/// The task type.
	/// This can be used to have different handlers for different types of task in your queue.
	/// </summary>
	public const string TaskTypeBroadcastTask = "TaskType.BroadcastTask";

	/// <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 AppTaskBatchProcessor
		(
			new AppTaskBatchProcessorSettings
			{
				QueueDef = new TaskQueueDef
				{
					TaskType = TaskQueueManager.TaskType.BroadcastMessages,
					Id = BroadcastTaskQueueId,
					ProcessingBehavior = MFTaskQueueProcessingBehavior.MFProcessingBehaviorConcurrent,
					MaximumPollingIntervalInSeconds = this.Configuration.MaxPollingIntervalInSeconds,
					LastBroadcastId = ""
				},
				PermanentVault = this.PermanentVault,
				MaxConcurrentBatches = this.Configuration.MaxConcurrentBatches,
				MaxConcurrentJobs = this.Configuration.MaxConcurrentJobs,
				TaskHandlers = new Dictionary<string, TaskProcessorJobHandler>
				{
					{ TaskTypeBroadcastTask, ProcessBroadcastTask }
				},
				TaskQueueManager = this.TaskQueueManager,
				EnableAutomaticTaskUpdates = true,
				DisableAutomaticProgressUpdates = false,
				PollTasksOnJobCompletion = true
			},
			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 ProcessBroadcastTask( TaskProcessorJob job )
{
	// TODO: Process the job itself.
}

/// <summary>
/// Processes a single job.
/// </summary>
/// <param name="job">Task processor job.</param>
private void ProcessBroadcastTask(TaskProcessorJob job)
{
	// 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.
	EmailByUserPatchDirective dir = TaskQueueDirective.Parse<EmailByUserPatchDirective>( job.Data.Value );

	// If this server generated it then ignore it.
	if( dir.GeneratedFromGuid == CurrentServer.ServerID )
	{
		return;
	}

	// TODO: Process the broadcast job.

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

}

public class EmailByUserPatchDirective
	: BroadcastDirective
{
	/// <summary>
	/// User account corresponding to the email.
	/// </summary>
	public int UserAccount { get; set; }

	/// <summary>
	/// Email corresponding to the User account.
	/// </summary>
	public string Email { get; set; }
}

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

	// Deserialize the directive.
	BroadcastDirective dir = TaskQueueDirective.Parse<EmailByUserPatchDirective>( job.Data?.Value );
	if( dir.GeneratedFromGuid == CurrentServer.ServerID )
	{
		return;
	}

	// TODO: Process the broadcast job.

}

var itemId = this.TaskProcessor.SendBroadcastMessageSafe
(
	true,
	VaultApplication.BroadcastTaskQueueId,
	VaultApplication.TaskTypeBroadcastTask,
	new EmailByUserPatchDirective
		{
			UserAccount = Int32.Parse(env?.Input.Split("****".ToCharArray())[0]),
			Email = env?.Input.Split("****".ToCharArray())[1]
		}.ToBytes()
);