Persistent stream provider to lifecycle

src/AWS/Orleans.Streaming.SQS/Hosting/ClientBuilderExtensions.cs

@@ -0,0 +1,46 @@
+
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using Orleans.Configuration;
+using Orleans.Hosting;
+using OrleansAWSUtils.Streams;
+
+namespace Orleans.Hosting
+{
+    public static class ClientBuilderExtensions
+    {
+        /// <summary>
+        /// Configure cluster client to use SQS persistent streams.
+        /// </summary>
+        public static IClientBuilder AddSqsStreams(this IClientBuilder builder, string name, Action<SqsStreamOptions> configureOptions)
+        {
+            return builder.ConfigureServices(services => services.AddClusterClientSqsStreams(name, configureOptions));
+        }
+
+        /// <summary>
+        /// Configure cluster client to use SQS persistent streams.
+        /// </summary>
+        public static IClientBuilder AddSqsStreams(this IClientBuilder builder, string name, Action<OptionsBuilder<SqsStreamOptions>> configureOptions = null)
+        {
+            return builder.ConfigureServices(services => services.AddClusterClientSqsStreams(name, configureOptions));
+        }
+
+        /// <summary>
+        /// Configure cluster client to use SQS persistent streams.
+        /// </summary>
+        public static IServiceCollection AddClusterClientSqsStreams(this IServiceCollection services, string name, Action<SqsStreamOptions> configureOptions)
+        {
+            return services.AddClusterClientSqsStreams(name, ob => ob.Configure(configureOptions));
+        }
+
+        /// <summary>
+        /// Configure cluster client to use SQS persistent streams.
+        /// </summary>
+        public static IServiceCollection AddClusterClientSqsStreams(this IServiceCollection services, string name,
+            Action<OptionsBuilder<SqsStreamOptions>> configureOptions = null)
+        {
+            return services.ConfigureNamedOptionForLogging<SqsStreamOptions>(name)
+                           .AddClusterClientPersistentStreams<SqsStreamOptions>(name, SQSAdapterFactory.Create, configureOptions);
+        }
+    }
+}

src/AWS/Orleans.Streaming.SQS/Hosting/SiloBuilderExtensions.cs

@@ -0,0 +1,45 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using Orleans.Configuration;
+using Orleans.Hosting;
+using OrleansAWSUtils.Streams;
+
+namespace Orleans.Hosting
+{
+    public static class SiloBuilderExtensions
+    {
+        /// <summary>
+        /// Configure silo to use SQS persistent streams.
+        /// </summary>
+        public static ISiloHostBuilder AddSqsStreams(this ISiloHostBuilder builder, string name, Action<SqsStreamOptions> configureOptions)
+        {
+            return builder.ConfigureServices(services => services.AddSiloSqsStreams(name, configureOptions));
+        }
+
+        /// <summary>
+        /// Configure silo to use SQS persistent streams.
+        /// </summary>
+        public static ISiloHostBuilder AddSqsStreams(this ISiloHostBuilder builder, string name, Action<OptionsBuilder<SqsStreamOptions>> configureOptions = null)
+        {
+            return builder.ConfigureServices(services => services.AddSiloSqsStreams(name, configureOptions));
+        }
+
+        /// <summary>
+        /// Configure silo to use SQS persistent streams.
+        /// </summary>
+        public static IServiceCollection AddSiloSqsStreams(this IServiceCollection services, string name, Action<SqsStreamOptions> configureOptions)
+        {
+            return services.AddSiloSqsStreams(name, ob => ob.Configure(configureOptions));
+        }
+
+        /// <summary>
+        /// Configure silo to use SQS persistent streams.
+        /// </summary>
+        public static IServiceCollection AddSiloSqsStreams(this IServiceCollection services, string name,
+            Action<OptionsBuilder<SqsStreamOptions>> configureOptions = null)
+        {
+            return services.ConfigureNamedOptionForLogging<SqsStreamOptions>(name)
+                           .AddSiloPersistentStreams<SqsStreamOptions>(name, SQSAdapterFactory.Create, configureOptions);
+        }
+    }
+}

