SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Simplify service bus API and cleanup

Browse source
This commit is contained in:
Sipke Schoorstra 2020-11-22 11:49:28 +01:00
parent f1349adc3f
commit e6227e887f
20 changed files with 103 additions and 143 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
src/activities/Elsa.Activities.Rebus/Activities/MessageReceived.cs
View file

@ -3,7 +3,6 @@ using Elsa.ActivityResults;
using Elsa.Attributes;
using Elsa.Services;
using Elsa.Services.Models;
using Rebus.Bus;
namespace Elsa.Activities.Rebus
{

1
src/activities/Elsa.Activities.Rebus/Activities/PublishMessage.cs
View file

@ -5,7 +5,6 @@ using Elsa.ActivityResults;
using Elsa.Attributes;
using Elsa.Services;
using Elsa.Services.Models;
using Rebus.Bus;
namespace Elsa.Activities.Rebus
{

1
src/activities/Elsa.Activities.Rebus/Activities/SendMessage.cs
View file

@ -5,7 +5,6 @@ using Elsa.ActivityResults;
using Elsa.Attributes;
using Elsa.Services;
using Elsa.Services.Models;
using Rebus.Bus;
namespace Elsa.Activities.Rebus
{

2
src/activities/Elsa.Activities.Rebus/Consumers/MessageConsumer.cs
View file

@ -8,7 +8,7 @@ using Rebus.Pipeline;
namespace Elsa.Activities.Rebus.Consumers
{
public class MessageConsumer<T> : IHandleMessages<T>
public class MessageConsumer<T> : IHandleMessages<T> where T:notnull
{
private readonly IWorkflowScheduler _workflowScheduler;

13
src/activities/Elsa.Activities.Rebus/Extensions/ServiceCollectionExtensions.cs
View file

@ -1,6 +1,8 @@
using System;
using Elsa.Activities.Rebus.Consumers;
using Elsa.Activities.Rebus.StartupTasks;
using Elsa.Activities.Rebus.Triggers;
using Elsa.Runtime;
using Microsoft.Extensions.DependencyInjection;
using Rebus.Handlers;
@ -19,19 +21,14 @@ namespace Elsa.Activities.Rebus.Extensions
return services
.AddTriggerProvider<MessageReceivedTriggerProvider>()
.AddStartupTask(sp => ActivatorUtilities.CreateInstance<CreateSubscriptions>(sp, (object)messageTypes))
.AddActivity<PublishMessage>()
.AddActivity<SendMessage>()
.AddActivity<MessageReceived>();
}
public static IServiceCollection AddRebusActivities<T>(this IServiceCollection services) => services.AddRebusActivities(typeof(T));
// public static IServiceCollection AddRebusActivities<T1, T2>(this IServiceCollection services) => services.AddRebusActivities().AddMessageType<T1>().AddMessageType<T2>();
// public static IServiceCollection AddRebusActivities<T1, T2, T3>(this IServiceCollection services) => services.AddRebusActivities().AddMessageType<T1>().AddMessageType<T2>().AddMessageType<T3>();
// public static IServiceCollection AddRebusActivities<T1, T2, T3, T4>(this IServiceCollection services) => services.AddRebusActivities().AddMessageType<T1>().AddMessageType<T2>().AddMessageType<T3>().AddMessageType<T4>();
//
// public static IServiceCollection AddRebusActivities<T1, T2, T3, T4, T5>(this IServiceCollection services) =>
// services.AddRebusActivities().AddMessageType<T1>().AddMessageType<T2>().AddMessageType<T3>().AddMessageType<T4>().AddMessageType<T5>();
//public static IServiceCollection AddMessageType<T>(this IServiceCollection services) => services.AddTransient<IHandleMessages<T>, MessageConsumer<T>>();
public static IServiceCollection AddRebusActivities<T1, T2>(this IServiceCollection services) => services.AddRebusActivities(typeof(T1), typeof(T2));
public static IServiceCollection AddRebusActivities<T1, T2, T3>(this IServiceCollection services) => services.AddRebusActivities(typeof(T1), typeof(T2), typeof(T3));
}
}

29
src/activities/Elsa.Activities.Rebus/StartupTasks/CreateSubscriptions.cs Normal file
View file

@ -0,0 +1,29 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Services;
namespace Elsa.Activities.Rebus.StartupTasks
{
public class CreateSubscriptions : IStartupTask
{
private readonly IServiceBusFactory _serviceBusFactory;
private readonly IEnumerable<Type> _messageTypes;
public CreateSubscriptions(IServiceBusFactory serviceBusFactory, IEnumerable<Type> messageTypes)
{
_serviceBusFactory = serviceBusFactory;
_messageTypes = messageTypes;
}
public async Task ExecuteAsync(CancellationToken cancellationToken = default)
{
foreach (var messageType in _messageTypes)
{
var bus = _serviceBusFactory.GetServiceBus(messageType);
await bus.Subscribe(messageType);
}
}
}
}

4
src/core/Elsa.Abstractions/Services/IServiceBusFactory.cs
View file

@ -1,12 +1,10 @@
using System;
using Rebus.Bus;
using Rebus.Config;
namespace Elsa.Services
{
public interface IServiceBusFactory
{
IBus CreateServiceBus(string name, Func<RebusConfigurer, IServiceProvider, RebusConfigurer> configure);
IBus GetServiceBus(string name);
IBus GetServiceBus(Type messageType);
}
}

43
src/core/Elsa.Core/ElsaOptions.cs
View file

@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.Data;
using Elsa.Caching;
using Elsa.DistributedLock;
@ -7,11 +6,9 @@ using Elsa.Extensions;
using Elsa.Serialization;
using Elsa.Services;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Newtonsoft.Json;
using Rebus.Bus;
using Rebus.Config;
using Rebus.DataBus.InMem;
using Rebus.Logging;
using Rebus.Persistence.InMem;
using Rebus.Routing.TypeBased;
using Rebus.Transport.InMem;
@ -44,9 +41,7 @@ namespace Elsa
services.AddSingleton<InMemorySubscriberStore>();
services.AddSingleton<InMemDataStore>();
CreateServiceBus = (name, configure, sp) => sp.GetRequiredService<IServiceBusFactory>().CreateServiceBus(name, configure);
ConfigureServiceBusEndpoint = ConfigureInMemoryServiceBusEndpoint;
ConfigureServiceBusPublishEndpoint = ConfigureInMemoryServiceBusPublishEndpoint;
CreateJsonSerializer = sp =>
{
@ -57,7 +52,6 @@ namespace Elsa
}
public IServiceCollection Services { get; }
public ICollection<Type> MessageTypes { get; } = new List<Type>();
internal Action<IServiceProvider, IConfiguration> ConfigurePersistence { get; set; }
internal Func<IServiceProvider, IBlobStorage> StorageFactory { get; set; }
internal Func<IServiceProvider, IDistributedLockProvider> DistributedLockProviderFactory { get; private set; }
@ -65,9 +59,7 @@ namespace Elsa
internal Func<IServiceProvider, JsonSerializer> CreateJsonSerializer { get; private set; }
internal Action<IServiceProvider, JsonSerializer> JsonSerializerConfigurer { get; private set; }
internal Action AddAutoMapper { get; private set; }
internal Func<string, Func<RebusConfigurer, IServiceProvider, RebusConfigurer>, IServiceProvider, IBus> CreateServiceBus { get; private set; }
internal Action<ServiceBusEndpointConfigurationContext> ConfigureServiceBusEndpoint { get; private set; }
internal Action<ServiceBusPublishEndpointConfigurationContext> ConfigureServiceBusPublishEndpoint { get; private set; }
public ElsaOptions UseDistributedLockProvider(Func<IServiceProvider, IDistributedLockProvider> factory)
{
@ -115,26 +107,12 @@ namespace Elsa
return this;
}
public ElsaOptions UseServiceBusFactory(Func<string, Func<RebusConfigurer, IServiceProvider, RebusConfigurer>, IServiceProvider, IBus> serviceBusFactory)
{
CreateServiceBus = serviceBusFactory;
return this;
}
public ElsaOptions SetupServiceBusEndpoint(Action<ServiceBusEndpointConfigurationContext> setup)
public ElsaOptions SetupServiceBus(Action<ServiceBusEndpointConfigurationContext> setup)
{
ConfigureServiceBusEndpoint = setup;
return this;
}
public ElsaOptions AddEndpoint<T>() => AddEndpoint(typeof(T));
public ElsaOptions AddEndpoint(Type messageType)
{
MessageTypes.Add(messageType);
return this;
}
private static void ConfigureInMemoryServiceBusEndpoint(ServiceBusEndpointConfigurationContext context)
{
var serviceProvider = context.ServiceProvider;
@ -143,23 +121,10 @@ namespace Elsa
var queueName = context.QueueName;
context.Configurer
.Logging(l => l.ColoredConsole())
.Logging(l => l.ColoredConsole(LogLevel.Info))
.Subscriptions(s => s.StoreInMemory(store))
.Transport(t => t.UseInMemoryTransport(transport, queueName))
.Routing(r => r.TypeBased().Map(context.MessageType, queueName));
}
private static void ConfigureInMemoryServiceBusPublishEndpoint(ServiceBusPublishEndpointConfigurationContext context)
{
var serviceProvider = context.ServiceProvider;
var transport = serviceProvider.GetService<InMemNetwork>();
var store = serviceProvider.GetRequiredService<InMemorySubscriberStore>();
var queueName = context.QueueName;
context.Configurer
.Logging(l => l.ColoredConsole())
.Subscriptions(s => s.StoreInMemory(store))
.Transport(t => t.UseInMemoryTransport(transport, queueName));
.Routing(r => r.TypeBased().Map(context.MessageTypeMap));
}
}
}

