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

Service bus API improvements

Browse source
This commit is contained in:
Sipke Schoorstra 2020-11-20 23:24:48 +01:00
parent 355038f27e
commit f1349adc3f
24 changed files with 281 additions and 219 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -10,13 +10,6 @@ namespace Elsa.Activities.Rebus
[Trigger(Category = "Rebus", Description = "Triggered when a message is received.", Outcomes = new[] { OutcomeNames.Done })]
public class MessageReceived : Activity
{
private readonly IBus _bus;
public MessageReceived(IBus bus)
{
_bus = bus;
}
[ActivityProperty(Hint = "The type of message to receive.")]
public Type MessageType { get; set; } = default!;

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

@ -12,11 +12,11 @@ namespace Elsa.Activities.Rebus
[Action(Category = "Rebus", Description = "Publishes a message.", Outcomes = new[] { OutcomeNames.Done })]
public class PublishMessage : Activity
{
private readonly IBus _bus;
private readonly IEventPublisher _eventPublisher;
public PublishMessage(IBus bus)
public PublishMessage(IEventPublisher eventPublisher)
{
_bus = bus;
_eventPublisher = eventPublisher;
}
[ActivityProperty(Hint = "The message to publish.")]
@ -27,7 +27,7 @@ namespace Elsa.Activities.Rebus
protected override async ValueTask<IActivityExecutionResult> OnExecuteAsync(ActivityExecutionContext context, CancellationToken cancellationToken)
{
await _bus.Publish(Message, Headers);
await _eventPublisher.PublishAsync(Message, Headers);
return Done();
}
}

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

@ -12,9 +12,9 @@ namespace Elsa.Activities.Rebus
[Action(Category = "Rebus", Description = "Publishes a message.", Outcomes = new[] { OutcomeNames.Done })]
public class SendMessage : Activity
{
private readonly IBus _bus;
private readonly ICommandSender _bus;
public SendMessage(IBus bus)
public SendMessage(ICommandSender bus)
{
_bus = bus;
}
@ -27,10 +27,7 @@ namespace Elsa.Activities.Rebus
protected override async ValueTask<IActivityExecutionResult> OnExecuteAsync(ActivityExecutionContext context, CancellationToken cancellationToken)
{
await _bus.Advanced.Routing.Send("greeting", Message, Headers);
await _bus.Advanced.Topics.Subscribe("greeting");
//await _bus.Send(Message, Headers);
await _bus.SendAsync(Message, Headers);
return Done();
}
}

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

@ -1,26 +1,37 @@
using System;
using Elsa.Activities.Rebus.Consumers;
using Elsa.Activities.Rebus.Triggers;
using Microsoft.Extensions.DependencyInjection;
using Rebus.Handlers;
namespace Elsa.Activities.Rebus.Extensions
{
public static class ServiceCollectionExtensions
{
public static IServiceCollection AddRebusActivities(this IServiceCollection services) =>
services
public static IServiceCollection AddRebusActivities(this IServiceCollection services, params Type[] messageTypes)
{
foreach (var messageType in messageTypes)
{
var handlerServiceType = typeof(IHandleMessages<>).MakeGenericType(messageType);
var handlerImplementationType = typeof(MessageConsumer<>).MakeGenericType(messageType);
services.AddTransient(handlerServiceType, handlerImplementationType);
}
return services
.AddTriggerProvider<MessageReceivedTriggerProvider>()
.AddActivity<PublishMessage>()
.AddActivity<SendMessage>()
.AddActivity<MessageReceived>();
}
// public static IServiceCollection AddRebusActivities<T>(this IServiceCollection services) => services.AddRebusActivities().AddMessageType<T>();
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.AddConsumer<T, MessageConsumer<T>>();
//public static IServiceCollection AddMessageType<T>(this IServiceCollection services) => services.AddTransient<IHandleMessages<T>, MessageConsumer<T>>();
}
}

10
src/core/Elsa.Abstractions/Services/ICommandSender.cs Normal file
View file

@ -0,0 +1,10 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Elsa.Services
{
public interface ICommandSender
{
Task SendAsync(object message, IDictionary<string, string>? headers = default);
}
}

10
src/core/Elsa.Abstractions/Services/IEventPublisher.cs Normal file
View file

@ -0,0 +1,10 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Elsa.Services
{
public interface IEventPublisher
{
Task PublishAsync(object message, IDictionary<string, string>? headers = default);
}
}

12
src/core/Elsa.Abstractions/Services/IServiceBusFactory.cs Normal file
View file

@ -0,0 +1,12 @@
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);
}
}

