0

I would like to start a pipeline batch and pass in the context required for a single contact to be processed so that I can trigger the sync process for individual contacts when they are created in the xDB.

So for example in the "Read Contacts from xConnect Pipeline" batch which will sync all contacts, I would like to have it only read a single contact that I would like processed.

I can see that I can trigger the pipeline via Sitecore.DataExchange.Local.Runners.InProcessPipelineBatchRunner. How can I set the contact on the context?

DEF 2.1 on Sitecore 9.1.1.

  • Do you have any criteria that which contact will be processed? – Swati Gupta Aug 30 '19 at 5:52
  • Yes, essentially the currently identified contact. – Mark Gibbons Aug 30 '19 at 5:53
0

I had the same requirement not so long ago. Sitecore provides a code to ran PipelineBatch programmatically: https://doc.sitecore.com/developers/def/v1.3/synchronization/start-pipeline-batch-programmatically.html

Guid tenantId = ...         //tenant that owns the pipeline batch
Guid pipelineBatchId = ...  //pipeline batch
IEnumerable<PipelineBatch> batches = null;
batches = Sitecore.DataExchange.Context.TenantRepository.GetPipelineBatches(tenantId);
if (batches != null)
{
  PipelineBatch pipelineBatch = batches.FirstOrDefault(x => x.ID == pipelineBatchId);
  if (pipelineBatch != null)
  {
    var runner = new InProcessPipelineBatchRunner();
    if (runner.Run(pipelineBatch))
    {
      //pipeline batch was started
    }
  }
}

But this code doesn't show how to put some data (like contact) to the pipeline. Also the code runs in the same thread where it is run from. That can cause a performance issue.

I have implemented own PipelineBatchRunner. You can find BatchRunner.cs class there with RunStandardBatch method:

public void RunStandardBatch(ID batchItemId, IPlugin[] plugins)
        {
            var batchItem = Sitecore.Configuration.Factory.GetDatabase("master")
                .GetItem(batchItemId);

            if (PipelineBatchRunner == null || batchItem == null || !Helper.IsPipelineBatchItem(batchItem) || !Helper.IsItemEnabled(batchItem))
                return;

            var pipelineBatch = Helper.GetPipelineBatch(batchItem);

            Run(pipelineBatch, plugins);
        }

You just need to pass a BatchItemID and list of Plugins where Plugin is any class which implements Sitecore.DataExchange.IPlugin interface. For example:

public class SingleContactSyncPlugin : Sitecore.DataExchange.IPlugin
    {
        public List<Guid> ContactIdsToSync { get; set; }
    }

This plugins will be put into Pipelines Context. Also, the BatchRunner runs the sync as Sitecore Job thereby not blocking the main thread.

So, we have a code that allows to put something into context. Now we need to read it. I have implemented a ReadContextContact pipeline step which I use instead of original Read Contacts pipeline step:

using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Linq;
using System.Linq.Expressions;
using System.Threading.Tasks;
using Sitecore.DataExchange.Attributes;
using Sitecore.DataExchange.Contexts;
using Sitecore.DataExchange.Models;
using Sitecore.DataExchange.Plugins;
using Sitecore.DataExchange.Processors.PipelineSteps;
using Sitecore.DataExchange.Providers.XConnect;
using Sitecore.DataExchange.Providers.XConnect.Client;
using Sitecore.DataExchange.Providers.XConnect.Expressions;
using Sitecore.DataExchange.Providers.XConnect.Models;
using Sitecore.DataExchange.Providers.XConnect.Plugins;
using Sitecore.Services.Core.Diagnostics;
using Sitecore.XConnect;
using Sitecore.XConnect.Operations;

namespace SomeNamespace.Processors
{
    [RequiredEndpointPlugins(typeof(XConnectClientEndpointSettings))]
    [RequiredPipelineStepPlugins(typeof(ReadEntitySettings), typeof(LoadInteractionsSettings))]
    public class ReadContextContactProcessor : BaseReadDataStepProcessor
    {

