SagaInstance creation infrastructure

src/NServiceBus.IntegrationTesting/EndpointConfigurationExtensions.cs

@@ -0,0 +1,31 @@
+using NServiceBus.AcceptanceTesting;
+using NServiceBus.ObjectBuilder;
+
+namespace NServiceBus.IntegrationTesting
+{
+ public static class EndpointConfigurationExtensions
+ {
+ public static void RegisterRequiredPipelineBehaviors(this EndpointConfiguration builder, string endpointName, IntegrationScenarioContext integrationScenarioContext) 
+ {
+ builder.Pipeline.Register(new InterceptInvokedHandlers(endpointName, integrationScenarioContext), "Intercept invoked Message Handlers and Sagas");
+ builder.Pipeline.Register(new InterceptSendOperations(endpointName, integrationScenarioContext), "Intercept send operations");
+ builder.Pipeline.Register(new InterceptPublishOperations(endpointName, integrationScenarioContext), "Intercept publish operations");
+ builder.Pipeline.Register(new InterceptReplyOperations(endpointName, integrationScenarioContext), "Intercept reply operations");
+ }
+
+ public static void RegisterScenarioContext(this EndpointConfiguration builder, ScenarioContext scenarioContext)
+ {
+ builder.RegisterComponents(r => { RegisterInheritanceHierarchyOfContextOnContainer(scenarioContext, r); });
+ }
+
+ static void RegisterInheritanceHierarchyOfContextOnContainer(ScenarioContext scenarioContext, IConfigureComponents r)
+ {
+ var type = scenarioContext.GetType();
+ while (type != typeof(object))
+ {
+ r.RegisterSingleton(type, scenarioContext);
+ type = type.BaseType;
+ }
+ }
+ }
+}

src/NServiceBus.IntegrationTesting/EndpointTemplate.cs

@@ -1,5 +1,7 @@
-using NServiceBus.AcceptanceTesting.Support;
+using NServiceBus.AcceptanceTesting.Customization;
+using NServiceBus.AcceptanceTesting.Support;
 using NServiceBus.Configuration.AdvancedExtensibility;
+using NServiceBus.IntegrationTesting.Messages.Handlers;
 using System;
 using System.Threading.Tasks;
 
@@ -13,13 +15,16 @@ public async Task<EndpointConfiguration> GetConfiguration(RunDescriptor runDescr
 
  var settings = configuration.GetSettings();
  endpointCustomizationConfiguration.EndpointName = settings.EndpointName();
-
  configurationBuilderCustomization(configuration);
 
- configuration.Pipeline.Register(new InterceptInvokedHandlers(endpointCustomizationConfiguration.EndpointName, (IntegrationScenarioContext)runDescriptor.ScenarioContext), "Intercept invoked Message Handlers and Sagas");
- configuration.Pipeline.Register(new InterceptSendOperations(endpointCustomizationConfiguration.EndpointName, (IntegrationScenarioContext)runDescriptor.ScenarioContext), "Intercept send operations");
- configuration.Pipeline.Register(new InterceptPublishOperations(endpointCustomizationConfiguration.EndpointName, (IntegrationScenarioContext)runDescriptor.ScenarioContext), "Intercept publish operations");
- configuration.Pipeline.Register(new InterceptReplyOperations(endpointCustomizationConfiguration.EndpointName, (IntegrationScenarioContext)runDescriptor.ScenarioContext), "Intercept reply operations");
+ configuration.TypesToIncludeInScan(new[]
+ {
+ typeof(CreateSagaInstanceHandler),
+ typeof(MarkSagaInstanceAsCreatedHandler)
+ });
+
+ configuration.RegisterScenarioContext(runDescriptor.ScenarioContext);
+ configuration.RegisterRequiredPipelineBehaviors(endpointCustomizationConfiguration.EndpointName, (IntegrationScenarioContext)runDescriptor.ScenarioContext);
 
  return configuration;
  }

src/NServiceBus.IntegrationTesting/IntegrationScenarioContext.cs

@@ -1,4 +1,5 @@
 using NServiceBus.AcceptanceTesting;
+using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
@@ -11,6 +12,7 @@ public class IntegrationScenarioContext : ScenarioContext
  readonly ConcurrentBag<HandlerInvocation> invokedHandlers = new ConcurrentBag<HandlerInvocation>();
  readonly ConcurrentBag<SagaInvocation> invokedSagas = new ConcurrentBag<SagaInvocation>();
  readonly ConcurrentBag<OutgoingMessageOperation> outgoingMessageOperations = new ConcurrentBag<OutgoingMessageOperation>();
+ readonly ConcurrentDictionary<Guid, Type> createdSagaInstances = new ConcurrentDictionary<Guid, Type>();
 
  public IEnumerable<HandlerInvocation> InvokedHandlers { get { return invokedHandlers; } }
  public IEnumerable<SagaInvocation> InvokedSagas { get { return invokedSagas; } }
