-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Processing a control message causes the outbox to throw a null refere…
…nce exception (#549) (#550) * Processing a control message causes the outbox to throw a null reference exception * Update src/NServiceBus.Persistence.CosmosDB/Outbox/OutboxBehavior.cs --------- Co-authored-by: Andreas Öhlund <andreas.ohlund@particular.net>
- Loading branch information
1 parent
64775ee
commit 85524a9
Showing
6 changed files
with
236 additions
and
3 deletions.
There are no files selected for viewing
82 changes: 82 additions & 0 deletions
82
....Persistence.CosmosDB.PhysicalOutbox.AcceptanceTests/When_using_outbox_control_message.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,82 @@ | ||
namespace NServiceBus.AcceptanceTests | ||
{ | ||
using System; | ||
using System.Threading; | ||
using System.Threading.Tasks; | ||
using AcceptanceTesting; | ||
using EndpointTemplates; | ||
using NServiceBus.AcceptanceTesting.Support; | ||
using NServiceBus.Features; | ||
using NServiceBus.Pipeline; | ||
using NServiceBus.Routing; | ||
using NServiceBus.Transport; | ||
using NServiceBus.Unicast.Transport; | ||
using NUnit.Framework; | ||
|
||
[TestFixture] | ||
public class When_using_outbox_control_message : NServiceBusAcceptanceTest | ||
{ | ||
[Test] | ||
public async Task Should_work() | ||
{ | ||
var runSettings = new RunSettings(); | ||
runSettings.DoNotRegisterDefaultPartitionKeyProvider(); | ||
|
||
var context = await Scenario.Define<Context>() | ||
.WithEndpoint<Endpoint>() | ||
.Done(c => c.ProcessedControlMessage) | ||
.Run(runSettings) | ||
.ConfigureAwait(false); | ||
|
||
Assert.True(context.ProcessedControlMessage); | ||
} | ||
|
||
public class Context : ScenarioContext | ||
{ | ||
public bool ProcessedControlMessage { get; set; } | ||
} | ||
|
||
public class Endpoint : EndpointConfigurationBuilder | ||
{ | ||
public Endpoint() => | ||
EndpointSetup<DefaultServer>((config, runDescriptor) => | ||
{ | ||
config.EnableOutbox(); | ||
config.ConfigureTransport().TransportTransactionMode = TransportTransactionMode.ReceiveOnly; | ||
config.RegisterStartupTask<ControlMessageSender>(); | ||
config.Pipeline.Register(new ControlMessageBehavior(runDescriptor.ScenarioContext as Context), "Checks that the control message was processed successfully"); | ||
}); | ||
|
||
class ControlMessageSender : FeatureStartupTask | ||
{ | ||
public ControlMessageSender(IMessageDispatcher dispatcher) => this.dispatcher = dispatcher; | ||
|
||
protected override Task OnStart(IMessageSession session, CancellationToken cancellationToken = default) | ||
{ | ||
var controlMessage = ControlMessageFactory.Create(MessageIntent.Subscribe); | ||
var messageOperation = new TransportOperation(controlMessage, new UnicastAddressTag(AcceptanceTesting.Customization.Conventions.EndpointNamingConvention(typeof(Endpoint)))); | ||
|
||
return dispatcher.Dispatch(new TransportOperations(messageOperation), new TransportTransaction(), cancellationToken); | ||
} | ||
|
||
protected override Task OnStop(IMessageSession session, CancellationToken cancellationToken = default) => Task.CompletedTask; | ||
|
||
readonly IMessageDispatcher dispatcher; | ||
} | ||
|
||
class ControlMessageBehavior : Behavior<IIncomingPhysicalMessageContext> | ||
{ | ||
public ControlMessageBehavior(Context testContext) => this.testContext = testContext; | ||
|
||
public override async Task Invoke(IIncomingPhysicalMessageContext context, Func<Task> next) | ||
{ | ||
await next(); | ||
|
||
testContext.ProcessedControlMessage = true; | ||
} | ||
|
||
readonly Context testContext; | ||
} | ||
} | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
19 changes: 19 additions & 0 deletions
19
src/NServiceBus.Persistence.CosmosDB/Outbox/SetAsDispatchedHolderExtensions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,19 @@ | ||
#nullable enable | ||
|
||
namespace NServiceBus.Persistence.CosmosDB | ||
{ | ||
using System; | ||
|
||
static class SetAsDispatchedHolderExtensions | ||
{ | ||
public static void ThrowIfContainerIsNotSet(this SetAsDispatchedHolder setAsDispatchedHolder) | ||
{ | ||
if (setAsDispatchedHolder.ContainerHolder is { Container: not null }) | ||
{ | ||
return; | ||
} | ||
|
||
throw new Exception($"For the outbox to work a container must be configured. Either configure a default one using '{nameof(CosmosPersistenceConfig.DefaultContainer)}' or set one via '{nameof(CosmosPersistenceConfig.TransactionInformation)}'."); | ||
} | ||
} | ||
} |
59 changes: 59 additions & 0 deletions
59
src/SharedAcceptanceTests.RequirePartitionKey/When_no_container_information_is_configured.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,59 @@ | ||
namespace NServiceBus.AcceptanceTests | ||
{ | ||
using System.Linq; | ||
using System.Threading.Tasks; | ||
using AcceptanceTesting; | ||
using EndpointTemplates; | ||
using NServiceBus.AcceptanceTesting.Support; | ||
using NUnit.Framework; | ||
|
||
[TestFixture] | ||
public class When_no_container_information_is_configured : NServiceBusAcceptanceTest | ||
{ | ||
[Test] | ||
public async Task Should_throw_meaningful_exception() | ||
{ | ||
var runSettings = new RunSettings(); | ||
runSettings.DoNotRegisterDefaultContainerInformationProvider(); | ||
|
||
var context = await Scenario.Define<Context>() | ||
.WithEndpoint<Endpoint>(b => | ||
{ | ||
b.DoNotFailOnErrorMessages(); | ||
b.When(s => s.SendLocal(new MyMessage())); | ||
}) | ||
.Done(c => c.FailedMessages.Any()) | ||
.Run(runSettings); | ||
|
||
var failure = context.FailedMessages.FirstOrDefault() | ||
.Value.First(); | ||
|
||
Assert.That(failure.Exception.Message, Does.Contain("container")); | ||
} | ||
|
||
class Context : ScenarioContext | ||
{ | ||
} | ||
|
||
class Endpoint : EndpointConfigurationBuilder | ||
{ | ||
public Endpoint() => | ||
EndpointSetup<DefaultServer>((config, runDescriptor) => | ||
{ | ||
config.EnableOutbox(); | ||
config.ConfigureTransport().TransportTransactionMode = TransportTransactionMode.ReceiveOnly; | ||
}); | ||
|
||
class MyMessageHandler : IHandleMessages<MyMessage> | ||
{ | ||
public Task Handle(MyMessage message, IMessageHandlerContext context) | ||
{ | ||
Assert.Fail("Should not be called"); | ||
return Task.CompletedTask; | ||
} | ||
} | ||
} | ||
|
||
class MyMessage : IMessage { } | ||
} | ||
} |
59 changes: 59 additions & 0 deletions
59
src/SharedAcceptanceTests.RequirePartitionKey/When_no_partition_key_is_configured.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,59 @@ | ||
namespace NServiceBus.AcceptanceTests | ||
{ | ||
using System.Linq; | ||
using System.Threading.Tasks; | ||
using AcceptanceTesting; | ||
using EndpointTemplates; | ||
using NServiceBus.AcceptanceTesting.Support; | ||
using NUnit.Framework; | ||
|
||
[TestFixture] | ||
public class When_no_partition_key_is_configured : NServiceBusAcceptanceTest | ||
{ | ||
[Test] | ||
public async Task Should_throw_meaningful_exception() | ||
{ | ||
var runSettings = new RunSettings(); | ||
runSettings.DoNotRegisterDefaultPartitionKeyProvider(); | ||
|
||
var context = await Scenario.Define<Context>() | ||
.WithEndpoint<Endpoint>(b => | ||
{ | ||
b.DoNotFailOnErrorMessages(); | ||
b.When(s => s.SendLocal(new MyMessage())); | ||
}) | ||
.Done(c => c.FailedMessages.Any()) | ||
.Run(runSettings); | ||
|
||
var failure = context.FailedMessages.FirstOrDefault() | ||
.Value.First(); | ||
|
||
Assert.That(failure.Exception.Message, Does.Contain("partition key")); | ||
} | ||
|
||
class Context : ScenarioContext | ||
{ | ||
} | ||
|
||
class Endpoint : EndpointConfigurationBuilder | ||
{ | ||
public Endpoint() => | ||
EndpointSetup<DefaultServer>((config, runDescriptor) => | ||
{ | ||
config.EnableOutbox(); | ||
config.ConfigureTransport().TransportTransactionMode = TransportTransactionMode.ReceiveOnly; | ||
}); | ||
|
||
class MyMessageHandler : IHandleMessages<MyMessage> | ||
{ | ||
public Task Handle(MyMessage message, IMessageHandlerContext context) | ||
{ | ||
Assert.Fail("Should not be called"); | ||
return Task.CompletedTask; | ||
} | ||
} | ||
} | ||
|
||
class MyMessage : IMessage { } | ||
} | ||
} |