8
src/core/Elsa.Core/Consumers/RunWorkflowConsumer.cs
View file

@ -1,4 +1,3 @@
using System;
using System.Threading.Tasks;
using Elsa.Exceptions;
using Elsa.Extensions;
@ -8,7 +7,7 @@ using Rebus.Handlers;
namespace Elsa.Consumers
{
public class RunWorkflowConsumer : IHandleMessages<RunWorkflow>, IDisposable
public class RunWorkflowConsumer : IHandleMessages<RunWorkflow>
{
private readonly IWorkflowRunner _workflowRunner;
private readonly IWorkflowInstanceManager _workflowInstanceManager;
@ -31,10 +30,5 @@ namespace Elsa.Consumers
message.ActivityId,
message.Input);
}
public void Dispose()
{
}
}
}

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

@ -1,18 +1,19 @@
using System;
using System.Collections.Generic;
using System.Data;
using Elsa.Caching;
using Elsa.DistributedLock;
using Elsa.Extensions;
using Elsa.Messages;
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.ServiceProvider;
using Rebus.Transport.InMem;
using YesSql;
using Storage.Net;
@ -31,22 +32,22 @@ namespace Elsa
StorageFactory = sp => Storage.Net.StorageFactory.Blobs.InMemory();
DistributedLockProviderFactory = sp => new DefaultLockProvider();
SignalFactory = sp => new Signal();
JsonSerializerConfigurer = (sp, serializer) => {};
JsonSerializerConfigurer = (sp, serializer) => { };
AddAutoMapper = () =>
{
services.AddAutoMapper(ServiceLifetime.Singleton);
services.AddSingleton(sp => sp.CreateAutoMapperConfiguration());
};
AddServiceBus = () =>
{
services.AddSingleton<InMemNetwork>();
services.AddSingleton<InMemorySubscriberStore>();
services.AddSingleton<InMemDataStore>();
services.AddRebus(ConfigureInMemoryServiceBus);
};
services.AddSingleton<InMemNetwork>();
services.AddSingleton<InMemorySubscriberStore>();
services.AddSingleton<InMemDataStore>();
CreateServiceBus = (name, configure, sp) => sp.GetRequiredService<IServiceBusFactory>().CreateServiceBus(name, configure);
ConfigureServiceBusEndpoint = ConfigureInMemoryServiceBusEndpoint;
ConfigureServiceBusPublishEndpoint = ConfigureInMemoryServiceBusPublishEndpoint;
CreateJsonSerializer = sp =>
{
var serializer = DefaultContentSerializer.CreateDefaultJsonSerializer();
@ -56,6 +57,7 @@ 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; }
@ -63,7 +65,9 @@ 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 Action AddServiceBus { 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)
{
@ -92,19 +96,13 @@ namespace Elsa
StorageFactory = factory;
return this;
}
public ElsaOptions UseAutoMapper(Action addAutoMapper)
{
AddAutoMapper = addAutoMapper;
return this;
}
public ElsaOptions UseServiceBus(Action addServiceBus)
{
AddServiceBus = addServiceBus;
return this;
}
public ElsaOptions UseJsonSerializer(Func<IServiceProvider, JsonSerializer> factory)
{
CreateJsonSerializer = factory;
@ -117,18 +115,51 @@ namespace Elsa
return this;
}
private static RebusConfigurer ConfigureInMemoryServiceBus(RebusConfigurer rebus, IServiceProvider serviceProvider)
public ElsaOptions UseServiceBusFactory(Func<string, Func<RebusConfigurer, IServiceProvider, RebusConfigurer>, IServiceProvider, IBus> serviceBusFactory)
{
var subscriberStore = serviceProvider.GetRequiredService<InMemorySubscriberStore>();
var dataStore = serviceProvider.GetRequiredService<InMemDataStore>();
var network = serviceProvider.GetRequiredService<InMemNetwork>();
return rebus
.Logging(logging => logging.ColoredConsole(LogLevel.Debug))
.Subscriptions(s => s.StoreInMemory(subscriberStore))
.DataBus(s => s.StoreInMemory(dataStore))
//.Routing(r => r.TypeBased().Map<RunWorkflow>("runworkflow"))
.Transport(t => t.UseInMemoryTransport(network, "elsa:publisher"));
CreateServiceBus = serviceBusFactory;
return this;
}
public ElsaOptions SetupServiceBusEndpoint(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;
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.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));
}
}
}

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

