ContentFlowAggregationProcessor.cs

public class ContentFlowAggregationProcessor : AggregationProcessor
{
    protected override void OnProcess(AggregationPipelineArgs args)
    {
        Assert.ArgumentNotNull(args, "args");

        VisitData visit = args.Context.Visit;

        if (visit.Pages == null || visit.Pages.Count <= 0)
        {
            return;
        }

        ContentFlow contentEvents = null;

        foreach (PageData page in visit.Pages)
        {
            if (page.PageEvents == null || page.PageEvents.Count <= 0)
            {
                continue;
            }

            foreach (PageEventData pageEvent in page.PageEvents.Where(pe => pe.IsGoal))
            {
                var itemId = new ID(pageEvent.ItemId);

		ContentFlowKey contentFlowKeyObj = new ContentFlowKey
		{
			Timestamp = visit.SaveDateTime,
			ContactId = visit.ContactId,
			ItemId = pageEvent.ItemId,
			ContentType = Factory.GetDatabase("master").GetItem(new ID(pageEvent.ItemId)).TemplateName,
			Type = "A custom type"
		};

		ContentFlowValue contentViewValue = new ContentFlowValue { Count = 1 };

		if (contentEvents == null)
		{
			contentEvents = args.GetFact<ContentFlow>();
		}

		contentEvents.Emit(contentFlowKeyObj, contentViewValue);
            }
        }
    }
}