Initial implementation of plugins

src/Temporalio/Client/ITemporalClientPlugin.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Threading.Tasks;
+
+namespace Temporalio.Client
+{
+    /// <summary>
+    /// Interface for temporal client plugins.
+    /// </summary>
+    /// <remarks>
+    /// WARNING: This API is experimental and may change in the future.
+    /// </remarks>
+    public interface ITemporalClientPlugin
+    {
+        /// <summary>
+        /// Gets the plugin name.
+        /// </summary>
+        string Name { get; }
+
+        /// <summary>
+        /// Configures the client options.
+        /// </summary>
+        /// <param name="options">The client options to configure.</param>
+        void ConfigureClient(TemporalClientOptions options);
+
+        /// <summary>
+        /// Handles temporal connection asynchronously.
+        /// </summary>
+        /// <param name="options">The connection options.</param>
+        /// <param name="continuation">The continuation function.</param>
+        /// <returns>A task representing the asynchronous operation.</returns>
+        Task<TemporalConnection> ConnectAsync(TemporalClientConnectOptions options, Func<TemporalClientConnectOptions, Task<TemporalConnection>> continuation);
+    }
+}

src/Temporalio/Client/TemporalClient.cs

@@ -21,6 +21,11 @@ public partial class TemporalClient : ITemporalClient
         /// <param name="options">Options for this client.</param>
         public TemporalClient(ITemporalConnection connection, TemporalClientOptions options)
         {
+            foreach (var plugin in options.Plugins ?? Enumerable.Empty<ITemporalClientPlugin>())
+            {
+                plugin.ConfigureClient(options);
+            }
+
             Connection = connection;
             Options = options;
             OutboundInterceptor = new Impl(this);
@@ -61,10 +66,21 @@ public TemporalClient(ITemporalConnection connection, TemporalClientOptions opti
         /// <param name="options">Options for connecting.</param>
         /// <returns>The connected client.</returns>
         public static async Task<TemporalClient> ConnectAsync(
-            TemporalClientConnectOptions options) =>
-            new(
-                await TemporalConnection.ConnectAsync(options).ConfigureAwait(false),
+            TemporalClientConnectOptions options)
+        {
+            Func<TemporalClientConnectOptions, Task<TemporalConnection>> connect = TemporalConnection.ConnectAsync;
+            if (options.Plugins != null)
+            {
+                foreach (var plugin in options.Plugins.Reverse())
+                {
+                    var localConnect = connect;
+                    connect = connectOptions => plugin.ConnectAsync(connectOptions, localConnect);
+                }
+            }
+            return new(
+                await connect(options).ConfigureAwait(false),
                 options.ToClientOptions());
+        }
 
         /// <summary>
         /// Create a client to a Temporal namespace that does not connect until first call.

src/Temporalio/Client/TemporalClientConnectOptions.cs

@@ -66,6 +66,14 @@ public TemporalClientConnectOptions(string targetHost)
         /// </summary>
         public QueryRejectCondition? QueryRejectCondition { get; set; }
 
+        /// <summary>
+        /// Gets or sets the plugins.
+        /// </summary>
+        /// <remarks>
+        /// WARNING: This API is experimental and may change in the future.
+        /// </remarks>
+        public IReadOnlyCollection<ITemporalClientPlugin>? Plugins { get; set; }
+
         /// <summary>
         /// Create client options from a subset of these options for use in
         /// <see cref="TemporalClient.TemporalClient" />.
@@ -79,6 +87,7 @@ public TemporalClientOptions ToClientOptions() =>
                 Interceptors = Interceptors,
                 LoggerFactory = LoggerFactory,
                 QueryRejectCondition = QueryRejectCondition,
+                Plugins = Plugins,
             };
     }
 }

src/Temporalio/Client/TemporalClientOptions.cs

@@ -43,6 +43,14 @@ public class TemporalClientOptions : ICloneable
         /// </summary>
         public QueryRejectCondition? QueryRejectCondition { get; set; }
 
+        /// <summary>
+        /// Gets or sets the plugins.
+        /// </summary>
+        /// <remarks>
+        /// WARNING: This API is experimental and may change in the future.
+        /// </remarks>
+        public IReadOnlyCollection<ITemporalClientPlugin>? Plugins { get; set; }
+
         /// <summary>
         /// Create a shallow copy of these options.
         /// </summary>