@@ -35,6 +37,16 @@ internal SagaInvocation CaptureInvokedSaga(SagaInvocation invocation)
  return invocation;
  }
 
+ internal void RegisterSagaInstanceAsCreated(Guid sagaId, Type sagaDataType)
+ {
+ createdSagaInstances.AddOrUpdate(sagaId, sagaDataType, (id, type) => type);
+ }
+
+ public bool IsSagaInstanceAvailable(Guid sagaId) 
+ {
+ return createdSagaInstances.ContainsKey(sagaId);
+ }
+
  public bool HandlerWasInvoked<THandler>()
  {
  return InvokedHandlers.Any(invocation => invocation.HandlerType == typeof(THandler));

src/NServiceBus.IntegrationTesting/MessageSessionExtensions.cs

@@ -0,0 +1,20 @@
+using System;
+using System.Threading.Tasks;
+
+namespace NServiceBus.IntegrationTesting
+{
+ public static class MessageSessionExtensions
+ {
+ public static Task CreateSagaInstance<TSagaData>(this IMessageSession messageSession, string sagaOwnerEndpoint, Guid sagaId, TSagaData sagaData, string originator = null, string originatingMessageId = null) where TSagaData : IContainSagaData
+ {
+ sagaData.Id = sagaId;
+ sagaData.Originator = originator;
+ sagaData.OriginalMessageId = originatingMessageId;
+
+ return messageSession.Send(sagaOwnerEndpoint, new Messages.CreateSagaInstance() 
+ { 
+ SagaData = sagaData
+ });
+ }
+ }
+}

src/NServiceBus.IntegrationTesting/Messages/CreateSagaInstance.cs

@@ -0,0 +1,9 @@
+namespace NServiceBus.IntegrationTesting.Messages
+{
+ class CreateSagaInstance
+ {
+ public string CorrelationPropertyName { get; set; }
+ public object CorrelationPropertyValue { get; set; }
+ public IContainSagaData SagaData { get; set; }
+ }
+}

src/NServiceBus.IntegrationTesting/Messages/Handlers/CreateSagaInstanceHandler.cs

@@ -0,0 +1,35 @@
+using NServiceBus.AcceptanceTesting;
+using NServiceBus.Sagas;
+using System.Threading.Tasks;
+
+namespace NServiceBus.IntegrationTesting.Messages.Handlers
+{
+ class CreateSagaInstanceHandler : IHandleMessages<CreateSagaInstance>
+ {
+ readonly ISagaPersister sagaPersister;
+
+ public CreateSagaInstanceHandler(ISagaPersister sagaPersister)
+ {
+ this.sagaPersister = sagaPersister;
+ }
+
+ public async Task Handle(CreateSagaInstance message, IMessageHandlerContext context)
+ {
+ var correlationProperty = new SagaCorrelationProperty(
+ message.CorrelationPropertyName, 
+ message.CorrelationPropertyValue);
+
+ await sagaPersister.Save(message.SagaData, 
+ correlationProperty, 
+ context.SynchronizedStorageSession, 
+ new Extensibility.ContextBag())
+ .ConfigureAwait(false);
+
+ await context.SendLocal(new MarkSagaInstanceAsCreated() 
+ {
+ SagaId = message.SagaData.Id,
+ SagaDataType = message.SagaData.GetType()
+ }).ConfigureAwait(false);
+ }
+ }
+}

src/NServiceBus.IntegrationTesting/Messages/Handlers/MarkSagaInstanceAsCreatedHandler.cs

@@ -0,0 +1,24 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace NServiceBus.IntegrationTesting.Messages.Handlers
+{
+ class MarkSagaInstanceAsCreatedHandler : IHandleMessages<MarkSagaInstanceAsCreated>
+ {
+ readonly IntegrationScenarioContext integrationScenarioContext;
+
+ public MarkSagaInstanceAsCreatedHandler(IntegrationScenarioContext integrationScenarioContext)
+ {
+ this.integrationScenarioContext = integrationScenarioContext;
+ }
+
+ public Task Handle(MarkSagaInstanceAsCreated message, IMessageHandlerContext context)
+ {
+ integrationScenarioContext.RegisterSagaInstanceAsCreated(message.SagaId, message.SagaDataType);
+
+ return Task.CompletedTask;
+ }
+ }
+}

src/NServiceBus.IntegrationTesting/Messages/MarkSagaInstanceAsCreated.cs

@@ -0,0 +1,10 @@
+using System;
+
+namespace NServiceBus.IntegrationTesting.Messages
+{
+ class MarkSagaInstanceAsCreated
+ {
+ public Guid SagaId { get; set; }
+ public Type SagaDataType { get; set; }
+ }
+}