        protected override void ReadData(Endpoint endpoint, PipelineStep pipelineStep, PipelineContext pipelineContext, ILogger logger)
        {
            var xconnectClientHelper = GetXConnectClientHelper(endpoint);
            if (xconnectClientHelper == null)
            {
                pipelineContext.CriticalError = true;
                Log(logger.Error, pipelineContext, "Cannot get xconnect client helper.", Array.Empty<string>());
            }
            else
            {
                var xconnectClientSettings = GetXConnectClientSettings(endpoint, pipelineStep, pipelineContext, logger);
                if (xconnectClientSettings == null)
                {
                    pipelineContext.CriticalError = true;
                    Log(logger.Error, pipelineContext, "Cannot get xconnect client settings.", Array.Empty<string>());
                }
                else
                {
                    var singleContactSyncPlugin = pipelineContext.PipelineBatchContext.GetPlugin<SingleContactSyncPlugin>();
                    if (singleContactSyncPlugin == null)
                    {
                        pipelineContext.CriticalError = true;
                        Log(logger.Error, pipelineContext, "Cannot get xconnect client helper.", Array.Empty<string>());
                    }
                    else
                    {
                        var xconnectClient = xconnectClientHelper.ToXConnectClient(xconnectClientSettings);

                        var contactExpandOptions = ResolveExpandOptions(pipelineStep, pipelineContext, logger);

                        var contacts = xconnectClient.GetContactsAsync(GetContactReference(singleContactSyncPlugin.ContactIdsToSync), contactExpandOptions);

                        List<EntityModel> result = GetContactModels(contacts, xconnectClientHelper, pipelineContext, logger).Result;

                        IterableDataSettings newPlugin = new IterableDataSettings(result);
                        string message = result.Any() ? $"Entities were read from xConnect. (total count: {result.Count})" : "No entities were read from xConnect.";
                        Log(logger.Info, pipelineContext, message, $"endpoint: {endpoint.Name}", "entity: contact");
                        pipelineContext.AddPlugin(newPlugin);
                    }
                }
            }
        }

        protected virtual ReadEntitySettings GetReadEntitySettings(PipelineStep pipelineStep, PipelineContext pipelineContext, ILogger logger)
        {
            return pipelineStep.GetPlugin<ReadEntitySettings>();
        }

        protected virtual IXConnectClientHelper GetXConnectClientHelper(Endpoint endpoint)
        {
            XConnectClientEndpointSettings plugin = endpoint?.GetPlugin<XConnectClientEndpointSettings>();
            return plugin?.ClientHelper;
        }

        protected virtual List<IEntityReference<Contact>> GetContactReference(List<Guid> ids)
        {
            return ids.Select(id => new ContactReference(id)).Cast<IEntityReference<Contact>>().ToList();
        }

        protected virtual XConnectClientSettings GetXConnectClientSettings(Endpoint endpoint, PipelineStep pipelineStep, PipelineContext pipelineContext, ILogger logger)
        {
            var plugin = endpoint?.GetPlugin<XConnectClientEndpointSettings>();

            return plugin?.ClientSettings;
        }

        protected virtual ContactExpandOptions ResolveExpandOptions(PipelineStep pipelineStep, PipelineContext pipelineContext, ILogger logger)
        {
            var contactExpandOptions = new ContactExpandOptions(GetReadEntitySettings(pipelineStep).FacetNames.ToArray());
            var interactionsSettings = GetLoadInteractionsSettings(pipelineStep);
            if (interactionsSettings.LoadInteractions)
                contactExpandOptions.Interactions = new RelatedInteractionsExpandOptions(interactionsSettings.FacetNames.ToArray())
                {
                    Limit = interactionsSettings.MaxCount
                };
            return contactExpandOptions;
        }

        protected virtual LoadInteractionsSettings GetLoadInteractionsSettings(PipelineStep pipelineStep)
        {
            return pipelineStep.GetPlugin<LoadInteractionsSettings>();
        }

        protected virtual ReadEntitySettings GetReadEntitySettings(PipelineStep pipelineStep)
        {
            return pipelineStep.GetPlugin<ReadEntitySettings>();
        }