src/Temporalio/Common/SimplePlugin.cs

@@ -0,0 +1,240 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Temporalio.Client;
+using Temporalio.Worker;
+
+namespace Temporalio.Common
+{
+    /// <summary>
+    /// A simple plugin that implements both client and worker plugin interfaces.
+    /// </summary>
+    /// <remarks>
+    /// WARNING: This API is experimental and may change in the future.
+    /// </remarks>
+    public class SimplePlugin : ITemporalClientPlugin, ITemporalWorkerPlugin
+    {
+        /// <summary>
+        /// Initializes a new instance of the <see cref="SimplePlugin"/> class.
+        /// </summary>
+        /// <param name="name">The plugin name.</param>
+        /// <param name="options">The plugin options.</param>
+        public SimplePlugin(string name, SimplePluginOptions? options = null)
+        {
+            Name = name;
+            Options = options ?? new SimplePluginOptions();
+        }
+
+        /// <summary>
+        /// Gets the plugin options.
+        /// </summary>
+        public SimplePluginOptions Options { get; }
+
+        /// <summary>
+        /// Gets the plugin name.
+        /// </summary>
+        public string Name { get; }
+
+        /// <summary>
+        /// Configures the client options.
+        /// </summary>
+        /// <param name="options">The client options to configure.</param>
+        public virtual void ConfigureClient(TemporalClientOptions options)
+        {
+            options.DataConverter = Resolve(options.DataConverter, Options.DataConverterOption);
+            options.Interceptors = ResolveAppend(options.Interceptors, Options.ClientInterceptorsOption);
+        }
+
+        /// <summary>
+        /// Handles temporal connection asynchronously.
+        /// </summary>
+        /// <param name="options">The connection options.</param>
+        /// <param name="continuation">The continuation function.</param>
+        /// <returns>A task representing the asynchronous operation.</returns>
+        public virtual Task<TemporalConnection> ConnectAsync(
+            TemporalClientConnectOptions options,
+            Func<TemporalClientConnectOptions, Task<TemporalConnection>> continuation)
+        {
+            return continuation(options);
+        }
+
+        /// <summary>
+        /// Configures the worker options.
+        /// </summary>
+        /// <param name="options">The worker options to configure.</param>
+        public virtual void ConfigureWorker(TemporalWorkerOptions options)
+        {
+            DoAppend(options.Activities, Options.Activities);
+            DoAppend(options.Workflows, Options.Workflows);
+            DoAppend(options.NexusServices, Options.NexusServices);
+            options.Interceptors = ResolveAppend(options.Interceptors, Options.WorkerInterceptorsOption);
+            options.WorkflowFailureExceptionTypes = ResolveAppend(
+                options.WorkflowFailureExceptionTypes, Options.WorkflowFailureExceptionTypesOption);
+        }
+
+        /// <summary>
+        /// Runs the worker asynchronously.
+        /// </summary>
+        /// <typeparam name="TResult">Result type. For most worker run calls, this is
+        /// <see cref="ValueTuple"/>.</typeparam>
+        /// <param name="worker">The worker to run.</param>
+        /// <param name="continuation">The continuation function.</param>
+        /// <param name="stoppingToken">Cancellation token to stop the worker.</param>
+        /// <returns>A task representing the asynchronous operation.</returns>
+        public virtual async Task<TResult> RunWorkerAsync<TResult>(
+            TemporalWorker worker,
+            Func<TemporalWorker, CancellationToken, Task<TResult>> continuation,
+            CancellationToken stoppingToken)
+        {
+            if (Options.RunContextBefore is { } before)
+            {
+                await before().ConfigureAwait(false);
+            }
+            try
+            {
+                return await continuation(worker, stoppingToken).ConfigureAwait(false);
+            }
+            finally
+            {
+                if (Options.RunContextAfter is { } after)
+                {
+                    await after().ConfigureAwait(false);
+                }
+            }
+        }
+
+        /// <summary>
+        /// Configures the replayer options.
+        /// </summary>
+        /// <param name="options">The replayer options to configure.</param>
+        public virtual void ConfigureReplayer(WorkflowReplayerOptions options)
+        {
+            options.DataConverter = Resolve(options.DataConverter, Options.DataConverterOption);
+            DoAppend(options.Workflows, Options.Workflows);
+            options.Interceptors = ResolveAppend(options.Interceptors, Options.WorkerInterceptorsOption);
+            options.WorkflowFailureExceptionTypes = ResolveAppend(
+                options.WorkflowFailureExceptionTypes, Options.WorkflowFailureExceptionTypesOption);
+        }
+
+        /// <summary>
+        /// Runs the replayer asynchronously.
+        /// </summary>
+        /// <param name="replayer">The replayer to run.</param>
+        /// <param name="continuation">The continuation function.</param>
+        /// <param name="cancellationToken">Cancellation token to stop the replay.</param>
+        /// <returns>A task representing the asynchronous operation.</returns>
+        public virtual async Task<IEnumerable<WorkflowReplayResult>> ReplayWorkflowsAsync(
+            WorkflowReplayer replayer,
+            Func<WorkflowReplayer, CancellationToken, Task<IEnumerable<WorkflowReplayResult>>> continuation,
+            CancellationToken cancellationToken)
+        {
+            if (Options.RunContextBefore is { } before)
+            {
+                await before().ConfigureAwait(false);
+            }
+            try
+            {
+                return await continuation(replayer, cancellationToken).ConfigureAwait(false);
+            }
+            finally
+            {
+                if (Options.RunContextAfter is { } after)
+                {
+                    await after().ConfigureAwait(false);
+                }
+            }
+        }
+
+#if NETCOREAPP3_0_OR_GREATER
+        /// <summary>
+        /// Runs the replayer asynchronously.
+        /// </summary>
+        /// <param name="replayer">The replayer to run.</param>
+        /// <param name="continuation">The continuation function.</param>
+        /// <param name="cancellationToken">Cancellation token to stop the replay.</param>
+        /// <returns>A task representing the asynchronous operation.</returns>
+        public virtual async IAsyncEnumerable<WorkflowReplayResult> ReplayWorkflowsAsync(
+            WorkflowReplayer replayer,
+            Func<WorkflowReplayer, IAsyncEnumerable<WorkflowReplayResult>> continuation,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            if (Options.RunContextBefore is { } before)
+            {
+                await before().ConfigureAwait(false);
+            }
+            try
+            {
+                var asyncEnum = continuation(replayer);
+                await foreach (var res in asyncEnum.ConfigureAwait(false).WithCancellation(cancellationToken))
+                {
+                    yield return res;
+                }
+            }
+            finally
+            {
+                if (Options.RunContextAfter is { } after)
+                {
+                    await after().ConfigureAwait(false);
+                }
+            }
+        }
+#endif
+
+        private static T Resolve<T>(T existing, SimplePluginOptions.SimplePluginOption<T>? parameter)
+        {
+            if (parameter == null)
+            {
+                return existing;
+            }
+            var option = parameter.Constant ?? existing;
+            if (parameter.Configurable != null)
+            {
+                return parameter.Configurable(option);
+            }
+            return option;
+        }
+
+        private static IReadOnlyCollection<T>? ResolveAppend<T>(
+            IReadOnlyCollection<T>? existing,
+            SimplePluginOptions.SimplePluginOption<IReadOnlyCollection<T>?>? parameter)
+        {
+            if (parameter == null)
+            {
+                return existing;
+            }
+
+            var option = existing;
+            if (existing != null && parameter.Constant != null)
+            {
+                option = existing.Concat(parameter.Constant).ToList();
+            }
+            else if (parameter.Constant != null)
+            {
+                option = parameter.Constant;
+            }
+            if (parameter.Configurable != null)
+            {
+                return parameter.Configurable(option);
+            }
+            return option;
+        }
+
+        private static void DoAppend<T>(
+            IList<T> existing,
+            IList<T>? parameter)
+        {
+            if (parameter == null)
+            {
+                return;
+            }
+
+            foreach (var item in parameter)
+            {
+                existing.Add(item);
+            }
+        }
+    }
+}