40
src/core/Elsa.Core/Extensions/ElsaServiceCollectionExtensions.cs
View file

@ -19,14 +19,12 @@ using Elsa.Metadata.Handlers;
using Elsa.Runtime;
using Elsa.Serialization;
using Elsa.Services;
using Elsa.StartupTasks;
using Elsa.Triggers;
using Elsa.WorkflowProviders;
using MediatR;
using Microsoft.Extensions.DependencyInjection.Extensions;
using NodaTime;
using Rebus.ServiceProvider;
using RunWorkflow = Elsa.Messages.RunWorkflow;
// ReSharper disable once CheckNamespace
namespace Microsoft.Extensions.DependencyInjection
@ -50,7 +48,6 @@ namespace Microsoft.Extensions.DependencyInjection
options.AddWorkflowsCore();
options.AddMediatR();
options.AddAutoMapper();
options.AddEndpoint<RunWorkflow>();
return services;
}
@ -121,11 +118,10 @@ namespace Microsoft.Extensions.DependencyInjection
.AddAutoMapperProfile<CloningProfile>()
.AddSingleton<ICloner, AutoMapperCloner>()
.AddNotificationHandlers(typeof(ElsaServiceCollectionExtensions))
.AddStartupTask<StartServiceBusTask>()
.AddSingleton<ServiceBusFactory>()
.AddSingleton<IServiceBusFactory, ServiceBusFactory>()
.AddSingleton<ICommandSender, CommandSender>(CreateCommandSender)
.AddSingleton<IEventPublisher, EventPublisher>(CreateEventPublisher)
.AddSingleton<ICommandSender, CommandSender>()
.AddSingleton<IEventPublisher, EventPublisher>()
.AutoRegisterHandlersFromAssemblyOf<RunWorkflowConsumer>()
.AddMetadataHandlers()
.AddCoreActivities();
@ -133,38 +129,6 @@ namespace Microsoft.Extensions.DependencyInjection
return configuration;
}
private static CommandSender CreateCommandSender(IServiceProvider serviceProvider)
{
var options = serviceProvider.GetRequiredService<ElsaOptions>();
var messageTypes = options.MessageTypes;
foreach (var messageType in messageTypes)
{
options.CreateServiceBus(messageType.Name, (bus, sp) =>
{
var context = new ServiceBusEndpointConfigurationContext(bus, messageType.Name, messageType, sp);
options.ConfigureServiceBusEndpoint(context);
return bus;
}, serviceProvider);
}
return ActivatorUtilities.CreateInstance<CommandSender>(serviceProvider);
}
private static EventPublisher CreateEventPublisher(IServiceProvider serviceProvider)
{
var options = serviceProvider.GetRequiredService<ElsaOptions>();
options.CreateServiceBus("elsa:publisher", (bus, sp) =>
{
var context = new ServiceBusPublishEndpointConfigurationContext(bus, "elsa:publisher", sp);
options.ConfigureServiceBusPublishEndpoint(context);
return bus;
}, serviceProvider);
return ActivatorUtilities.CreateInstance<EventPublisher>(serviceProvider);
}
private static IServiceCollection AddMetadataHandlers(this IServiceCollection services) =>
services
.AddSingleton<IActivityPropertyOptionsProvider, SelectOptionsProvider>();