        private async Task<List<EntityModel>> GetContactModels(Task<ReadOnlyCollection<IEntityLookupResult<Contact>>> task, IXConnectClientHelper clientHelper, PipelineContext pipelineContext, ILogger logger)
        {
            List<EntityModel> items = new List<EntityModel>();

            var result = await task;

            foreach (var contact in result)
            {
                if (!contact.Exists)
                {
                    Log(logger.Info, pipelineContext, $"Contact with Contact ID = {contact.Reference.Id} is not found and was skipped.", Array.Empty<string>());
                    continue;
                }

                items.Add(clientHelper.ToContactModel(contact.Entity));
            }

            return items;
        }
    }
}

Also, you require to implement a converter for the processor above:

using System.Collections.Generic;
using Sitecore.DataExchange.Attributes;
using Sitecore.DataExchange.Converters.PipelineSteps;
using Sitecore.DataExchange.Models;
using Sitecore.DataExchange.Providers.XConnect.Plugins;
using Sitecore.DataExchange.Repositories;
using Sitecore.Services.Core.Model;

namespace SomeNamespace.Processors
{
    [SupportedIds("{YOUR CUSTOM PROCESSOR TEMPLATE ID}")]
    public class ReadContextContactConverter : BaseResolveObjectFromEndpointStepConverter
    {
        public const string FieldNameContactFacetsToRead = "ContactFacetsToRead";
        public const string FieldNameInteractionFacetsToRead = "InteractionFacetsToRead";
        public const string FieldNameLoadInteractions = "LoadInteractions";
        public const string FieldNameInteractionMaxCount = "InteractionMaxCount";
        public const string FieldNameFacetName = "FacetName";

        public ReadContactsFromExternalDataStepConverter(IItemModelRepository repository)
          : base(repository)
        {
        }

        protected override void AddPlugins(ItemModel source, PipelineStep pipelineStep)
        {
            base.AddPlugins(source, pipelineStep);
            AddLoadInteractionsSettings(source, pipelineStep);
            AddReadEntitySettings(source, pipelineStep);
        }

        protected virtual void AddReadEntitySettings(ItemModel source, PipelineStep pipelineStep)
        {
            ReadEntitySettings newPlugin = new ReadEntitySettings
            {
                FacetNames = new List<string>(0)
            };
            var referencesAsModels = GetReferencesAsModels(source, FieldNameContactFacetsToRead);
            if (referencesAsModels != null)
            {
                foreach (var itemModel in referencesAsModels)
                {
                    string stringValue = GetStringValue(itemModel, FieldNameFacetName);
                    newPlugin.FacetNames.Add(stringValue);
                }
            }
            pipelineStep.AddPlugin(newPlugin);
        }

        protected virtual void AddLoadInteractionsSettings(ItemModel source, PipelineStep pipelineStep)
        {
            LoadInteractionsSettings interactionsSettings =
                new LoadInteractionsSettings { MaxCount = GetIntValue(source, FieldNameInteractionMaxCount) };
            int num = GetBoolValue(source, FieldNameLoadInteractions) ? 1 : 0;
            interactionsSettings.LoadInteractions = num != 0;
            List<string> stringList = new List<string>();
            interactionsSettings.FacetNames = stringList;
            LoadInteractionsSettings newPlugin = interactionsSettings;
            IEnumerable<ItemModel> referencesAsModels = GetReferencesAsModels(source, FieldNameInteractionFacetsToRead);
            if (referencesAsModels != null)
            {
                foreach (var itemModel in referencesAsModels)
                {
                    string stringValue = GetStringValue(itemModel, FieldNameFacetName);
                    newPlugin.FacetNames.Add(stringValue);
                }
            }
            pipelineStep.AddPlugin(newPlugin);
        }
    }
}

The PipelineStap above gets Contact Ids from the plugin which you put into the context when run the PipelineBatch and reads this contacts from xConnect.

| improve this answer | |
  • Is it helped? Or do you look for something else? – Artsem Prashkovich Sep 6 '19 at 12:35
0

To process single contact, you can create a custom pipeline using below steps:

1.Create a custom template which is inherited from /sitecore/templates/Data Exchange/Framework/Pipeline Steps/Base Pipeline Steps/Base Pipeline Step.

  1. Add two fields IterableDataLocation,DataLocation in the template, set Source and using this template create custom pipeline step