@ -18,7 +18,6 @@ using Elsa.Metadata;
using Elsa.Metadata.Handlers;
using Elsa.Runtime;
using Elsa.Serialization;
using Elsa.ServiceBus;
using Elsa.Services;
using Elsa.StartupTasks;
using Elsa.Triggers;
@ -26,12 +25,7 @@ using Elsa.WorkflowProviders;
using MediatR;
using Microsoft.Extensions.DependencyInjection.Extensions;
using NodaTime;
using Rebus.Bus;
using Rebus.Config;
using Rebus.Handlers;
using Rebus.Routing.TypeBased;
using Rebus.ServiceProvider;
using Rebus.Transport.InMem;
using RunWorkflow = Elsa.Messages.RunWorkflow;
// ReSharper disable once CheckNamespace
@ -55,8 +49,8 @@ namespace Microsoft.Extensions.DependencyInjection
options.AddWorkflowsCore();
options.AddMediatR();
options.AddServiceBus();
options.AddAutoMapper();
options.AddEndpoint<RunWorkflow>();
return services;
}
@ -83,14 +77,6 @@ namespace Microsoft.Extensions.DependencyInjection
.AddTransient(sp => workflow);
}
public static void AddConsumer<TMessage, TConsumer>(this IServiceProvider sp, string name, Func<RebusConfigurer, IServiceProvider, RebusConfigurer> configureBus)
where TConsumer : class, IHandleMessages<TMessage>
{
var factory = sp.GetRequiredService<IServiceBusContainerFactory>();
var container = factory.CreateBusContainer(name, configureBus);
container.Bus.Subscribe<TMessage>();
}
private static IServiceCollection AddMediatR(this ElsaOptions options) => options.Services.AddMediatR(mediatr => mediatr.AsScoped(), typeof(IActivity));
private static ElsaOptions AddWorkflowsCore(this ElsaOptions configuration)
@ -136,16 +122,49 @@ namespace Microsoft.Extensions.DependencyInjection
.AddSingleton<ICloner, AutoMapperCloner>()
.AddNotificationHandlers(typeof(ElsaServiceCollectionExtensions))
.AddStartupTask<StartServiceBusTask>()
.AddSingleton<IServiceBusContainerFactory, ServiceBusContainerFactory>()
.AddTransient<RunWorkflowConsumer>()
.AddTransient<IHandleMessages>(bsp => bsp.GetRequiredService<RunWorkflowConsumer>())
.AddTransient<IHandleMessages<RunWorkflow>, RunWorkflowConsumer>(bsp => bsp.GetRequiredService<RunWorkflowConsumer>())
.AddSingleton<ServiceBusFactory>()
.AddSingleton<IServiceBusFactory, ServiceBusFactory>()
.AddSingleton<ICommandSender, CommandSender>(CreateCommandSender)
.AddSingleton<IEventPublisher, EventPublisher>(CreateEventPublisher)
.AutoRegisterHandlersFromAssemblyOf<RunWorkflowConsumer>()
.AddMetadataHandlers()
.AddCoreActivities();
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>();