15
src/core/Elsa.Core/Extensions/TypeBaseRouterConfigurationBuilderExtensions.cs Normal file
View file

@ -0,0 +1,15 @@
using System;
using System.Collections.Generic;
using Rebus.Routing.TypeBased;
namespace Elsa.Extensions
{
public static class TypeBaseRouterConfigurationBuilderExtensions
{
public static TypeBasedRouterConfigurationExtensions.TypeBasedRouterConfigurationBuilder Map(this TypeBasedRouterConfigurationExtensions.TypeBasedRouterConfigurationBuilder builder, IDictionary<Type, string> map)
{
foreach (var entry in map) builder.Map(entry.Key, entry.Value);
return builder;
}
}
}

11
src/core/Elsa.Core/Runtime/ServiceCollectionExtensions.cs
View file

@ -1,3 +1,4 @@
using System;
using Elsa.Services;
using Microsoft.Extensions.DependencyInjection;
@ -12,12 +13,18 @@ namespace Elsa.Runtime
.AddHostedService<StartupRunnerHostedService>();
}
public static IServiceCollection AddStartupTask<TStartupTask>(this IServiceCollection services)
where TStartupTask : class, IStartupTask
public static IServiceCollection AddStartupTask<TStartupTask>(this IServiceCollection services) where TStartupTask : class, IStartupTask
{
return services
.AddTransient<IStartupTask, TStartupTask>()
.AddTransient<TStartupTask>();
}
public static IServiceCollection AddStartupTask<TStartupTask>(this IServiceCollection services, Func<IServiceProvider, TStartupTask> factory) where TStartupTask : class, IStartupTask
{
return services
.AddTransient<IStartupTask, TStartupTask>(factory)
.AddTransient<TStartupTask>();
}
}
}