enter image description here

  1. I have created custom processor CustomFilterPipelineStepConverter
 [SupportedIds(new string[]
     {
        "{12032AD0-E66E-408E-B1D4-048A0B2D659C}"
     })]
    public class CustomFilterPipelineStepConverter : BasePipelineStepConverter
    {
        public CustomFilterPipelineStepConverter(IItemModelRepository repository) : base(repository)
        {
        }

        protected override void AddPlugins(ItemModel source, PipelineStep pipelineStep)
        {
            this.AddDataLocationSettingsPlugin(source, pipelineStep);
            this.AddReadEntitySettings(source, pipelineStep);
        }

        private void AddDataLocationSettingsPlugin(ItemModel source, PipelineStep pipelineStep)
        {
            Guid guidValue = base.GetGuidValue(source, "DataLocation");
            Guid guidValue2 = base.GetGuidValue(source, "IterableDataLocation");
            pipelineStep.AddPlugin<IterateDataSettings>(new IterateDataSettings
            {
                DataLocation = guidValue,
                IterableDataLocation = guidValue2
            });
        }
        protected virtual void AddReadEntitySettings(ItemModel source, PipelineStep pipelineStep)
        {
           //IF you want to read some field
        }
    }


  1. Create custom processor CustomFilterPipelineStepProcessor
[RequiredPipelineStepPlugins(new Type[]
     {
        typeof(PipelinesSettings),
        typeof(IterateDataSettings)
     })]
    public class CustomFilterPipelineStepProcessor : BasePipelineStepProcessor
    {
        protected override bool CanStartProcessing(PipelineStep pipelineStep, PipelineContext pipelineContext, ILogger logger)
        {
            if (pipelineStep.GetPlugin<IterateDataSettings>().DataLocation == Guid.Empty)
            {
                this.Log(new Action<string>(logger.Error), pipelineContext, "No data location is specified.", Array.Empty<string>());
                return false;
            }
            return true;
        }

        protected override void ProcessPipelineStep(PipelineStep pipelineStep, PipelineContext pipelineContext, ILogger logger)
        {
            Sitecore.Diagnostics.Log.Info("Batch has proceeded to Custom Pipeline Step" + DateTime.Now.ToString(), this);
            IEnumerable iterableData = this.GetIterableData(pipelineStep, pipelineContext, logger);
            if (iterableData == null)
            {
                this.Log(new Action<string>(logger.Debug), pipelineContext, "Iterable data is null.", Array.Empty<string>());
                return;
            }
            //IF you have custom field
            //ReadFacetSettings facetPlugin = pipelineStep.GetPlugin<ReadFacetSettings>();
            //if (facetPlugin == null )
            //{
            //    this.Log(new Action<string>(logger.Debug), pipelineContext, "Contact Facets To Read is null.", Array.Empty<string>());
            //    return;
            //}


            List<EntityModel> list = iterableData.Cast<EntityModel>().ToList();
            List<EntityModel> filterList = new List<EntityModel>();
            this.Log(new Action<string>(logger.Debug), pipelineContext, "CustomFilterlist" + list.Count, Array.Empty<string>());
            foreach (var data in list)
            {
                //Apply your logic to get currently identified contact
            }
            IterableDataSettings newPlugin = new IterableDataSettings(filterList);
            pipelineContext.AddPlugin<IterableDataSettings>(newPlugin);
        }

        protected virtual IEnumerable GetIterableData(PipelineStep pipelineStep, PipelineContext pipelineContext, ILogger logger)
        {
            IterateDataSettings plugin = pipelineStep.GetPlugin<IterateDataSettings>();
            return this.GetObjectFromPipelineContext(plugin.IterableDataLocation, pipelineContext, logger) as IEnumerable;
        }

    }
  1. In the custom pipeline, add namespace and assembly name for Converter Type and Processor Type

enter image description here

You can read more details in this blog https://www.codinglittlethings.com/2019/06/sitecore-9-dynamic-crm-connector-custom.html

| improve this answer | |

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.