src/AWS/Orleans.Streaming.SQS/Orleans.Streaming.SQS.csproj

@@ -22,6 +22,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\..\Orleans.Core\Orleans.Core.csproj" />
+    <ProjectReference Include="..\..\Orleans.Runtime.Abstractions\Orleans.Runtime.Abstractions.csproj" />
     <ProjectReference Include="..\..\OrleansProviders\OrleansProviders.csproj" />
   </ItemGroup>
 </Project>

src/AWS/Orleans.Streaming.SQS/Streams/SQSAdapterFactory.cs

@@ -1,60 +1,46 @@
-using Orleans.Providers;
-using Orleans.Providers.Streams.Common;
-using Orleans.Runtime;
-using Orleans.Streams;
-using System;
+using System;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Orleans.Providers.Streams.Common;
+using Orleans.Streams;
 using Orleans.Serialization;
+using Orleans.Configuration;
 
 namespace OrleansAWSUtils.Streams
 {
     /// <summary> Factory class for Azure Queue based stream provider.</summary>
     public class SQSAdapterFactory : IQueueAdapterFactory
     {
-        internal const int CacheSizeDefaultValue = 4096;
-        internal const string NumQueuesPropertyName = "NumQueues";
-
-        /// <summary> Default number of Azure Queue used in this stream provider.</summary>
-        public const int NumQueuesDefaultValue = 8; // keep as power of 2.
-
-        private string deploymentId;
-        private string dataConnectionString;
-        private string providerName;
-        private int cacheSize;
-        private int numQueues;
+        private readonly string providerName;
+        private readonly SqsStreamOptions options;
+        private readonly SiloOptions siloOptions;
+        private readonly SerializationManager serializationManager;
+        private readonly ILoggerFactory loggerFactory;
         private HashRingBasedStreamQueueMapper streamQueueMapper;
         private IQueueAdapterCache adapterCache;
-        private SerializationManager serializationManager;
-        private ILoggerFactory loggerFactory;
-        /// <summary>"DataConnectionString".</summary>
-        public const string DataConnectionStringPropertyName = "DataConnectionString";
-        /// <summary>"DeploymentId".</summary>
-        public const string DeploymentIdPropertyName = "DeploymentId";
 
         /// <summary>
         /// Application level failure handler override.
         /// </summary>
         protected Func<QueueId, Task<IStreamFailureHandler>> StreamFailureHandlerFactory { private get; set; }
 
-        /// <summary> Init the factory.</summary>
-        public virtual void Init(IProviderConfiguration config, string providerName, IServiceProvider serviceProvider)
+        public SQSAdapterFactory(string name, SqsStreamOptions options, IServiceProvider serviceProvider, IOptions<SiloOptions> siloOptions, SerializationManager serializationManager, ILoggerFactory loggerFactory)
         {
-            if (config == null) throw new ArgumentNullException("config");
-            if (!config.Properties.TryGetValue(DataConnectionStringPropertyName, out dataConnectionString))
-                throw new ArgumentException(string.Format("{0} property not set", DataConnectionStringPropertyName));
-            if (!config.Properties.TryGetValue(DeploymentIdPropertyName, out deploymentId))
-                throw new ArgumentException(string.Format("{0} property not set", DeploymentIdPropertyName));
+            this.providerName = name;
+            this.options = options;
+            this.siloOptions = siloOptions.Value;
+            this.serializationManager = serializationManager;
+            this.loggerFactory = loggerFactory;
+        }
 
-            cacheSize = SimpleQueueAdapterCache.ParseSize(config, CacheSizeDefaultValue);
-            this.loggerFactory = serviceProvider.GetRequiredService<ILoggerFactory>();
-            numQueues = config.GetIntProperty(NumQueuesPropertyName, NumQueuesDefaultValue);
 
-            this.providerName = providerName;
-            streamQueueMapper = new HashRingBasedStreamQueueMapper(numQueues, providerName);
-            adapterCache = new SimpleQueueAdapterCache(cacheSize, providerName, serviceProvider.GetRequiredService<ILoggerFactory>());
-            this.serializationManager = serviceProvider.GetRequiredService<SerializationManager>();
+        /// <summary> Init the factory.</summary>
+        public virtual void Init()
+        {
+            streamQueueMapper = new HashRingBasedStreamQueueMapper(this.options.NumQueues, this.providerName);
+            adapterCache = new SimpleQueueAdapterCache(this.options.CacheSize, this.providerName, this.loggerFactory);
             if (StreamFailureHandlerFactory == null)
             {
                 StreamFailureHandlerFactory =
@@ -65,7 +51,7 @@ public virtual void Init(IProviderConfiguration config, string providerName, ISe
         /// <summary>Creates the Azure Queue based adapter.</summary>
         public virtual Task<IQueueAdapter> CreateAdapter()
         {
-            var adapter = new SQSAdapter(this.serializationManager, streamQueueMapper, this.loggerFactory, dataConnectionString, deploymentId, providerName);
+            var adapter = new SQSAdapter(this.serializationManager, this.streamQueueMapper, this.loggerFactory, this.options.ConnectionString, this.options.ClusterId ?? this.siloOptions.ClusterId, this.providerName);
             return Task.FromResult<IQueueAdapter>(adapter);
         }
 
@@ -90,5 +76,13 @@ public Task<IStreamFailureHandler> GetDeliveryFailureHandler(QueueId queueId)
         {
             return StreamFailureHandlerFactory(queueId);
         }
+
+        public static SQSAdapterFactory Create(IServiceProvider services, string name)
+        {
+            IOptionsSnapshot<SqsStreamOptions> streamOptionsSnapshot = services.GetRequiredService<IOptionsSnapshot<SqsStreamOptions>>();
+            var factory = ActivatorUtilities.CreateInstance<SQSAdapterFactory>(services, name, streamOptionsSnapshot.Get(name));
+            factory.Init();
+            return factory;
+        }
     }
 }

src/AWS/Orleans.Streaming.SQS/Streams/SQSStreamProviderUtils.cs

@@ -1,9 +1,10 @@
-using Orleans.Streams;
-using OrleansAWSUtils.Storage;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
+using Orleans.Streams;
+using OrleansAWSUtils.Storage;
+using Orleans.Configuration;
 
 namespace OrleansAWSUtils.Streams
 {
@@ -20,7 +21,7 @@ public static async Task DeleteAllUsedQueues(string providerName, string cluster
         {
             if (clusterId != null)
             {
-                var queueMapper = new HashRingBasedStreamQueueMapper(SQSAdapterFactory.NumQueuesDefaultValue, providerName);
+                var queueMapper = new HashRingBasedStreamQueueMapper(SqsStreamOptions.NumQueuesDefaultValue, providerName);
                 List<QueueId> allQueues = queueMapper.GetAllQueues().ToList();
 
                 var deleteTasks = new List<Task>();

src/AWS/Orleans.Streaming.SQS/Streams/SqsStreamOptions.cs

@@ -0,0 +1,17 @@
+
+namespace Orleans.Configuration
+{
+    public class SqsStreamOptions : PersistentStreamOptions
+    {
+        public string ClusterId { get; set; }
+
+        [Redact]
+        public string ConnectionString { get; set; }
+
+        public int CacheSize { get; set; } = CacheSizeDefaultValue;
+        public const int CacheSizeDefaultValue = 4096;
+
+        public int NumQueues { get; set; } = NumQueuesDefaultValue;
+        public const int NumQueuesDefaultValue = 8; // keep as power of 2.
+    }
+}

src/AdoNet/Orleans.Persistence.AdoNet/Storage/Provider/SiloBuilderExtensions.cs

@@ -1,6 +1,4 @@
 using System;
-using System.Collections.Generic;
-using System.Text;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Options;

src/Azure/Orleans.Persistence.AzureStorage/Providers/Storage/AzureBlobStorage.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Generic;
 using System.Diagnostics;
 using System.Text;
 using System.Threading;
@@ -10,8 +9,7 @@
 using Microsoft.WindowsAzure.Storage;
 using Microsoft.WindowsAzure.Storage.Blob;
 using Newtonsoft.Json;
-using Orleans.Hosting;
-using Orleans.Providers;
+using Orleans.Configuration;
 using Orleans.Providers.Azure;
 using Orleans.Runtime;
 using Orleans.Runtime.Configuration;

src/Azure/Orleans.Persistence.AzureStorage/Providers/Storage/AzureBlobStorageOptions.cs

@@ -1,13 +1,13 @@
-using Microsoft.Extensions.Options;
+using System;
+using System.Collections.Generic;
+using Microsoft.Extensions.Options;
 using Microsoft.WindowsAzure.Storage;
 using Newtonsoft.Json;
 using Orleans.Persistence.AzureStorage;
 using Orleans.Runtime;
 using Orleans.Runtime.Configuration;
-using System;
-using System.Collections.Generic;
 
-namespace Orleans.Hosting
+namespace Orleans.Configuration
 {
     public class AzureBlobStorageOptions
     {

src/Azure/Orleans.Streaming.AzureStorage/Hosting/ClientBuilderExtensions.cs

@@ -0,0 +1,49 @@
+
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using Orleans.Configuration;
+using Orleans.Providers.Streams.AzureQueue;
+
+namespace Orleans.Hosting
+{
+    public static class ClientBuilderExtensions
+    {
+        /// <summary>
+        /// Configure cluster client to use azure queue persistent streams.
+        /// </summary>
+        public static IClientBuilder AddAzureQueueStreams<TDataAdapter>(this IClientBuilder builder, string name, Action<AzureQueueStreamOptions> configureOptions)
+            where TDataAdapter : IAzureQueueDataAdapter
+        {
+            return builder.ConfigureServices(services => services.AddClusterClientAzureQueueStreams<TDataAdapter>(name, configureOptions));
+        }
+
+        /// <summary>
+        /// Configure cluster client to use azure queue persistent streams.
+        /// </summary>
+        public static IClientBuilder AddAzureQueueStreams<TDataAdapter>(this IClientBuilder builder, string name, Action<OptionsBuilder<AzureQueueStreamOptions>> configureOptions = null)
+            where TDataAdapter : IAzureQueueDataAdapter
+        {
+            return builder.ConfigureServices(services => services.AddClusterClientAzureQueueStreams<TDataAdapter>(name, configureOptions));
+        }
+
+        /// <summary>
+        /// Configure cluster client to use azure queue persistent streams.
+        /// </summary>
+        public static IServiceCollection AddClusterClientAzureQueueStreams<TDataAdapter>(this IServiceCollection services, string name, Action<AzureQueueStreamOptions> configureOptions)
+            where TDataAdapter : IAzureQueueDataAdapter
+        {
+            return services.AddClusterClientAzureQueueStreams<TDataAdapter>(name, ob => ob.Configure(configureOptions));
+        }
+
+        /// <summary>
+        /// Configure cluster client to use azure queue persistent streams.
+        /// </summary>
+        public static IServiceCollection AddClusterClientAzureQueueStreams<TDataAdapter>(this IServiceCollection services, string name,
+            Action<OptionsBuilder<AzureQueueStreamOptions>> configureOptions = null)
+            where TDataAdapter : IAzureQueueDataAdapter
+        {
+            return services.ConfigureNamedOptionForLogging<AzureQueueStreamOptions>(name)
+                           .AddClusterClientPersistentStreams<AzureQueueStreamOptions>(name, AzureQueueAdapterFactory<TDataAdapter>.Create, configureOptions);
+        }
+    }
+}