10
src/core/Elsa.Core/ServiceBus/IServiceBusContainer.cs
View file

@ -1,10 +0,0 @@
using System;
using Rebus.Bus;
namespace Elsa.ServiceBus
{
public interface IServiceBusContainer
{
IBus Bus { get; }
}
}

12
src/core/Elsa.Core/ServiceBus/IServiceBusContainerFactory.cs
View file

@ -1,12 +0,0 @@
using System;
using Microsoft.Extensions.DependencyInjection;
using Rebus.Config;
namespace Elsa.ServiceBus
{
public interface IServiceBusContainerFactory
{
IServiceBusContainer CreateBusContainer(string name, Func<RebusConfigurer, IServiceProvider, RebusConfigurer> configure);
IServiceBusContainer GetBusContainer(string name);
}
}

25
src/core/Elsa.Core/ServiceBus/ServiceBusContainer.cs
View file

@ -1,25 +0,0 @@
using System;
using Microsoft.Extensions.DependencyInjection;
using Rebus.Bus;
using Rebus.Config;
using Rebus.ServiceProvider;
namespace Elsa.ServiceBus
{
public class ServiceBusContainer : IServiceBusContainer, IDisposable
{
public ServiceBusContainer(string name, IServiceProvider serviceProvider, Func<RebusConfigurer, IServiceProvider, RebusConfigurer> configure)
{
Name = name;
var configurer = Configure.With(new DependencyInjectionHandlerActivator(serviceProvider));
configurer = configure(configurer, serviceProvider);
Bus = configurer.Start();
}
public string Name { get; }
public IBus Bus { get; }
public void Dispose() => Bus.Dispose();
}
}

27
src/core/Elsa.Core/ServiceBus/ServiceBusContainerFactory.cs
View file

@ -1,27 +0,0 @@
using System;
using System.Collections.Generic;
using Microsoft.Extensions.DependencyInjection;
using Rebus.Config;
namespace Elsa.ServiceBus
{
public class ServiceBusContainerFactory : IServiceBusContainerFactory
{
private readonly IServiceProvider _serviceProvider;
private readonly IDictionary<string, IServiceBusContainer> _containers = new Dictionary<string, IServiceBusContainer>();
public ServiceBusContainerFactory(IServiceProvider serviceProvider)
{
_serviceProvider = serviceProvider;
}
public IServiceBusContainer CreateBusContainer(string name, Func<RebusConfigurer, IServiceProvider, RebusConfigurer> configure)
{
var container = new ServiceBusContainer(name, _serviceProvider, configure);
_containers.Add(name, container);
return container;
}
public IServiceBusContainer GetBusContainer(string name) => _containers[name];
}
}

14
src/core/Elsa.Core/ServiceBus/ServiceCollectionExtensions.cs
View file

@ -1,14 +0,0 @@
using System;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Rebus.Config;
using Rebus.ServiceProvider;
namespace Elsa.ServiceBus
{
public static class ServiceCollectionExtensions
{
// public static IServiceCollection WithServiceBus(this ElsaOptions configuration, Func<RebusConfigurer, RebusConfigurer> setup) => configuration.Services.AddRebus(setup);
// public static IServiceCollection WithServiceBus(this ElsaOptions configuration, Func<RebusConfigurer, IServiceProvider, RebusConfigurer> setup) => configuration.Services.AddRebus(setup);
}
}

21
src/core/Elsa.Core/Services/CommandSender.cs Normal file
View file

@ -0,0 +1,21 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Elsa.Services
{
public class CommandSender : ICommandSender
{
private readonly IServiceBusFactory _serviceBusFactory;
public CommandSender(IServiceBusFactory serviceBusFactory)
{
_serviceBusFactory = serviceBusFactory;
}
public async Task SendAsync(object message, IDictionary<string, string>? headers = default)
{
var bus = _serviceBusFactory.GetServiceBus(message.GetType().Name);
await bus.Send(message, headers);
}
}
}

