Custom workers

Create a deferred worker

Version:

The following sample deferred worker uses the xConnect Client API to create a known contact. The worker accepts a custom worker options dictionary (SampleDeferredWorkerOptionsDictionary) which defines:

An identifier and an identifier source to use for the contact
The fully qualified name of the worker

Create deferred worker options dictionary

You can use the DistributedWorkerOptionsDictionary class to register a task for a custom deferred worker. However, you must know the fully qualified name of the worker and correct name and format of each dictionary entry. To reduce the risk of error, it is recommended that you create custom worker options dictionary for all deferred workers. To create a worker options dictionary:

Create a class named SampleDeferredWorkerOptionsDictionary that inherits DeferredWorkerOptionsDictionary as shown:

using System.Collections.Generic;
using Newtonsoft.Json;
using Sitecore.Processing.Engine.Abstractions;

namespace Sitecore.Documentation.Examples
{
    public class SampleDeferredWorkerOptionsDictionary : DeferredWorkerOptionsDictionary
    {
        public SampleDeferredWorkerOptionsDictionary(string identifier, string identifierSource) : base("Sitecore.Documentation.Examples.SampleDeferredWorker, Sitecore.Documentation.Examples", CreateDictionary(identifier, identifierSource))
        {
        }

        [JsonConstructor]
        protected SampleDeferredWorkerOptionsDictionary(IDictionary<string, string> dictionary) : base(dictionary)
        {
        }

        public static SampleDeferredWorkerOptionsDictionary Parse(IReadOnlyDictionary<string, string> options)
        {
            return new SampleDeferredWorkerOptionsDictionary(options["Identifier"], options["IdentifierSource"]);
        }

        private static IDictionary<string, string> CreateDictionary(string identifier, string identifierSource)
        {
            var dictionary = new Dictionary<string, string>
            {
                { "Identifier", identifier },
                { "IdentifierSource", identifierSource }
            };

            return dictionary;
        }
    }
}

When creating a deferred worker options dictionary:

You must include a constructor that accepts a IDictionary<string, string> parameter. The constructor must be decorated with the [JsonConstructor] attribute. Without this constructor, the Processing Engine will throw the following error when it tries to execute the worker: "System.FormatException: Could not deserialize JSON text."

Create deferred worker class

To create a deferred worker:

Create a class named SampleDeferredWorker that implements IDeferredWorker as shown:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Sitecore.Framework.Conditions;
using Sitecore.Processing.Engine.Abstractions;
using Sitecore.XConnect;

namespace Sitecore.Documentation.Examples
{
    public class SampleDeferredWorker : IDeferredWorker
    {
        private readonly ILogger _logger;
        private readonly SampleDeferredWorkerOptionsDictionary _options;
        private readonly IServiceProvider _serviceProvider;

        public SampleDeferredWorker(ILogger logger, SampleDeferredWorkerOptionsDictionary options, IServiceProvider serviceProvider)
        {
            _logger = logger;
            _serviceProvider = serviceProvider;
            _options = options;
        }

        public SampleDeferredWorker(ILogger logger, IReadOnlyDictionary<string, string> options, IServiceProvider serviceProvider) : this(logger, SampleDeferredWorkerOptionsDictionary.Parse(options), serviceProvider)
        {
        }

        public async Task RunAsync(CancellationToken token)
        {
            _logger.LogInformation($"Run Worker: {GetType().Name}");

            using (IServiceScope scope = _serviceProvider.CreateScope())
            {
                using (var xdbContext = scope.ServiceProvider.GetRequiredService<IXdbContext>())
                {
                    var testContact = new Contact(
                        new ContactIdentifier(_options["IdentifierSource"], _options["Identifier"], ContactIdentifierType.Known));

                    xdbContext.AddContact(testContact);

                    await xdbContext.SubmitAsync(CancellationToken.None);

                    // We need to execute search operation to make sure that contacts are indexed
                    List<Contact> contacts = await xdbContext.Contacts.Where(c => c.Identifiers.Any(i => i.Source == _options["Identifier"])).ToList(token);
                    Condition.Ensures(contacts, nameof(contacts)).IsNotEmpty();
                }
            }
        }

        public void Dispose()
        {
        }
    }
}

When creating a deferred worker:

You must have constructor that accepts an IReadOnlyDictionary<string, string> parameter. If you do not include this constructor, the Processing Engine will not start and you will get the following error: "System.InvalidOperationException: A suitable constructor for type could not be located."

Register worker in configuration

To register the worker in the Cortex Processing Engine:

Create a file named C:\<PathToxConnect>\App_data\jobs\continuous\ProcessingEngine\App_Data\Config\Global\Processing\sc.Processing.Documentation.Examples.xml and add the following configuration:

<Settings>
    <Sitecore>
        <Processing>
        <Services>
        <TaskServicesFactory>
            <Options>
                <SampleDeferredWorker>
                <Type>Sitecore.Documentation.Examples.SampleDeferredWorker, Sitecore.Documentation.Examples</Type>
                </SampleDeferredWorker>
            </Options>
            </TaskServicesFactory>
        </Services>
        </Processing>
    </Sitecore>
</Settings>

Restart the Cortex Processing Engine. This is required to register changes in configuration.

Use custom deferred worker

To use the custom worker in a core role environment (such as Content Management):

using System;
using Microsoft.Extensions.DependencyInjection;
using Sitecore.DependencyInjection;
using Sitecore.Diagnostics;
using Sitecore.Processing.Engine.Abstractions;

namespace Sitecore.Documentation.Examples
{
    public class Example
    {
        public async void AsyncExamples()
        {
            var _taskManager = ServiceLocator.ServiceProvider.GetRequiredService<ITaskManager>();

            Guid taskId = await _taskManager.RegisterDeferredTaskAsync(
                    new SampleDeferredWorkerOptionsDictionary("MyrtleMcSitecore", "SampleIdentifierSource"),
                    null,
                    TimeSpan.FromHours(1))
                .ConfigureAwait(false);
        }
    }
}

If you have suggestions for improving this article, let us know!