using System; using System.Collections.Generic; using Elsa.Builders; using Elsa.Models; using Elsa.Persistence; using Elsa.Persistence.InMemory; using Elsa.Providers.WorkflowStorage; using Elsa.Serialization; using Elsa.Services; using Elsa.Services.Dispatch; using Elsa.Services.Messaging; using Elsa.Services.Startup; using Humanizer; using Microsoft.Extensions.DependencyInjection; using Newtonsoft.Json; using NodaTime; using Rebus.Persistence.InMem; using Rebus.Routing.TypeBased; using Rebus.Transport.InMem; using Storage.Net; using Storage.Net.Blobs; namespace Elsa { public record CompetingMessageType(Type MessageType, string? Queue = default); public class ElsaOptions { public static string FormatChannelQueueName(string channel) => FormatChannelQueueName(typeof(TMessage), channel); public static string FormatChannelQueueName(Type messageType, string channel) { var queue = !string.IsNullOrWhiteSpace(channel) ? $"{messageType.Name}{channel}" : messageType.Name; return FormatQueueName(queue); } public static string FormatQueueName(string queue) => queue.Dehumanize().Underscore().Dasherize(); internal ElsaOptions() { WorkflowDefinitionStoreFactory = sp => ActivatorUtilities.CreateInstance(sp); WorkflowInstanceStoreFactory = sp => ActivatorUtilities.CreateInstance(sp); WorkflowExecutionLogStoreFactory = sp => ActivatorUtilities.CreateInstance(sp); WorkflowTriggerStoreFactory = sp => ActivatorUtilities.CreateInstance(sp); WorkflowDefinitionDispatcherFactory = sp => ActivatorUtilities.CreateInstance(sp); WorkflowInstanceDispatcherFactory = sp => ActivatorUtilities.CreateInstance(sp); CorrelatingWorkflowDispatcherFactory = sp => ActivatorUtilities.CreateInstance(sp); StorageFactory = sp => Storage.Net.StorageFactory.Blobs.InMemory(); JsonSerializerConfigurer = (sp, serializer) => { }; DefaultWorkflowStorageProviderType = typeof(WorkflowInstanceWorkflowStorageProvider); DistributedLockingOptions = new DistributedLockingOptions(); ConfigureServiceBusEndpoint = ConfigureInMemoryServiceBusEndpoint; CreateJsonSerializer = sp => { var serializer = DefaultContentSerializer.CreateDefaultJsonSerializer(); JsonSerializerConfigurer(sp, serializer); return serializer; }; } public string ContainerName { get; set; } = Environment.MachineName; public ServiceFactory ActivityFactory { get; } = new(); public ServiceFactory WorkflowFactory { get; } = new(); public IEnumerable ActivityTypes => ActivityFactory.Types; public IList WorkflowTypes { get; } = new List(); public IList CompetingMessageTypes { get; } = new List(); public IList PubSubMessageTypes { get; } = new List(); public ServiceBusOptions ServiceBusOptions { get; } = new(); public DistributedLockingOptions DistributedLockingOptions { get; set; } /// /// The amount of time to wait before giving up on trying to acquire a lock. /// public Duration DistributedLockTimeout { get; set; } = Duration.FromHours(1); public Type DefaultWorkflowStorageProviderType { get; set; } public WorkflowChannelOptions WorkflowChannelOptions { get; set; } = new(); internal Func StorageFactory { get; set; } internal Func WorkflowDefinitionStoreFactory { get; set; } internal Func WorkflowInstanceStoreFactory { get; set; } internal Func WorkflowExecutionLogStoreFactory { get; set; } internal Func WorkflowTriggerStoreFactory { get; set; } internal Func CreateJsonSerializer { get; set; } internal Action JsonSerializerConfigurer { get; set; } internal Func WorkflowDefinitionDispatcherFactory { get; set; } internal Func WorkflowInstanceDispatcherFactory { get; set; } internal Func CorrelatingWorkflowDispatcherFactory { get; set; } internal Action ConfigureServiceBusEndpoint { get; set; } internal ICollection Startups { get; } = new List(); private static void ConfigureInMemoryServiceBusEndpoint(ServiceBusEndpointConfigurationContext context) { var serviceProvider = context.ServiceProvider; var transport = serviceProvider.GetService(); var store = serviceProvider.GetRequiredService(); var queueName = context.QueueName; context.Configurer .Subscriptions(s => s.StoreInMemory(store)) .Transport(t => t.UseInMemoryTransport(transport, queueName)); } } }