21
src/core/Elsa.Core/Services/EventPublisher.cs Normal file
View file

@ -0,0 +1,21 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Elsa.Services
{
public class EventPublisher : IEventPublisher
{
private readonly IServiceBusFactory _serviceBusFactory;
public EventPublisher(IServiceBusFactory serviceBusFactory)
{
_serviceBusFactory = serviceBusFactory;
}
public async Task PublishAsync(object message, IDictionary<string, string>? headers = default)
{
var bus = _serviceBusFactory.GetServiceBus("elsa:publisher");
await bus.Publish(message, headers);
}
}
}

21
src/core/Elsa.Core/Services/ServiceBusEndpointConfigurationContext.cs Normal file
View file

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

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

@ -0,0 +1,32 @@
using System;
using System.Collections.Generic;
using Rebus.Bus;
using Rebus.Config;
using Rebus.ServiceProvider;
namespace Elsa.Services
{
public class ServiceBusFactory : IServiceBusFactory
{
private readonly IServiceProvider _serviceProvider;
private readonly IDictionary<string, IBus> _serviceBuses = new Dictionary<string, IBus>();
public ServiceBusFactory(IServiceProvider serviceProvider)
{
_serviceProvider = serviceProvider;
}
public IBus CreateServiceBus(string name, Func<RebusConfigurer, IServiceProvider, RebusConfigurer> configure)
{
var configurer = Configure.With(new DependencyInjectionHandlerActivator(_serviceProvider));
configurer = configure(configurer, _serviceProvider);
var bus = configurer.Start();
_serviceBuses.Add(name, bus);
return bus;
}
public IBus GetServiceBus(string name) => _serviceBuses[name];
}
}

19
src/core/Elsa.Core/Services/ServiceBusPublishEndpointConfigurationContext.cs Normal file
View file

@ -0,0 +1,19 @@
using System;
using Rebus.Config;
namespace Elsa.Services
{
public class ServiceBusPublishEndpointConfigurationContext
{
public ServiceBusPublishEndpointConfigurationContext(RebusConfigurer configurer, string queueName, IServiceProvider serviceProvider)
{
Configurer = configurer;
QueueName = queueName;
ServiceProvider = serviceProvider;
}
public RebusConfigurer Configurer { get; }
public string QueueName { get; }
public IServiceProvider ServiceProvider { get; set; }
}
}

16
src/core/Elsa.Core/Services/WorkflowScheduler.cs
View file