2
src/core/Elsa.Core/Services/CommandSender.cs
View file

@ -14,7 +14,7 @@ namespace Elsa.Services
public async Task SendAsync(object message, IDictionary<string, string>? headers = default)
{
var bus = _serviceBusFactory.GetServiceBus(message.GetType().Name);
var bus = _serviceBusFactory.GetServiceBus(message.GetType());
await bus.Send(message, headers);
}
}

2
src/core/Elsa.Core/Services/EventPublisher.cs
View file

@ -14,7 +14,7 @@ namespace Elsa.Services
public async Task PublishAsync(object message, IDictionary<string, string>? headers = default)
{
var bus = _serviceBusFactory.GetServiceBus("elsa:publisher");
var bus = _serviceBusFactory.GetServiceBus(message.GetType());
await bus.Publish(message, headers);
}
}

7
src/core/Elsa.Core/Services/ServiceBusEndpointConfigurationContext.cs
View file

@ -1,21 +1,22 @@
using System;
using System.Collections.Generic;
using Rebus.Config;
namespace Elsa.Services
{
public class ServiceBusEndpointConfigurationContext
{
public ServiceBusEndpointConfigurationContext(RebusConfigurer configurer, string queueName, Type messageType, IServiceProvider serviceProvider)
public ServiceBusEndpointConfigurationContext(RebusConfigurer configurer, string queueName, IDictionary<Type, string> messageTypeMap, IServiceProvider serviceProvider)
{
Configurer = configurer;
QueueName = queueName;
MessageType = messageType;
MessageTypeMap = messageTypeMap;
ServiceProvider = serviceProvider;
}
public RebusConfigurer Configurer { get; }
public string QueueName { get; }
public Type MessageType { get; }
public IDictionary<Type, string> MessageTypeMap { get; }
public IServiceProvider ServiceProvider { get; set; }
}
}

32
src/core/Elsa.Core/Services/ServiceBusFactory.cs
View file

@ -8,25 +8,35 @@ namespace Elsa.Services
{
public class ServiceBusFactory : IServiceBusFactory
{
private readonly ElsaOptions _elsaOptions;
private readonly IServiceProvider _serviceProvider;
private readonly IDictionary<string, IBus> _serviceBuses = new Dictionary<string, IBus>();
private readonly DependencyInjectionHandlerActivator _handlerActivator;
public ServiceBusFactory(IServiceProvider serviceProvider)
public ServiceBusFactory(ElsaOptions elsaOptions, IServiceProvider serviceProvider)
{
_elsaOptions = elsaOptions;
_serviceProvider = serviceProvider;
_handlerActivator = new DependencyInjectionHandlerActivator(serviceProvider);
}
public IBus CreateServiceBus(string name, Func<RebusConfigurer, IServiceProvider, RebusConfigurer> configure)
public IBus GetServiceBus(Type messageType)
{
var configurer = Configure.With(new DependencyInjectionHandlerActivator(_serviceProvider));
configurer = configure(configurer, _serviceProvider);
var queueName = messageType.Name;
var bus = configurer.Start();
_serviceBuses.Add(name, bus);
return bus;
if (_serviceBuses.TryGetValue(queueName, out var bus))
return bus;
var configurer = Configure.With(_handlerActivator);
var map = new Dictionary<Type, string> { [messageType] = queueName };
var configureContext = new ServiceBusEndpointConfigurationContext(configurer, queueName, map, _serviceProvider);
_elsaOptions.ConfigureServiceBusEndpoint(configureContext);
var newBus = configurer.Start();
_serviceBuses.Add(queueName, newBus);
return newBus;
}
public IBus GetServiceBus(string name) => _serviceBuses[name];
}
}

