Enable custom processor environments in the Engine. (#383)

* Add ProcessorEnvironmentFactory

* ProcessEnvironmentFactory => ProcessorEnvironmentFactory

* Small cleanup.

src/Microsoft.Performance.Toolkit.Engine/Engine.cs

@@ -1065,9 +1065,9 @@ namespace Microsoft.Performance.Toolkit.Engine
 
             foreach (var psDsgPair in processingOptionsMap)
             {
-                var processingSource = psDsgPair.Key.Item1;
+                ProcessingSourceReference processingSource = psDsgPair.Key.Item1;
-                var dsg = psDsgPair.Key.Item2;
+                IDataSourceGroup dsg = psDsgPair.Key.Item2;
-                var processorOptions = psDsgPair.Value;
+                ProcessorOptions processorOptions = psDsgPair.Value;
 
                 try
                 {
@@ -1087,11 +1087,11 @@ namespace Microsoft.Performance.Toolkit.Engine
 
                     var executionContext = new SDK.Runtime.ExecutionContext(
                         new DataProcessorProgress(),
-                        x => ConsoleLogger.Create(x.GetType()),
+                        x => ConsoleLogger.Create(x.GetType()), // todo: this shouldn't be using a console logger by default
                         processingSource,
                         dsg, // todo #214
                         processingSource.Instance.MetadataTables,
-                        new RuntimeProcessorEnvironment(this.Extensions, this.compositeCookers, this.CreateLogger),
+                        this.CreateInfo.ProcessEnvironmentFactory.CreateProcessorEnvironment(processingSource.Guid, dsg),
                         processorOptions);
 
                     var executor = new ProcessingSourceExecutor();
@@ -1221,62 +1221,6 @@ namespace Microsoft.Performance.Toolkit.Engine
             }
         }
 
-        private sealed class RuntimeProcessorEnvironment
-            : IProcessorEnvironment
-        {
-            private readonly ProcessingSystemCompositeCookers compositeCookers;
-            private readonly IDataExtensionRepository repository;
-            private readonly Func<Type, ILogger> loggerFactory;
-            private readonly object loggerLock = new object();
-
-            private ILogger logger;
-            private Type processorType;
-
-            public RuntimeProcessorEnvironment(
-                IDataExtensionRepository repository,
-                ProcessingSystemCompositeCookers compositeCookers,
-                Func<Type, ILogger> loggerFactory)
-            {
-                Debug.Assert(repository != null);
-                Debug.Assert(compositeCookers != null);
-                Debug.Assert(loggerFactory != null);
-
-                this.compositeCookers = compositeCookers;
-                this.repository = repository;
-                this.loggerFactory = loggerFactory;
-            }
-
-            public ILogger CreateLogger(Type processorType)
-            {
-                Guard.NotNull(processorType, nameof(processorType));
-
-                lock (this.loggerLock)
-                {
-                    if (logger != null)
-                    {
-                        if (this.processorType != processorType)
-                        {
-                            throw new ArgumentException(
-                                $"{nameof(CreateLogger)} cannot be called with multiple types in a single instance.",
-                                nameof(processorType));
-                        }
-
-                        return this.logger;
-                    }
-
-                    this.processorType = processorType;
-                    this.logger = this.loggerFactory(processorType);
-                    return this.logger;
-                }
-            }
-
-            public IDynamicTableBuilder RequestDynamicTableBuilder(
-                TableDescriptor descriptor)
-            {
-                return null;
-            }
-        }
-
         private abstract class RuntimeMessageBox
             : IMessageBox
         {

src/Microsoft.Performance.Toolkit.Engine/EngineCreateInfo.cs

@@ -7,6 +7,7 @@ using System.Collections.ObjectModel;
 using Microsoft.Performance.SDK;
 using Microsoft.Performance.SDK.Auth;
 using Microsoft.Performance.SDK.Processing;
+using Microsoft.Performance.SDK.Runtime;
 
 namespace Microsoft.Performance.Toolkit.Engine
 {
@@ -16,10 +17,12 @@ namespace Microsoft.Performance.Toolkit.Engine
     /// </summary>
     public sealed class EngineCreateInfo
     {
-        private static string DefaultRuntimeName;
+        private static readonly string DefaultRuntimeName;
 
         private readonly Dictionary<Type, object> authProviders = new Dictionary<Type, object>();
 
+        private ProcessorEnvironmentFactory processEnvironmentFactory;
+
         /// <summary>
         ///     Initializes the statc members of the <see cref="EngineCreateInfo"/>
         ///     class.
@@ -134,6 +137,27 @@ namespace Microsoft.Performance.Toolkit.Engine
             return this;
         }
 
+        /// <summary>
+        ///     Registers a <see cref="ProcessEnvironmentFactory"/> to use for generating a custom processor 
+        ///     environment.
+        /// </summary>
+        /// <param name="factory">
+        ///     Factory for generating a <see cref="ProcessorEnvironment"/>.
+        /// </param>
+        /// <returns>
+        ///     The instance of <see cref="EngineCreateInfo"/>.
+        /// </returns>
+        /// <exception cref="ArgumentNullException">
+        ///     <paramref name="factory"/> is <c>null</c>.
+        /// </exception>
+        public EngineCreateInfo WithProcessorEnvironmentFactory(ProcessorEnvironmentFactory factory)
+        {
+            Guard.NotNull(factory, nameof(factory));
+
+            this.ProcessEnvironmentFactory = factory;
+            return this;
+        }
+
         /// <summary>
         ///     Gets or sets the name of the runtime on which the application is built.
         /// </summary>
@@ -174,6 +198,22 @@ namespace Microsoft.Performance.Toolkit.Engine
         /// </summary>
         public bool IsInteractive { get; set; }
 
+        internal ProcessorEnvironmentFactory ProcessEnvironmentFactory
+        {
+            get
+            {
+                // Wrap any custom factory in the default factory. If there is no custom factory, or if it returns null,
+                // then a default runtime will be created.
+                return new RuntimeProcessorEnvironmentFactory(
+                    (type) => this.LoggerFactory?.Invoke(type) ?? Logger.Create(type),
+                    this.processEnvironmentFactory);
+            }
+            private set
+            {
+                this.processEnvironmentFactory = value;
+            }
+        }
+
         internal ReadOnlyDictionary<Type, object> AuthProviders => new ReadOnlyDictionary<Type, object>(this.authProviders);
     }
 }

src/Microsoft.Performance.Toolkit.Engine/ProcessorEnvironment.cs

@@ -0,0 +1,26 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using Microsoft.Performance.SDK.Processing;
+
+namespace Microsoft.Performance.Toolkit.Engine
+{
+    /// <inheritdoc cref="IProcessorEnvironment"/>
+    public abstract class ProcessorEnvironment
+        : IProcessorEnvironment
+    {
+        /// <inheritdoc/>
+        public abstract ILogger CreateLogger(Type processorType);
+
+        /// <inheritdoc />
+        /// <remarks>
+        ///     This implementation does not support the concept of dynamic table builder and always returns 
+        ///     <c>null</c>.
+        /// </remarks>
+        public virtual IDynamicTableBuilder RequestDynamicTableBuilder(TableDescriptor descriptor)
+        {
+            return null;
+        }
+    }
+}

src/Microsoft.Performance.Toolkit.Engine/ProcessorEnvironmentFactory.cs

@@ -0,0 +1,33 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using Microsoft.Performance.SDK.Processing;
+using Microsoft.Performance.SDK.Processing.DataSourceGrouping;
+
+namespace Microsoft.Performance.Toolkit.Engine
+{
+    /// <summary>
+    ///     This is used to generate processor environments.
+    /// </summary>
+    public abstract class ProcessorEnvironmentFactory
+    {
+        /// <summary>
+        ///     Creates a <see cref="ProcessorEnvironment"/> for a processor.
+        /// </summary>
+        /// <param name="processingSourceIdentifier">
+        ///     Identifies the source processor used to generate the data processor.
+        /// </param>
+        /// <param name="dataSourceGroup">
+        ///     A collection of <see cref="IDataSource"/>s that a <see cref="ICustomDataProcessor"/> can process together
+        ///     in a specified <see cref="IProcessingMode"/>.
+        /// </param>
+        /// <returns>
+        ///     A <see cref="ProcessorEnvironment"/> or <c>null</c>. When <c>null</c> is returned, a default processor 
+        ///     environment will be used.
+        /// </returns>
+        public abstract ProcessorEnvironment CreateProcessorEnvironment(
+            Guid processingSourceIdentifier,
+            IDataSourceGroup dataSourceGroup);
+    }
+}

src/Microsoft.Performance.Toolkit.Engine/RuntimeProcessorEnvironmentFactory.cs

@@ -0,0 +1,75 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Diagnostics;
+using Microsoft.Performance.SDK;
+using Microsoft.Performance.SDK.Processing;
+using Microsoft.Performance.SDK.Processing.DataSourceGrouping;
+
+namespace Microsoft.Performance.Toolkit.Engine
+{
+    internal sealed class RuntimeProcessorEnvironmentFactory
+        : ProcessorEnvironmentFactory
+    {
+        private readonly Func<Type, ILogger> loggerFactory;
+        private readonly ProcessorEnvironmentFactory wrappedFactory;
+
+        public RuntimeProcessorEnvironmentFactory(Func<Type, ILogger> loggerFactory, ProcessorEnvironmentFactory wrappedFactory)
+        {
+            Debug.Assert(loggerFactory != null);
+            this.loggerFactory = loggerFactory;
+            this.wrappedFactory = wrappedFactory;
+        }
+
+        public override ProcessorEnvironment CreateProcessorEnvironment(
+            Guid processingSourceIdentifier,
+            IDataSourceGroup dataSourceGroup)
+        {
+            return this.wrappedFactory?.CreateProcessorEnvironment(processingSourceIdentifier, dataSourceGroup)
+                ?? new RuntimeProcessorEnvironment(this.loggerFactory);
+        }
+
+        private sealed class RuntimeProcessorEnvironment
+            : ProcessorEnvironment
+        {
+            private readonly Func<Type, ILogger> loggerFactory;
+            private readonly object loggerLock = new object();
+
+            private ILogger logger;
+            private Type processorType;
+
+            public RuntimeProcessorEnvironment(
+                Func<Type, ILogger> loggerFactory)
+            {
+                Debug.Assert(loggerFactory != null);
+
+                this.loggerFactory = loggerFactory;
+            }
+
+            public override ILogger CreateLogger(Type processorType)
+            {
+                Guard.NotNull(processorType, nameof(processorType));
+
+                lock (this.loggerLock)
+                {
+                    if (logger != null)
+                    {
+                        if (this.processorType != processorType)
+                        {
+                            throw new ArgumentException(
+                                $"{nameof(CreateLogger)} cannot be called with multiple types in a single instance.",
+                                nameof(processorType));
+                        }
+
+                        return this.logger;
+                    }
+
+                    this.processorType = processorType;
+                    this.logger = this.loggerFactory(processorType);
+                    return this.logger;
+                }
+            }
+        }
+    }
+}