@ -10,41 +10,36 @@ using Elsa.Extensions;
using Elsa.Indexes;
using Elsa.Messages;
using Elsa.Models;
using Elsa.ServiceBus;
using Elsa.Services.Models;
using Elsa.Triggers;
using MediatR;
using Open.Linq.AsyncExtensions;
using Rebus.Bus;
namespace Elsa.Services
{
public class WorkflowScheduler : IWorkflowScheduler, INotificationHandler<WorkflowCompleted>
{
private readonly IBus _serviceBus;
private readonly ICommandSender _commandSender;
private readonly IWorkflowInstanceManager _workflowInstanceManager;
private readonly IWorkflowFactory _workflowFactory;
private readonly IWorkflowSelector _workflowSelector;
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IWorkflowSchedulerQueue _queue;
private readonly IServiceBusContainerFactory _serviceBusContainerFactory;
public WorkflowScheduler(
IBus serviceBus,
ICommandSender commandSender,
IWorkflowInstanceManager workflowInstanceManager,
IWorkflowFactory workflowFactory,
IWorkflowSelector workflowSelector,
IWorkflowRegistry workflowRegistry,
IWorkflowSchedulerQueue queue,
IServiceBusContainerFactory serviceBusContainerFactory)
IWorkflowSchedulerQueue queue)
{
_serviceBus = serviceBus;
_commandSender = commandSender;
_workflowInstanceManager = workflowInstanceManager;
_workflowFactory = workflowFactory;
_workflowSelector = workflowSelector;
_workflowRegistry = workflowRegistry;
_queue = queue;
_serviceBusContainerFactory = serviceBusContainerFactory;
}
public async Task ScheduleWorkflowInstanceAsync(
@ -53,8 +48,7 @@ namespace Elsa.Services
object? input = default,
CancellationToken cancellationToken = default)
{
var bus = _serviceBusContainerFactory.GetBusContainer("run_workflow:client");
await bus.Bus.Send(new RunWorkflow(instanceId, activityId, input));
await _commandSender.SendAsync(new RunWorkflow(instanceId, activityId, input));
}
public async Task<IEnumerable<WorkflowInstance>> ScheduleWorkflowDefinitionAsync(

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

@ -1,54 +1,19 @@
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Consumers;
using Elsa.Messages;
using Elsa.Services;
using Microsoft.Extensions.DependencyInjection;
using Rebus.DataBus.InMem;
using Rebus.Handlers;
using Rebus.Persistence.InMem;
using Rebus.Routing.TypeBased;
using Rebus.ServiceProvider;
using Rebus.Transport.InMem;
namespace Elsa.StartupTasks
{
public class StartServiceBusTask : IStartupTask
{
private readonly IServiceProvider _serviceProvider;
public StartServiceBusTask(IServiceProvider serviceProvider) => _serviceProvider = serviceProvider;
public Task ExecuteAsync(CancellationToken cancellationToken = default)
{
var consumers = _serviceProvider.GetServices<IHandleMessages>();
var messageTypes = consumers.Select(consumer => consumer.GetType().GetInterfaces().First(x => x.GenericTypeArguments.Any()).GenericTypeArguments.First());
_serviceProvider.UseRebus(async bus =>
{
foreach (var messageType in messageTypes)
await bus.Subscribe(messageType);
});
var transport = _serviceProvider.GetService<InMemNetwork>();
var store = _serviceProvider.GetRequiredService<InMemorySubscriberStore>();
var queueName = "run_Workflow";
// Sender
_serviceProvider
.AddConsumer<RunWorkflow, RunWorkflowConsumer>("run_workflow:sender", (bus, _) => bus
.Logging(l => l.ColoredConsole())
.Transport(t => t.UseInMemoryTransportAsOneWayClient(transport)));
// Receiver
_serviceProvider
.AddConsumer<RunWorkflow, RunWorkflowConsumer>("run_workflow:client", (bus, _) => bus
.Logging(l => l.ColoredConsole())
.Subscriptions(s => s.StoreInMemory(store))
.Transport(t => t.UseInMemoryTransport(transport, queueName))
.Routing(r => r.TypeBased().Map<RunWorkflow>(queueName)));
{
return Task.CompletedTask;
}
}

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

@ -21,11 +21,11 @@ namespace Elsa.Samples.RebusWorker
{
services
.AddElsa(option => option
.UsePersistence(db => db.UseSqLite("Data Source=elsa.db;Cache=Shared")))
.UsePersistence(db => db.UseSqLite("Data Source=elsa.db;Cache=Shared"))
.AddEndpoint<Greeting>())
.AddConsoleActivities()
.AddTimerActivities(options => options.SweepInterval = Duration.FromSeconds(1))
.AddRebusActivities()
//.AddConsumer<Greeting, Gree>()
.AddRebusActivities<Greeting>()
.AddWorkflow<ProducerWorkflow>()
.AddWorkflow<ConsumerWorkflow>();
});

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

@ -25,8 +25,8 @@ 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))
//.Then<PublishMessage>(sendMessage => sendMessage.Set(x => x.Message, GetRandomGreeting))
.Then<SendMessage>(sendMessage => sendMessage.Set(x => x.Message, GetRandomGreeting))
.WriteLine(() => $"Message sent at {_clock.GetCurrentInstant()}");
}