20
src/core/Elsa.Core/StartupTasks/StartServiceBusTask.cs
View file

@ -1,20 +0,0 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Messages;
using Elsa.Services;
using Microsoft.Extensions.DependencyInjection;
using Rebus.Persistence.InMem;
using Rebus.Routing.TypeBased;
using Rebus.Transport.InMem;
namespace Elsa.StartupTasks
{
public class StartServiceBusTask : IStartupTask
{
public Task ExecuteAsync(CancellationToken cancellationToken = default)
{
return Task.CompletedTask;
}
}
}

13
src/providers/Elsa.DistributedLocking.AzureBlob/ElsaOptionsExtensions.cs
View file

@ -6,14 +6,15 @@ namespace Elsa
{
public static class ElsaOptionsExtensions
{
public static ElsaOptions UseAzureBlobLockProvider(this ElsaOptions options, string connectionString, TimeSpan? leaseTime=null, TimeSpan? renewInterval=null)
public static ElsaOptions UseAzureBlobLockProvider(this ElsaOptions options, string connectionString, TimeSpan? leaseTime = null, TimeSpan? renewInterval = null)
{
options
.UseDistributedLockProvider(sp => new AzureBlobLockProvider(connectionString,
leaseTime ?? TimeSpan.FromSeconds(60),
renewInterval ?? TimeSpan.FromSeconds(45),
sp.GetRequiredService<ILogger<AzureBlobLockProvider>>()));
.UseDistributedLockProvider(sp =>
new AzureBlobLockProvider(connectionString,
leaseTime ?? TimeSpan.FromSeconds(60),
renewInterval ?? TimeSpan.FromSeconds(45),
sp.GetRequiredService<ILogger<AzureBlobLockProvider>>()));
return options;
}
}
}
}

1
src/samples/Elsa.Samples.RebusWorker/Messages/Greeting.cs
View file

@ -5,5 +5,6 @@
public string From { get; set; }
public string To { get; set; }
public string Message { get; set; }
public override string ToString() => $"{From} says \"{Message}\" to {To}.";
}
}

4
src/samples/Elsa.Samples.RebusWorker/Program.cs
View file

@ -20,9 +20,7 @@ namespace Elsa.Samples.RebusWorker
.ConfigureServices((hostContext, services) =>
{
services
.AddElsa(option => option
.UsePersistence(db => db.UseSqLite("Data Source=elsa.db;Cache=Shared"))
.AddEndpoint<Greeting>())
.AddElsa(option => option.UsePersistence(db => db.UseSqLite("Data Source=elsa.db;Cache=Shared")))
.AddConsoleActivities()
.AddTimerActivities(options => options.SweepInterval = Duration.FromSeconds(1))
.AddRebusActivities<Greeting>()

5
src/samples/Elsa.Samples.RebusWorker/Workflows/ProducerWorkflow.cs
View file

@ -1,5 +1,4 @@
using System;
using System.Linq;
using Elsa.Activities.Console;
using Elsa.Activities.Rebus;
using Elsa.Activities.Timers;
@ -25,9 +24,7 @@ namespace Elsa.Samples.RebusWorker.Workflows
workflow
.TimerEvent(Duration.FromSeconds(5))
.WriteLine("Sending a random greeting to the \"greetings\" queue.")
//.Then<PublishMessage>(sendMessage => sendMessage.Set(x => x.Message, GetRandomGreeting))
.Then<SendMessage>(sendMessage => sendMessage.Set(x => x.Message, GetRandomGreeting))
.WriteLine(() => $"Message sent at {_clock.GetCurrentInstant()}");
.Then<SendMessage>(sendMessage => sendMessage.Set(x => x.Message, GetRandomGreeting));
}
private Greeting GetRandomGreeting()