GitHub-Mirror/Tapeti
1
0
Fork 0
mirror of synced 2024-11-21 08:53:50 +00:00
Code Issues Projects Releases Wiki Activity

Fixed queue arguments error due to wrong value types

Browse Source 
Added test for publish overflows
Removed support for Unity Container
Changed third party package references to ranges
Fixed XML documentation
This commit is contained in:
Mark van Renswoude 2022-11-22 13:20:47 +01:00
parent 35bd5e920d
commit bcdb376256
88 changed files with 257 additions and 440 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Examples/01-PublishSubscribe/01-PublishSubscribe.csproj
View File

@ -11,7 +11,6 @@
<PackageReference Include="Castle.Windsor" Version="5.1.2" />
<PackageReference Include="Ninject" Version="3.3.6" />
<PackageReference Include="SimpleInjector" Version="5.4.1" />
<PackageReference Include="Unity" Version="5.11.10" />
</ItemGroup>
<ItemGroup>
@ -20,7 +19,6 @@
<ProjectReference Include="..\..\Tapeti.DataAnnotations\Tapeti.DataAnnotations.csproj" />
<ProjectReference Include="..\..\Tapeti.Ninject\Tapeti.Ninject.csproj" />
<ProjectReference Include="..\..\Tapeti.SimpleInjector\Tapeti.SimpleInjector.csproj" />
<ProjectReference Include="..\..\Tapeti.UnityContainer\Tapeti.UnityContainer.csproj" />
<ProjectReference Include="..\..\Tapeti\Tapeti.csproj" />
<ProjectReference Include="..\ExampleLib\ExampleLib.csproj" />
<ProjectReference Include="..\Messaging.TapetiExample\Messaging.TapetiExample.csproj" />

14
Examples/01-PublishSubscribe/Program.cs
View File

@ -13,8 +13,6 @@ using Tapeti.DataAnnotations;
using Tapeti.Default;
using Tapeti.Ninject;
using Tapeti.SimpleInjector;
using Tapeti.UnityContainer;
using Unity;
using Container = SimpleInjector.Container;
// ReSharper disable UnusedMember.Global
@ -30,7 +28,6 @@ namespace _01_PublishSubscribe
// or use your IoC container of choice:
//var dependencyResolver = GetAutofacDependencyResolver();
//var dependencyResolver = GetCastleWindsorDependencyResolver();
//var dependencyResolver = GetUnityDependencyResolver();
//var dependencyResolver = GetNinjectDependencyResolver();
// This helper is used because this example is not run as a service. You do not
@ -131,17 +128,6 @@ namespace _01_PublishSubscribe
}
internal static IDependencyContainer GetUnityDependencyResolver()
{
var container = new UnityContainer();
container.RegisterType<ILogger, ConsoleLogger>();
container.RegisterType<ExamplePublisher>();
return new UnityDependencyResolver(container);
}
internal static IDependencyContainer GetNinjectDependencyResolver()
{
var kernel = new StandardKernel();

2
Examples/ExampleLib/ExampleConsoleApp.cs
View File

@ -79,7 +79,7 @@ namespace ExampleLib
{
while (true)
{
if (!(e is AggregateException aggregateException))
if (e is not AggregateException aggregateException)
return e;
if (aggregateException.InnerExceptions.Count != 1)

2
Examples/Messaging.TapetiExample/Messaging.TapetiExample.csproj
View File

@ -6,7 +6,7 @@
<ItemGroup>
<PackageReference Include="System.ComponentModel.Annotations" Version="5.0.0" />
<PackageReference Include="Tapeti.Annotations" Version="3.0.0" />
<PackageReference Include="Tapeti.Annotations" Version="3.*-*" />
</ItemGroup>
</Project>

7
Tapeti.Autofac/AutofacDependencyResolver.cs
View File

@ -2,9 +2,10 @@
using Autofac;
using Autofac.Builder;
// ReSharper disable UnusedMember.Global
namespace Tapeti.Autofac
{
/// <inheritdoc />
/// <summary>
/// Dependency resolver and container implementation for Autofac.
/// Since this class needs access to both the ContainerBuilder and the built IContainer,
@ -83,7 +84,7 @@ namespace Tapeti.Autofac
{
CheckContainerBuilder();
containerBuilder
.Register(context => factory())
.Register(_ => factory())
.As<TService>()
.PreserveExistingDefaults();
}
@ -116,7 +117,7 @@ namespace Tapeti.Autofac
{
CheckContainerBuilder();
containerBuilder
.Register(context => factory())
.Register(_ => factory())
.As<TService>()
.SingleInstance()
.PreserveExistingDefaults();

2
Tapeti.Autofac/Tapeti.Autofac.csproj
View File

@ -15,7 +15,7 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Autofac" Version="6.5.0" />
<PackageReference Include="Autofac" Version="6.*" />
</ItemGroup>
<ItemGroup>

2
Tapeti.CastleWindsor/Tapeti.CastleWindsor.csproj
View File

@ -15,7 +15,7 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Castle.Windsor" Version="5.1.2" />
<PackageReference Include="Castle.Windsor" Version="5.*" />
</ItemGroup>
<ItemGroup>

1
Tapeti.CastleWindsor/WindsorDependencyResolver.cs
View File

@ -4,7 +4,6 @@ using Castle.Windsor;
namespace Tapeti.CastleWindsor
{
/// <inheritdoc />
/// <summary>
/// Dependency resolver and container implementation for Castle Windsor.
/// </summary>

1
Tapeti.DataAnnotations/DataAnnotationsExtension.cs
View File

@ -3,7 +3,6 @@ using Tapeti.Config;
namespace Tapeti.DataAnnotations
{
/// <inheritdoc />
/// <summary>
/// Provides the DataAnnotations validation middleware.
/// </summary>

1
Tapeti.DataAnnotations/DataAnnotationsMessageMiddleware.cs
View File

@ -5,7 +5,6 @@ using Tapeti.Config;
namespace Tapeti.DataAnnotations
{
/// <inheritdoc />
/// <summary>
/// Validates consumed messages using System.ComponentModel.DataAnnotations
/// </summary>

1
Tapeti.DataAnnotations/DataAnnotationsPublishMiddleware.cs
View File

@ -5,7 +5,6 @@ using Tapeti.Config;
namespace Tapeti.DataAnnotations
{
/// <inheritdoc />
/// <summary>
/// Validates published messages using System.ComponentModel.DataAnnotations
/// </summary>

1
Tapeti.Flow.SQL/SqlConnectionFlowRepository.cs
View File

@ -11,7 +11,6 @@ using Newtonsoft.Json;
namespace Tapeti.Flow.SQL
{
/// <inheritdoc />
/// <summary>
/// IFlowRepository implementation for SQL server.
/// </summary>

8
Tapeti.Flow.SQL/SqlExceptionHelper.cs
View File

@ -13,7 +13,7 @@ namespace Tapeti.Flow.SQL
// 2627: Violation of %ls constraint '%.*ls'. Cannot insert duplicate key in object '%.*ls'. The duplicate key value is %ls.
public static bool IsDuplicateKey(SqlException e)
{
return e != null && (e.Number == 2601 || e.Number == 2627);
return e is { Number: 2601 or 2627 };
}
@ -21,12 +21,12 @@ namespace Tapeti.Flow.SQL
{
switch (e)
{
case TimeoutException _:
case TimeoutException:
return true;
case Exception exception:
case not null:
{
var sqlExceptions = ExtractSqlExceptions(exception);
var sqlExceptions = ExtractSqlExceptions(e);
return sqlExceptions.Select(UnwrapSqlErrors).Any(IsRecoverableSQLError);
}

1
Tapeti.Flow/Annotations/ContinuationAttribute.cs
View File

@ -2,7 +2,6 @@
namespace Tapeti.Flow.Annotations
{
/// <inheritdoc />
/// <summary>
/// Marks a message handler as a response message handler which continues a Tapeti Flow.
/// The method only receives direct messages, and does not create a routing key based binding to the queue.

1
Tapeti.Flow/Annotations/StartAttribute.cs
View File

@ -3,7 +3,6 @@ using JetBrains.Annotations;
namespace Tapeti.Flow.Annotations
{
/// <inheritdoc />
/// <summary>
/// Marks this method as the start of a Tapeti Flow. Use IFlowStarter.Start to begin a new flow and
/// call this method. Must return an IYieldPoint.

2
Tapeti.Flow/ConfigExtensions.cs
View File

@ -1,5 +1,7 @@
using Tapeti.Config;
// ReSharper disable UnusedMember.Global
namespace Tapeti.Flow
{
/// <summary>

2
Tapeti.Flow/Default/FlowBindingMiddleware.cs
View File

@ -60,7 +60,7 @@ namespace Tapeti.Flow.Default
}
else if (context.Result.Info.ParameterType == typeof(void))
{
context.Result.SetHandler((messageContext, value) => HandleParallelResponse(messageContext));
context.Result.SetHandler((messageContext, _) => HandleParallelResponse(messageContext));
}
else
throw new ArgumentException($"Result type must be IYieldPoint, Task or void in controller {context.Method.DeclaringType?.FullName}, method {context.Method.Name}");

1
Tapeti.Flow/Default/FlowHandlerContext.cs
View File

@ -3,7 +3,6 @@ using Tapeti.Config;
namespace Tapeti.Flow.Default
{
/// <inheritdoc />
/// <summary>
/// Default implementation for IFlowHandlerContext
/// </summary>

26
Tapeti.Flow/Default/FlowProvider.cs
View File

@ -206,7 +206,7 @@ namespace Tapeti.Flow.Default
/// <inheritdoc />
public async ValueTask Execute(IFlowHandlerContext context, IYieldPoint yieldPoint)
{
if (!(yieldPoint is DelegateYieldPoint executableYieldPoint))
if (yieldPoint is not DelegateYieldPoint executableYieldPoint)
throw new YieldPointException($"Yield point is required in controller {context.Controller.GetType().Name} for method {context.Method.Name}");
FlowContext flowContext = null;
@ -297,8 +297,8 @@ namespace Tapeti.Flow.Default
{
private class RequestInfo
{
public object Message { get; set; }
public ResponseHandlerInfo ResponseHandlerInfo { get; set; }
public object Message { get; init; }
public ResponseHandlerInfo ResponseHandlerInfo { get; init; }
}
@ -372,21 +372,13 @@ namespace Tapeti.Flow.Default
{
if (requests.Count == 0)
{
switch (noRequestsBehaviour)
return noRequestsBehaviour switch
{
case FlowNoRequestsBehaviour.Exception:
throw new YieldPointException("At least one request must be added before yielding a parallel request");
case FlowNoRequestsBehaviour.Converge:
return new DelegateYieldPoint(context =>
flowProvider.Converge(context, convergeMethod.Method.Name, convergeMethodSync));
case FlowNoRequestsBehaviour.EndFlow:
return new DelegateYieldPoint(EndFlow);
default:
throw new ArgumentOutOfRangeException(nameof(noRequestsBehaviour), noRequestsBehaviour, null);
}
FlowNoRequestsBehaviour.Exception => throw new YieldPointException("At least one request must be added before yielding a parallel request"),
FlowNoRequestsBehaviour.Converge => new DelegateYieldPoint(context => flowProvider.Converge(context, convergeMethod.Method.Name, convergeMethodSync)),
FlowNoRequestsBehaviour.EndFlow => new DelegateYieldPoint(EndFlow),
_ => throw new ArgumentOutOfRangeException(nameof(noRequestsBehaviour), noRequestsBehaviour, null)
};
}
if (convergeMethod?.Method == null)

1
Tapeti.Flow/Default/FlowStarter.cs
View File

@ -6,7 +6,6 @@ using Tapeti.Config;
namespace Tapeti.Flow.Default
{
/// <inheritdoc />
/// <summary>
/// Default implementation for IFlowStarter.
/// </summary>

4
Tapeti.Flow/Default/FlowState.cs
View File

@ -18,7 +18,7 @@ namespace Tapeti.Flow.Default
/// </summary>
public FlowMetadata Metadata
{
get => metadata ?? (metadata = new FlowMetadata());
get => metadata ??= new FlowMetadata();
set => metadata = value;
}
@ -34,7 +34,7 @@ namespace Tapeti.Flow.Default
/// </summary>
public Dictionary<Guid, ContinuationMetadata> Continuations
{
get => continuations ?? (continuations = new Dictionary<Guid, ContinuationMetadata>());
get => continuations ??= new Dictionary<Guid, ContinuationMetadata>();
set => continuations = value;
}

3
Tapeti.Flow/Default/FlowStore.cs
View File

@ -9,7 +9,6 @@ using Tapeti.Flow.FlowHelpers;
namespace Tapeti.Flow.Default
{
/// <inheritdoc />
/// <summary>
/// Default implementation of IFlowStore.
/// </summary>
@ -119,7 +118,7 @@ namespace Tapeti.Flow.Default
if (!loaded)
throw new InvalidOperationException("Flow store is not yet loaded.");
return new ValueTask<Guid?>(continuationLookup.TryGetValue(continuationID, out var result) ? result : (Guid?)null);
return new ValueTask<Guid?>(continuationLookup.TryGetValue(continuationID, out var result) ? result : null);
}

1
Tapeti.Flow/Default/NonPersistentFlowRepository.cs
View File

@ -5,7 +5,6 @@ using System.Threading.Tasks;
namespace Tapeti.Flow.Default
{
/// <inheritdoc />
/// <summary>
/// Default implementation for IFlowRepository. Does not persist any state, relying on the FlowStore's cache instead.
/// </summary>

1
Tapeti.Flow/FlowExtension.cs
View File

@ -4,7 +4,6 @@ using Tapeti.Flow.Default;
namespace Tapeti.Flow
{
/// <inheritdoc />
/// <summary>
/// Provides the Flow middleware.
/// </summary>

2
Tapeti.Flow/FlowHelpers/MethodSerializer.cs
View File

@ -35,8 +35,6 @@ namespace Tapeti.Flow.FlowHelpers
try
{
assembly = Assembly.Load(match.Groups["assembly"].Value);
if (assembly == null)
return null;
}
catch
{

1
Tapeti.Flow/IFlowHandlerContext.cs
View File

@ -4,7 +4,6 @@ using Tapeti.Config;
namespace Tapeti.Flow
{
/// <inheritdoc />
/// <summary>
/// Provides information about the handler for the flow.
/// </summary>

1
Tapeti.Flow/IFlowStore.cs
View File

@ -42,7 +42,6 @@ namespace Tapeti.Flow
}
/// <inheritdoc />
/// <summary>
/// Represents a lock on the flow state, to provide thread safety.
/// </summary>

1
Tapeti.Flow/ResponseExpectedException.cs
View File

@ -2,7 +2,6 @@
namespace Tapeti.Flow
{
/// <inheritdoc />
/// <summary>
/// Raised when a response is expected to end a flow, but none was provided.
/// </summary>

2
Tapeti.Flow/Tapeti.Flow.csproj
View File

@ -36,6 +36,6 @@
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
<PackageReference Include="Tapeti.Annotations" Version="3.0.0" />
<PackageReference Include="Tapeti.Annotations" Version="3.*-*" />
</ItemGroup>
</Project>

1
Tapeti.Flow/YieldPointException.cs
View File

@ -2,7 +2,6 @@
namespace Tapeti.Flow
{
/// <inheritdoc />
/// <summary>
/// Raised when an invalid yield point is returned.
/// </summary>

5
Tapeti.Ninject/NinjectDependencyResolver.cs
View File

@ -4,7 +4,6 @@ using Ninject;
namespace Tapeti.Ninject
{
/// <inheritdoc />
/// <summary>
/// Dependency resolver and container implementation for Ninject.
/// </summary>
@ -49,7 +48,7 @@ namespace Tapeti.Ninject
if (kernel.GetBindings(typeof(TService)).Any())
return;
kernel.Bind<TService>().ToMethod(context => factory());
kernel.Bind<TService>().ToMethod(_ => factory());
}
@ -77,7 +76,7 @@ namespace Tapeti.Ninject
if (kernel.GetBindings(typeof(TService)).Any())
return;
kernel.Bind<TService>().ToMethod(context => factory()).InSingletonScope();
kernel.Bind<TService>().ToMethod(_ => factory()).InSingletonScope();
}

2
Tapeti.Ninject/Tapeti.Ninject.csproj
View File

@ -15,7 +15,7 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Ninject" Version="3.3.6" />
<PackageReference Include="Ninject" Version="3.*" />
</ItemGroup>
<ItemGroup>

2
Tapeti.Serilog/Tapeti.Serilog.csproj
View File

@ -19,7 +19,7 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Serilog" Version="2.12.0" />
<PackageReference Include="Serilog" Version="2.*" />
</ItemGroup>
<ItemGroup>

3
Tapeti.Serilog/TapetiSeriLogger.cs
View File

@ -2,6 +2,7 @@
using System.Collections.Generic;
using System.Text;
using Tapeti.Config;
using Tapeti.Connection;
using ISerilogLogger = Serilog.ILogger;
// ReSharper disable UnusedMember.Global
@ -129,7 +130,7 @@ namespace Tapeti.Serilog
}
/// <inheritdoc />
public void QueueExistsWarning(string queueName, IReadOnlyDictionary<string, string> existingArguments, IReadOnlyDictionary<string, string> arguments)
public void QueueExistsWarning(string queueName, IRabbitMQArguments existingArguments, IRabbitMQArguments arguments)
{
seriLogger.Warning("Tapeti: durable queue {queueName} exists with incompatible x-arguments ({existingArguments} vs. {arguments}) and will not be redeclared, queue will be consumed as-is",
queueName,

1
Tapeti.SimpleInjector/SimpleInjectorDependencyResolver.cs
View File

@ -4,7 +4,6 @@ using SimpleInjector;
namespace Tapeti.SimpleInjector
{
/// <inheritdoc />
/// <summary>
/// Dependency resolver and container implementation for SimpleInjector.
/// </summary>

2
Tapeti.SimpleInjector/Tapeti.SimpleInjector.csproj
View File

@ -19,7 +19,7 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="SimpleInjector" Version="5.4.1" />
<PackageReference Include="SimpleInjector" Version="5.*" />
</ItemGroup>
<ItemGroup>

59
Tapeti.Tests/Client/TapetiClientTests.cs
View File

@ -1,9 +1,12 @@
// Do not include in the Release build for AppVeyor due to the Docker requirement
#if DEBUG
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using FluentAssertions;
using Tapeti.Connection;
using Tapeti.Default;
using Tapeti.Exceptions;
using Tapeti.Tests.Mock;
using Xunit;
using Xunit.Abstractions;
@ -11,11 +14,14 @@ using Xunit.Abstractions;
namespace Tapeti.Tests.Client
{
[Collection(RabbitMQCollection.Name)]
public class TapetiClientTests
public class TapetiClientTests : IAsyncLifetime
{
private readonly RabbitMQFixture fixture;
private readonly MockDependencyResolver dependencyResolver = new();
private TapetiClient client;
public TapetiClientTests(RabbitMQFixture fixture, ITestOutputHelper testOutputHelper)
{
this.fixture = fixture;
@ -24,6 +30,22 @@ namespace Tapeti.Tests.Client
}
public Task InitializeAsync()
{
client = CreateClient();
return Task.CompletedTask;
}
public async Task DisposeAsync()
{
await client.Close();
client = null;
}
[Fact]
public void Fixture()
{
@ -35,31 +57,50 @@ namespace Tapeti.Tests.Client
[Fact]
public async Task DynamicQueueDeclareNoPrefix()
{
var client = CreateCilent();
var queueName = await client.DynamicQueueDeclare(null, null, CancellationToken.None);
queueName.Should().NotBeNullOrEmpty();
await client.Close();
}
[Fact]
public async Task DynamicQueueDeclarePrefix()
{
var client = CreateCilent();
var queueName = await client.DynamicQueueDeclare("dynamicprefix", null, CancellationToken.None);
queueName.Should().StartWith("dynamicprefix");
}
await client.Close();
[Fact]
public async Task PublishHandleOverflow()
{
var queue1 = await client.DynamicQueueDeclare(null, new RabbitMQArguments
{
{ "x-max-length", 5 },
{ "x-overflow", "reject-publish" }
}, CancellationToken.None);
var queue2 = await client.DynamicQueueDeclare(null, null, CancellationToken.None);
var body = Encoding.UTF8.GetBytes("Hello world!");
var properties = new MessageProperties();
for (var i = 0; i < 5; i++)
await client.Publish(body, properties, null, queue1, true);
var publishOverMaxLength = () => client.Publish(body, properties, null, queue1, true);
await publishOverMaxLength.Should().ThrowAsync<NackException>();
// The channel should recover and allow further publishing
await client.Publish(body, properties, null, queue2, true);
}
// TODO test the other methods
private TapetiClient CreateCilent()
private TapetiClient CreateClient()
{
return new TapetiClient(
new TapetiConfig.Config(dependencyResolver),

24
Tapeti.Tests/Config/QueueArgumentsTest.cs
View File

@ -18,7 +18,7 @@ namespace Tapeti.Tests.Config
private static readonly MockRepository MoqRepository = new(MockBehavior.Strict);
private readonly Mock<ITapetiClient> client;
private readonly Dictionary<string, IReadOnlyDictionary<string, string>> declaredQueues = new();
private readonly Dictionary<string, IRabbitMQArguments> declaredQueues = new();
public QueueArgumentsTest()
@ -45,8 +45,8 @@ namespace Tapeti.Tests.Config
var queue = 0;
client
.Setup(c => c.DynamicQueueDeclare(null, It.IsAny<IReadOnlyDictionary<string, string>>(), It.IsAny<CancellationToken>()))
.Callback((string _, IReadOnlyDictionary<string, string> arguments, CancellationToken _) =>
.Setup(c => c.DynamicQueueDeclare(null, It.IsAny<IRabbitMQArguments>(), It.IsAny<CancellationToken>()))
.Callback((string _, IRabbitMQArguments arguments, CancellationToken _) =>
{
queue++;
declaredQueues.Add($"queue-{queue}", arguments);
@ -54,8 +54,8 @@ namespace Tapeti.Tests.Config
.ReturnsAsync(() => $"queue-{queue}");
client
.Setup(c => c.DurableQueueDeclare(It.IsAny<string>(), It.IsAny<IEnumerable<QueueBinding>>(), It.IsAny<IReadOnlyDictionary<string, string>>(), It.IsAny<CancellationToken>()))
.Callback((string queueName, IEnumerable<QueueBinding> _, IReadOnlyDictionary<string, string> arguments, CancellationToken _) =>
.Setup(c => c.DurableQueueDeclare(It.IsAny<string>(), It.IsAny<IEnumerable<QueueBinding>>(), It.IsAny<IRabbitMQArguments>(), It.IsAny<CancellationToken>()))
.Callback((string queueName, IEnumerable<QueueBinding> _, IRabbitMQArguments arguments, CancellationToken _) =>
{
declaredQueues.Add(queueName, arguments);
})
@ -89,10 +89,10 @@ namespace Tapeti.Tests.Config
var arguments = declaredQueues["queue-1"];
arguments.Should().ContainKey("x-custom").WhoseValue.Should().Be("custom value");
arguments.Should().ContainKey("x-another").WhoseValue.Should().Be("another value");
arguments.Should().ContainKey("x-max-length").WhoseValue.Should().Be("100");
arguments.Should().ContainKey("x-max-length-bytes").WhoseValue.Should().Be("100000");
arguments.Should().ContainKey("x-message-ttl").WhoseValue.Should().Be("4269");
arguments.Should().ContainKey("x-another").WhoseValue.Should().Be(true);
arguments.Should().ContainKey("x-max-length").WhoseValue.Should().Be(100);
arguments.Should().ContainKey("x-max-length-bytes").WhoseValue.Should().Be(100000);
arguments.Should().ContainKey("x-message-ttl").WhoseValue.Should().Be(4269);
arguments.Should().ContainKey("x-overflow").WhoseValue.Should().Be("reject-publish");
}
@ -108,10 +108,10 @@ namespace Tapeti.Tests.Config
declaredQueues.Should().HaveCount(2);
var arguments1 = declaredQueues["queue-1"];
arguments1.Should().ContainKey("x-max-length").WhoseValue.Should().Be("100");
arguments1.Should().ContainKey("x-max-length").WhoseValue.Should().Be(100);
var arguments2 = declaredQueues["queue-2"];
arguments2.Should().ContainKey("x-max-length-bytes").WhoseValue.Should().Be("100000");
arguments2.Should().ContainKey("x-max-length-bytes").WhoseValue.Should().Be(100000);
}
@ -148,7 +148,7 @@ namespace Tapeti.Tests.Config
[DynamicQueue]
[QueueArguments("x-custom", "custom value", "x-another", "another value", MaxLength = 100, MaxLengthBytes = 100000, MessageTTL = 4269, Overflow = RabbitMQOverflow.RejectPublish)]
[QueueArguments("x-custom", "custom value", "x-another", true, MaxLength = 100, MaxLengthBytes = 100000, MessageTTL = 4269, Overflow = RabbitMQOverflow.RejectPublish)]
private class TestController
{
public void HandleMessage1(TestMessage1 message)

4
Tapeti.Tests/Mock/MockLogger.cs
View File

@ -1,7 +1,7 @@
using System;
using System.Collections.Generic;
using System.Text;
using Tapeti.Config;
using Tapeti.Connection;
using Xunit.Abstractions;
namespace Tapeti.Tests.Mock
@ -49,7 +49,7 @@ namespace Tapeti.Tests.Mock
: $"Declaring {(durable ? "durable" : "dynamic")} queue {queueName}");
}
public void QueueExistsWarning(string queueName, IReadOnlyDictionary<string, string> existingArguments, IReadOnlyDictionary<string, string> arguments)
public void QueueExistsWarning(string queueName, IRabbitMQArguments existingArguments, IRabbitMQArguments arguments)
{
var argumentsText = new StringBuilder();
foreach (var pair in arguments)

2
Tapeti.Transient/ConfigExtensions.cs
View File

@ -1,6 +1,8 @@
using System;
using Tapeti.Config;
// ReSharper disable UnusedMember.Global
namespace Tapeti.Transient
{
/// <summary>

1
Tapeti.Transient/TransientGenericBinding.cs
View File

@ -4,7 +4,6 @@ using Tapeti.Config;
namespace Tapeti.Transient
{
/// <inheritdoc />
/// <summary>
/// Implements a binding for transient request response messages.
/// Register this binding using the WithTransient config extension method.

1
Tapeti.Transient/TransientPublisher.cs
View File

@ -2,7 +2,6 @@
namespace Tapeti.Transient
{
/// <inheritdoc />
/// <summary>
/// Default implementation of ITransientPublisher
/// </summary>

4
Tapeti.Transient/TransientRouter.cs
View File

@ -55,7 +55,7 @@ namespace Tapeti.Transient
public async Task<object> RequestResponse(IPublisher publisher, object request)
{
var correlation = Guid.NewGuid();
var tcs = map.GetOrAdd(correlation, c => new TaskCompletionSource<object>());
var tcs = map.GetOrAdd(correlation, _ => new TaskCompletionSource<object>());
try
{
@ -77,7 +77,7 @@ namespace Tapeti.Transient
throw;
}
using (new Timer(TimeoutResponse, tcs, defaultTimeoutMs, -1))
await using (new Timer(TimeoutResponse, tcs, defaultTimeoutMs, -1))
{
return await tcs.Task;
}

35
Tapeti.UnityContainer/Tapeti.UnityContainer.csproj
View File

@ -1,35 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net6.0;net7.0</TargetFrameworks>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<Authors>Mark van Renswoude</Authors>
<Company />
<Description>Unity container integration package for Tapeti</Description>
<PackageTags>rabbitmq tapeti unity</PackageTags>
<PackageLicenseExpression>Unlicense</PackageLicenseExpression>
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.SimpleInjector.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Unity" Version="5.11.10" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Tapeti\Tapeti.csproj" />
</ItemGroup>
<ItemGroup>
<None Include="..\resources\icons\Tapeti.SimpleInjector.png">
<Pack>True</Pack>
<PackagePath></PackagePath>
</None>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All" />
</ItemGroup>
</Project>

90
Tapeti.UnityContainer/UnityDependencyResolver.cs
View File

@ -1,90 +0,0 @@
using System;
using Unity;
using Unity.Lifetime;
namespace Tapeti.UnityContainer
{
/// <inheritdoc />
/// <summary>
/// Dependency resolver and container implementation for SimpleInjector.
/// </summary>
public class UnityDependencyResolver : IDependencyContainer
{
private readonly IUnityContainer container;
/// <summary>
/// </summary>
public UnityDependencyResolver(IUnityContainer container)
{
this.container = container;
}
/// <inheritdoc />
public T Resolve<T>() where T : class
{
return container.Resolve<T>();
}
/// <inheritdoc />
public object Resolve(Type type)
{
return container.Resolve(type);
}
/// <inheritdoc />
public void RegisterDefault<TService, TImplementation>() where TService : class where TImplementation : class, TService
{
if (container.IsRegistered(typeof(TService)))
return;
container.RegisterType<TService, TImplementation>();
}
/// <inheritdoc />
public void RegisterDefault<TService>(Func<TService> factory) where TService : class
{
if (container.IsRegistered(typeof(TService)))
return;
container.RegisterFactory<TService>(c => factory());
}
/// <inheritdoc />
public void RegisterDefaultSingleton<TService, TImplementation>() where TService : class where TImplementation : class, TService
{
if (container.IsRegistered(typeof(TService)))
return;
container.RegisterSingleton<TService, TImplementation>();
}
/// <inheritdoc />
public void RegisterDefaultSingleton<TService>(TService instance) where TService : class
{
if (container.IsRegistered(typeof(TService)))
return;
container.RegisterInstance(instance);
}
/// <inheritdoc />
public void RegisterDefaultSingleton<TService>(Func<TService> factory) where TService : class
{
if (container.IsRegistered(typeof(TService)))
return;
container.RegisterFactory<TService>(c => factory(), new SingletonLifetimeManager());
}
/// <inheritdoc />
public void RegisterController(Type type)
{
container.RegisterType(type);
}
}
}

9
Tapeti.sln
View File

@ -45,8 +45,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.CastleWindsor", "Tap
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Autofac", "Tapeti.Autofac\Tapeti.Autofac.csproj", "{B3802005-C941-41B6-A9A5-20573A7C24AE}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.UnityContainer", "Tapeti.UnityContainer\Tapeti.UnityContainer.csproj", "{BA8CA9A2-BAFF-42BB-8439-3DD9D1F6C32E}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Ninject", "Tapeti.Ninject\Tapeti.Ninject.csproj", "{29478B10-FC53-4E93-ADEF-A775D9408131}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "06-StatelessRequestResponse", "Examples\06-StatelessRequestResponse\06-StatelessRequestResponse.csproj", "{152227AA-3165-4550-8997-6EA80C84516E}"
@ -55,7 +53,7 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "07-ParallelizationTest", "E
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "08-MessageHandlerLogging", "Examples\08-MessageHandlerLogging\08-MessageHandlerLogging.csproj", "{906605A6-2CAB-4B29-B0DD-B735BF265E39}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Tapeti.Benchmarks", "Tapeti.Benchmarks\Tapeti.Benchmarks.csproj", "{DBE56131-9207-4CEA-BA3E-031351677C48}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Benchmarks", "Tapeti.Benchmarks\Tapeti.Benchmarks.csproj", "{DBE56131-9207-4CEA-BA3E-031351677C48}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
@ -131,10 +129,6 @@ Global
{B3802005-C941-41B6-A9A5-20573A7C24AE}.Debug|Any CPU.Build.0 = Debug|Any CPU
{B3802005-C941-41B6-A9A5-20573A7C24AE}.Release|Any CPU.ActiveCfg = Release|Any CPU
{B3802005-C941-41B6-A9A5-20573A7C24AE}.Release|Any CPU.Build.0 = Release|Any CPU
{BA8CA9A2-BAFF-42BB-8439-3DD9D1F6C32E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{BA8CA9A2-BAFF-42BB-8439-3DD9D1F6C32E}.Debug|Any CPU.Build.0 = Debug|Any CPU
{BA8CA9A2-BAFF-42BB-8439-3DD9D1F6C32E}.Release|Any CPU.ActiveCfg = Release|Any CPU
{BA8CA9A2-BAFF-42BB-8439-3DD9D1F6C32E}.Release|Any CPU.Build.0 = Release|Any CPU
{29478B10-FC53-4E93-ADEF-A775D9408131}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{29478B10-FC53-4E93-ADEF-A775D9408131}.Debug|Any CPU.Build.0 = Debug|Any CPU
{29478B10-FC53-4E93-ADEF-A775D9408131}.Release|Any CPU.ActiveCfg = Release|Any CPU
@ -176,7 +170,6 @@ Global
{330D05CE-5321-4C7D-8017-2070B891289E} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}
{374AAE64-598B-4F67-8870-4A05168FF987} = {99380F97-AD1A-459F-8AB3-D404E1E6AD4F}
{B3802005-C941-41B6-A9A5-20573A7C24AE} = {99380F97-AD1A-459F-8AB3-D404E1E6AD4F}
{BA8CA9A2-BAFF-42BB-8439-3DD9D1F6C32E} = {99380F97-AD1A-459F-8AB3-D404E1E6AD4F}
{29478B10-FC53-4E93-ADEF-A775D9408131} = {99380F97-AD1A-459F-8AB3-D404E1E6AD4F}
{152227AA-3165-4550-8997-6EA80C84516E} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}
{E69E6BA5-68E7-4A4D-A38C-B2526AA66E96} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}

2
Tapeti.sln.DotSettings
View File

@ -4,7 +4,9 @@
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=ID/@EntryIndexedValue">ID</s:String>
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=JSON/@EntryIndexedValue">JSON</s:String>
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=KV/@EntryIndexedValue">KV</s:String>
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=MQ/@EntryIndexedValue">MQ</s:String>
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=SQL/@EntryIndexedValue">SQL</s:String>
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=UTF/@EntryIndexedValue">UTF</s:String>
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/PredefinedNamingRules/=PrivateInstanceFields/@EntryIndexedValue">&lt;Policy Inspect="True" Prefix="" Suffix="" Style="aaBb" /&gt;</s:String>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>

1
Tapeti/Config/ControllerMessageContextPayload.cs
View File

@ -1,6 +1,5 @@
namespace Tapeti.Config
{
/// <inheritdoc />
/// <summary>
/// Extends the message context with information about the controller.
/// </summary>

12
Tapeti/Config/IBinding.cs
View File

@ -1,6 +1,6 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Tapeti.Connection;
namespace Tapeti.Config
{
@ -82,7 +82,7 @@ namespace Tapeti.Config
/// <param name="messageClass">The message class to be bound to the queue</param>
/// <param name="queueName">The name of the durable queue</param>
/// <param name="arguments">Optional arguments</param>
ValueTask BindDurable(Type messageClass, string queueName, IReadOnlyDictionary<string, string> arguments);
ValueTask BindDurable(Type messageClass, string queueName, IRabbitMQArguments arguments);
/// <summary>
/// Binds the messageClass to a dynamic auto-delete queue.
@ -95,7 +95,7 @@ namespace Tapeti.Config
/// <param name="queuePrefix">An optional prefix for the dynamic queue's name. If not provided, RabbitMQ's default logic will be used to create an amq.gen queue.</param>
/// <param name="arguments">Optional arguments</param>
/// <returns>The generated name of the dynamic queue</returns>
ValueTask<string> BindDynamic(Type messageClass, string queuePrefix, IReadOnlyDictionary<string, string> arguments);
ValueTask<string> BindDynamic(Type messageClass, string queuePrefix, IRabbitMQArguments arguments);
/// <summary>
/// Declares a durable queue but does not add a binding for a messageClass' routing key.
@ -103,7 +103,7 @@ namespace Tapeti.Config
/// </summary>
/// <param name="queueName">The name of the durable queue</param>
/// <param name="arguments">Optional arguments</param>
ValueTask BindDurableDirect(string queueName, IReadOnlyDictionary<string, string> arguments);
ValueTask BindDurableDirect(string queueName, IRabbitMQArguments arguments);
/// <summary>
/// Declares a dynamic queue but does not add a binding for a messageClass' routing key.
@ -113,7 +113,7 @@ namespace Tapeti.Config
/// <param name="queuePrefix">An optional prefix for the dynamic queue's name. If not provided, RabbitMQ's default logic will be used to create an amq.gen queue.</param>
/// <param name="arguments">Optional arguments</param>
/// <returns>The generated name of the dynamic queue</returns>
ValueTask<string> BindDynamicDirect(Type messageClass, string queuePrefix, IReadOnlyDictionary<string, string> arguments);
ValueTask<string> BindDynamicDirect(Type messageClass, string queuePrefix, IRabbitMQArguments arguments);
/// <summary>
/// Declares a dynamic queue but does not add a binding for a messageClass' routing key.
@ -122,7 +122,7 @@ namespace Tapeti.Config
/// <param name="queuePrefix">An optional prefix for the dynamic queue's name. If not provided, RabbitMQ's default logic will be used to create an amq.gen queue.</param>
/// <param name="arguments">Optional arguments</param>
/// <returns>The generated name of the dynamic queue</returns>
ValueTask<string> BindDynamicDirect(string queuePrefix, IReadOnlyDictionary<string, string> arguments);
ValueTask<string> BindDynamicDirect(string queuePrefix, IRabbitMQArguments arguments);
/// <summary>
/// Marks the specified durable queue as having an obsolete binding. If after all bindings have subscribed, the queue only contains obsolete

1
Tapeti/Config/IControllerBindingMiddleware.cs
View File

@ -2,7 +2,6 @@
namespace Tapeti.Config
{
/// <inheritdoc />
/// <summary>
/// Called when a Controller method is registered.
/// </summary>

1
Tapeti/Config/IControllerFilterMiddleware.cs
View File

@ -3,7 +3,6 @@ using System.Threading.Tasks;
namespace Tapeti.Config
{
/// <inheritdoc />
/// <summary>
/// Denotes middleware that runs before the controller is instantiated.
/// </summary>

1
Tapeti/Config/IControllerMethodBinding.cs
View File

@ -3,7 +3,6 @@ using System.Reflection;
namespace Tapeti.Config
{
/// <inheritdoc />
/// <summary>
/// Represents a binding to a method in a controller class to handle incoming messages.
/// </summary>

1
Tapeti/Config/ITapetiConfig.cs
View File

@ -77,7 +77,6 @@ namespace Tapeti.Config
}
/// <inheritdoc />
/// <summary>
/// Contains a list of registered bindings, with a few added helpers.
/// </summary>

1
Tapeti/Config/ITapetiExtensionBinding.cs
View File

@ -2,7 +2,6 @@
namespace Tapeti.Config
{
/// <inheritdoc />
/// <summary>
/// Provides a way for Tapeti extensions to register custom bindings.
/// </summary>

19
Tapeti/Connection/ITapetiClient.cs
View File

@ -6,7 +6,6 @@ using Tapeti.Config;
namespace Tapeti.Connection
{
/// <inheritdoc cref="IEquatable{T}" />
/// <summary>
/// Defines a queue binding to an exchange using a routing key
/// </summary>
@ -52,6 +51,18 @@ namespace Tapeti.Connection
return ((Exchange != null ? Exchange.GetHashCode() : 0) * 397) ^ (RoutingKey != null ? RoutingKey.GetHashCode() : 0);
}
}
/// <summary></summary>
public static bool operator ==(QueueBinding left, QueueBinding right)
{
return left.Equals(right);
}
/// <summary></summary>
public static bool operator !=(QueueBinding left, QueueBinding right)
{
return !left.Equals(right);
}
}
@ -93,7 +104,7 @@ namespace Tapeti.Connection
/// <param name="bindings">A list of bindings. Any bindings already on the queue which are not in this list will be removed</param>
/// <param name="arguments">Optional arguments</param>
/// <param name="cancellationToken">Cancelled when the connection is lost</param>
Task DurableQueueDeclare(string queueName, IEnumerable<QueueBinding> bindings, IReadOnlyDictionary<string, string> arguments, CancellationToken cancellationToken);
Task DurableQueueDeclare(string queueName, IEnumerable<QueueBinding> bindings, IRabbitMQArguments arguments, CancellationToken cancellationToken);
/// <summary>
/// Verifies a durable queue exists. Will raise an exception if it does not.
@ -101,7 +112,7 @@ namespace Tapeti.Connection
/// <param name="queueName">The name of the queue to verify</param>
/// <param name="arguments">Optional arguments</param>
/// <param name="cancellationToken">Cancelled when the connection is lost</param>
Task DurableQueueVerify(string queueName, IReadOnlyDictionary<string, string> arguments, CancellationToken cancellationToken);
Task DurableQueueVerify(string queueName, IRabbitMQArguments arguments, CancellationToken cancellationToken);
/// <summary>
/// Deletes a durable queue.
@ -117,7 +128,7 @@ namespace Tapeti.Connection
/// <param name="queuePrefix">An optional prefix for the dynamic queue's name. If not provided, RabbitMQ's default logic will be used to create an amq.gen queue.</param>
/// <param name="arguments">Optional arguments</param>
/// <param name="cancellationToken">Cancelled when the connection is lost</param>
Task<string> DynamicQueueDeclare(string queuePrefix, IReadOnlyDictionary<string, string> arguments, CancellationToken cancellationToken);
Task<string> DynamicQueueDeclare(string queuePrefix, IRabbitMQArguments arguments, CancellationToken cancellationToken);
/// <summary>
/// Add a binding to a dynamic queue.

36
Tapeti/Connection/RabbitMQArguments.cs Normal file
View File

@ -0,0 +1,36 @@
using System.Collections.Generic;
using System.Text;
namespace Tapeti.Connection
{
/// <inheritdoc />
public interface IRabbitMQArguments : IReadOnlyDictionary<string, object>
{
}
internal class RabbitMQArguments : Dictionary<string, object>, IRabbitMQArguments
{
public RabbitMQArguments()
{
}
#if NETSTANDARD2_1_OR_GREATER
public RabbitMQArguments(IReadOnlyDictionary<string, object> values) : base(values)
{
}
#else
public RabbitMQArguments(IReadOnlyDictionary<string, object> values)
{
foreach (var pair in values)
Add(pair.Key, pair.Value);
}
#endif
public void AddUTF8(string key, string value)
{
Add(key, Encoding.UTF8.GetBytes(value));
}
}
}

1
Tapeti/Connection/TapetiBasicConsumer.cs
View File

@ -14,7 +14,6 @@ namespace Tapeti.Connection
public delegate Task ResponseFunc(long expectedConnectionReference, ulong deliveryTag, ConsumeResult result);
/// <inheritdoc />
/// <summary>
/// Implements the bridge between the RabbitMQ Client consumer and a Tapeti Consumer
/// </summary>

73
Tapeti/Connection/TapetiClient.cs
View File

@ -7,6 +7,7 @@ using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
@ -24,7 +25,6 @@ namespace Tapeti.Connection
}
/// <inheritdoc />
/// <summary>
/// Implementation of ITapetiClient for the RabbitMQ Client library
/// </summary>
@ -185,15 +185,18 @@ namespace Tapeti.Connection
var replyCode = publishResultTask.Result;
// There is no RabbitMQ.Client.Framing.Constants value for this "No route" reply code
// at the time of writing...
if (replyCode == 312)
throw new NoRouteException(
$"Mandatory message with exchange '{exchange}' and routing key '{routingKey}' does not have a route");
switch (replyCode)
{
// There is no RabbitMQ.Client.Framing.Constants value for this "No route" reply code
// at the time of writing...
case 312:
throw new NoRouteException(
$"Mandatory message with exchange '{exchange}' and routing key '{routingKey}' does not have a route");
if (replyCode > 0)
throw new NoRouteException(
$"Mandatory message with exchange '{exchange}' and routing key '{routingKey}' could not be delivered, reply code: {replyCode}");
case > 0:
throw new NoRouteException(
$"Mandatory message with exchange '{exchange}' and routing key '{routingKey}' could not be delivered, reply code: {replyCode}");
}
});
}
@ -286,7 +289,7 @@ namespace Tapeti.Connection
}
private async Task<bool> GetDurableQueueDeclareRequired(string queueName, IReadOnlyDictionary<string, string> arguments)
private async Task<bool> GetDurableQueueDeclareRequired(string queueName, IRabbitMQArguments arguments)
{
var existingQueue = await GetQueueInfo(queueName);
if (existingQueue == null)
@ -298,17 +301,44 @@ namespace Tapeti.Connection
if (arguments == null && existingQueue.Arguments.Count == 0)
return true;
if (existingQueue.Arguments.NullSafeSameValues(arguments))
var existingArguments = ConvertJsonArguments(existingQueue.Arguments);
if (existingArguments.NullSafeSameValues(arguments))
return true;
(logger as IBindingLogger)?.QueueExistsWarning(queueName, existingQueue.Arguments, arguments);
(logger as IBindingLogger)?.QueueExistsWarning(queueName, existingArguments, arguments);
return false;
}
private static RabbitMQArguments ConvertJsonArguments(IReadOnlyDictionary<string, JObject> arguments)
{
if (arguments == null)
return null;
var result = new RabbitMQArguments();
foreach (var pair in arguments)
{
// ReSharper disable once SwitchExpressionHandlesSomeKnownEnumValuesWithExceptionInDefault - by design
object value = pair.Value.Type switch
{
JTokenType.Integer => pair.Value.Value<int>(),
JTokenType.Float => pair.Value.Value<double>(),
JTokenType.String => Encoding.UTF8.GetBytes(pair.Value.Value<string>() ?? string.Empty),
JTokenType.Boolean => pair.Value.Value<bool>(),
JTokenType.Null => null,
_ => throw new ArgumentOutOfRangeException(nameof(arguments))
};
result.Add(pair.Key, value);
}
return result;
}
/// <inheritdoc />
public async Task DurableQueueDeclare(string queueName, IEnumerable<QueueBinding> bindings, IReadOnlyDictionary<string, string> arguments, CancellationToken cancellationToken)
public async Task DurableQueueDeclare(string queueName, IEnumerable<QueueBinding> bindings, IRabbitMQArguments arguments, CancellationToken cancellationToken)
{
var declareRequired = await GetDurableQueueDeclareRequired(queueName, arguments);
@ -343,17 +373,16 @@ namespace Tapeti.Connection
}
private static IDictionary<string, object> GetDeclareArguments(IReadOnlyDictionary<string, string> arguments)
private static IDictionary<string, object> GetDeclareArguments(IRabbitMQArguments arguments)
{
if (arguments == null || arguments.Count == 0)
return null;
return arguments.ToDictionary(p => p.Key, p => (object)Encoding.UTF8.GetBytes(p.Value));
return arguments == null || arguments.Count == 0
? null
: arguments.ToDictionary(p => p.Key, p => p.Value);
}
/// <inheritdoc />
public async Task DurableQueueVerify(string queueName, IReadOnlyDictionary<string, string> arguments, CancellationToken cancellationToken)
public async Task DurableQueueVerify(string queueName, IRabbitMQArguments arguments, CancellationToken cancellationToken)
{
if (!await GetDurableQueueDeclareRequired(queueName, arguments))
return;
@ -455,7 +484,7 @@ namespace Tapeti.Connection
/// <inheritdoc />
public async Task<string> DynamicQueueDeclare(string queuePrefix, IReadOnlyDictionary<string, string> arguments, CancellationToken cancellationToken)
public async Task<string> DynamicQueueDeclare(string queuePrefix, IRabbitMQArguments arguments, CancellationToken cancellationToken)
{
string queueName = null;
var bindingLogger = logger as IBindingLogger;
@ -564,7 +593,7 @@ namespace Tapeti.Connection
public bool Exclusive { get; set; }
[JsonProperty("arguments")]
public Dictionary<string, string> Arguments { get; set; }
public Dictionary<string, JObject> Arguments { get; set; }
[JsonProperty("messages")]
public uint Messages { get; set; }
@ -675,7 +704,7 @@ namespace Tapeti.Connection
}
catch (WebException e)
{
if (!(e.Response is HttpWebResponse response))
if (e.Response is not HttpWebResponse response)
throw;
if (!TransientStatusCodes.Contains(response.StatusCode))

4
Tapeti/Connection/TapetiConsumer.cs
View File

@ -9,7 +9,6 @@ using Tapeti.Helpers;
namespace Tapeti.Connection
{
/// <inheritdoc />
/// <summary>
/// Implements a RabbitMQ consumer to pass messages to the Tapeti middleware.
/// </summary>
@ -172,8 +171,7 @@ namespace Tapeti.Connection
return e switch
{
AggregateException aggregateException => aggregateException.InnerExceptions.Any(IgnoreExceptionDuringShutdown),
TaskCanceledException _ => true,
OperationCanceledException _ => true,
OperationCanceledException => true,
_ => e.InnerException != null && IgnoreExceptionDuringShutdown(e.InnerException)
};
}

8
Tapeti/Connection/TapetiPublisher.cs
View File

@ -151,11 +151,11 @@ namespace Tapeti.Connection
private class PublishContext : IPublishContext
{
public ITapetiConfig Config { get; set; }
public ITapetiConfig Config { get; init; }
public string Exchange { get; set; }
public string RoutingKey { get; set; }
public object Message { get; set; }
public IMessageProperties Properties { get; set; }
public string RoutingKey { get; init; }
public object Message { get; init; }
public IMessageProperties Properties { get; init; }
}
}
}

34
Tapeti/Connection/TapetiSubscriber.cs
View File

@ -72,14 +72,12 @@ namespace Tapeti.Connection
/// </summary>
public void Reconnect()
{
CancellationToken cancellationToken;
initializeCancellationTokenSource?.Cancel();
initializeCancellationTokenSource = new CancellationTokenSource();
consumerTags.Clear();
cancellationToken = initializeCancellationTokenSource.Token;
var cancellationToken = initializeCancellationTokenSource.Token;
Task.Run(async () =>
{
@ -166,7 +164,7 @@ namespace Tapeti.Connection
{
public string QueueName;
public List<Type> MessageClasses;
public IReadOnlyDictionary<string, string> Arguments;
public IRabbitMQArguments Arguments;
}
private readonly Dictionary<string, List<DynamicQueueInfo>> dynamicQueues = new();
@ -187,12 +185,12 @@ namespace Tapeti.Connection
}
public abstract ValueTask BindDurable(Type messageClass, string queueName, IReadOnlyDictionary<string, string> arguments);
public abstract ValueTask BindDurableDirect(string queueName, IReadOnlyDictionary<string, string> arguments);
public abstract ValueTask BindDurable(Type messageClass, string queueName, IRabbitMQArguments arguments);
public abstract ValueTask BindDurableDirect(string queueName, IRabbitMQArguments arguments);
public abstract ValueTask BindDurableObsolete(string queueName);
public async ValueTask<string> BindDynamic(Type messageClass, string queuePrefix, IReadOnlyDictionary<string, string> arguments)
public async ValueTask<string> BindDynamic(Type messageClass, string queuePrefix, IRabbitMQArguments arguments)
{
var result = await DeclareDynamicQueue(messageClass, queuePrefix, arguments);
if (!result.IsNewMessageClass)
@ -207,14 +205,14 @@ namespace Tapeti.Connection
}
public async ValueTask<string> BindDynamicDirect(Type messageClass, string queuePrefix, IReadOnlyDictionary<string, string> arguments)
public async ValueTask<string> BindDynamicDirect(Type messageClass, string queuePrefix, IRabbitMQArguments arguments)
{
var result = await DeclareDynamicQueue(messageClass, queuePrefix, arguments);
return result.QueueName;
}
public async ValueTask<string> BindDynamicDirect(string queuePrefix, IReadOnlyDictionary<string, string> arguments)
public async ValueTask<string> BindDynamicDirect(string queuePrefix, IRabbitMQArguments arguments)
{
// If we don't know the routing key, always create a new queue to ensure there is no overlap.
// Keep it out of the dynamicQueues dictionary, so it can't be re-used later on either.
@ -228,7 +226,7 @@ namespace Tapeti.Connection
public bool IsNewMessageClass;
}
private async Task<DeclareDynamicQueueResult> DeclareDynamicQueue(Type messageClass, string queuePrefix, IReadOnlyDictionary<string, string> arguments)
private async Task<DeclareDynamicQueueResult> DeclareDynamicQueue(Type messageClass, string queuePrefix, IRabbitMQArguments arguments)
{
// Group by prefix
var key = queuePrefix ?? "";
@ -284,7 +282,7 @@ namespace Tapeti.Connection
private struct DurableQueueInfo
{
public List<Type> MessageClasses;
public IReadOnlyDictionary<string, string> Arguments;
public IRabbitMQArguments Arguments;
}
@ -297,7 +295,7 @@ namespace Tapeti.Connection
}
public override ValueTask BindDurable(Type messageClass, string queueName, IReadOnlyDictionary<string, string> arguments)
public override ValueTask BindDurable(Type messageClass, string queueName, IRabbitMQArguments arguments)
{
// Collect the message classes per queue so we can determine afterwards
// if any of the bindings currently set on the durable queue are no
@ -326,7 +324,7 @@ namespace Tapeti.Connection
}
public override ValueTask BindDurableDirect(string queueName, IReadOnlyDictionary<string, string> arguments)
public override ValueTask BindDurableDirect(string queueName, IRabbitMQArguments arguments)
{
if (!durableQueues.TryGetValue(queueName, out var durableQueueInfo))
{
@ -398,12 +396,12 @@ namespace Tapeti.Connection
}
public override async ValueTask BindDurable(Type messageClass, string queueName, IReadOnlyDictionary<string, string> arguments)
public override async ValueTask BindDurable(Type messageClass, string queueName, IRabbitMQArguments arguments)
{
await VerifyDurableQueue(queueName, arguments);
}
public override async ValueTask BindDurableDirect(string queueName, IReadOnlyDictionary<string, string> arguments)
public override async ValueTask BindDurableDirect(string queueName, IRabbitMQArguments arguments)
{
await VerifyDurableQueue(queueName, arguments);
}
@ -414,7 +412,7 @@ namespace Tapeti.Connection
}
private async Task VerifyDurableQueue(string queueName, IReadOnlyDictionary<string, string> arguments)
private async Task VerifyDurableQueue(string queueName, IRabbitMQArguments arguments)
{
if (!durableQueues.Add(queueName))
return;
@ -431,12 +429,12 @@ namespace Tapeti.Connection
}
public override ValueTask BindDurable(Type messageClass, string queueName, IReadOnlyDictionary<string, string> arguments)
public override ValueTask BindDurable(Type messageClass, string queueName, IRabbitMQArguments arguments)
{
return default;
}
public override ValueTask BindDurableDirect(string queueName, IReadOnlyDictionary<string, string> arguments)
public override ValueTask BindDurableDirect(string queueName, IRabbitMQArguments arguments)
{
return default;
}

1
Tapeti/Default/CancellationTokenBinding.cs
View File

@ -5,7 +5,6 @@ using Tapeti.Config;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Binds a parameter of type CancellationToken to a token which is cancelled when the RabbitMQ connection is closed.
/// Similar to and very much inspired by ASP.NET's RequestAborted CancellationToken.

7
Tapeti/Default/ConsoleLogger.cs
View File

@ -1,13 +1,12 @@
using System;
using System.Collections.Generic;
using System.Text;
using Tapeti.Config;
using Tapeti.Connection;
// ReSharper disable UnusedMember.Global - public API
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Default ILogger implementation for console applications.
/// </summary>
@ -81,13 +80,13 @@ namespace Tapeti.Default
}
/// <inheritdoc />
public void QueueExistsWarning(string queueName, IReadOnlyDictionary<string, string> existingArguments, IReadOnlyDictionary<string, string> arguments)
public void QueueExistsWarning(string queueName, IRabbitMQArguments existingArguments, IRabbitMQArguments arguments)
{
Console.WriteLine($"[Tapeti] Durable queue {queueName} exists with incompatible x-arguments ({GetArgumentsText(existingArguments)} vs. {GetArgumentsText(arguments)}) and will not be redeclared, queue will be consumed as-is");
}
private static string GetArgumentsText(IReadOnlyDictionary<string, string> arguments)
private static string GetArgumentsText(IRabbitMQArguments arguments)
{
var argumentsText = new StringBuilder();
foreach (var pair in arguments)

2
Tapeti/Default/ControllerBindingContext.cs
View File

@ -99,7 +99,6 @@ namespace Tapeti.Default
}
/// <inheritdoc />
/// <summary>
/// Default implementation for IBindingParameter
/// </summary>
@ -139,7 +138,6 @@ namespace Tapeti.Default
}
/// <inheritdoc />
/// <summary>
/// Default implementation for IBindingResult
/// </summary>

4
Tapeti/Default/ControllerMethodBinding.cs
View File

@ -4,11 +4,11 @@ using System.Linq;
using System.Reflection;
using System.Threading.Tasks;
using Tapeti.Config;
using Tapeti.Connection;
using Tapeti.Helpers;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Binding implementation for controller methods. Do not instantiate this class yourself,
/// instead use the ITapetiConfigBuilder RegisterController / RegisterAllControllers extension
@ -319,7 +319,7 @@ namespace Tapeti.Default
/// <summary>
/// Optional arguments (x-arguments) passed when declaring the queue.
/// </summary>
public IReadOnlyDictionary<string, string> QueueArguments { get; set; }
public IRabbitMQArguments QueueArguments { get; set; }
/// <summary>
/// Determines if the QueueInfo properties contain a valid combination.

1
Tapeti/Default/DependencyResolverBinding.cs
View File

@ -4,7 +4,6 @@ using Tapeti.Config;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Attempts to resolve any unhandled parameters to Controller methods using the IoC container.
/// This middleware is included by default in the standard TapetiConfig.

1
Tapeti/Default/DevNullLogger.cs
View File

@ -3,7 +3,6 @@ using Tapeti.Config;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Default ILogger implementation which does not log anything.
/// </summary>

1
Tapeti/Default/FallbackStringEnumConverter.cs
View File

@ -4,7 +4,6 @@ using Newtonsoft.Json;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Converts an <see cref="T:System.Enum" /> to and from its name string value. If an unknown string value is encountered
/// it will translate to 0xDEADBEEF (-559038737) so it can be gracefully handled.

3
Tapeti/Default/JsonMessageSerializer.cs
View File

@ -6,7 +6,6 @@ using Tapeti.Config;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// IMessageSerializer implementation for JSON encoding and decoding using Newtonsoft.Json.
/// </summary>
@ -49,7 +48,7 @@ namespace Tapeti.Default
/// <inheritdoc />
public object Deserialize(byte[] body, IMessageProperties properties)
{
if (!(properties.ContentType is ContentType))
if (properties.ContentType is not ContentType)
throw new ArgumentException($"content_type must be {ContentType}");
var typeName = properties.GetHeader(ClassTypeHeader);

1
Tapeti/Default/MessageBinding.cs
View File

@ -3,7 +3,6 @@ using Tapeti.Config;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Gets the message class from the first parameter of a controller method.
/// This middleware is included by default in the standard TapetiConfig.

1
Tapeti/Default/MessageProperties.cs
View File

@ -4,7 +4,6 @@ using Tapeti.Config;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// IMessagePropertiesReader implementation for providing properties manually
/// </summary>

1
Tapeti/Default/NackExceptionStrategy.cs
View File

@ -3,7 +3,6 @@ using Tapeti.Config;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Default implementation of an exception strategy which marks the messages as Error.
/// </summary>

1
Tapeti/Default/NamespaceMatchExchangeStrategy.cs
View File

@ -3,7 +3,6 @@ using System.Text.RegularExpressions;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// IExchangeStrategy implementation which uses the first identifier in the namespace in lower case,
/// skipping the first identifier if it is 'Messaging'.

1
Tapeti/Default/PublishResultBinding.cs
View File

@ -8,7 +8,6 @@ using Tapeti.Helpers;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Attempts to publish a return value for Controller methods as a response to the incoming message.
/// </summary>

1
Tapeti/Default/RequeueExceptionStrategy.cs
View File

@ -5,7 +5,6 @@ using Tapeti.Config;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// Example exception strategy which requeues all messages that result in an error.
/// </summary>

1
Tapeti/Default/TypeNameRoutingKeyStrategy.cs
View File

@ -7,7 +7,6 @@ using Tapeti.Helpers;
namespace Tapeti.Default
{
/// <inheritdoc />
/// <summary>
/// IRoutingKeyStrategy implementation which transforms the class name into a dot-separated routing key based
/// on the casing. Accounts for acronyms. If the class name ends with 'Message' it is not included in the routing key.

1
Tapeti/Exceptions/NackException.cs
View File

@ -2,7 +2,6 @@
namespace Tapeti.Exceptions
{
/// <inheritdoc />
/// <summary>
/// Raised when a message is nacked by the message bus.
/// </summary>

1
Tapeti/Exceptions/NoRouteException.cs
View File

@ -2,7 +2,6 @@
namespace Tapeti.Exceptions
{
/// <inheritdoc />
/// <summary>
/// Raised when a mandatory message has no route.
/// </summary>

2
Tapeti/Helpers/DictionaryHelper.cs
View File

@ -10,7 +10,7 @@ namespace Tapeti.Helpers
/// <summary>
/// Checks if two dictionaries are considered compatible. If either is null they are considered empty.
/// </summary>
public static bool NullSafeSameValues(this IReadOnlyDictionary<string, string> arguments1, IReadOnlyDictionary<string, string> arguments2)
public static bool NullSafeSameValues(this IReadOnlyDictionary<string, object> arguments1, IReadOnlyDictionary<string, object> arguments2)
{
if (arguments1 == null || arguments2 == null)
return (arguments1 == null || arguments1.Count == 0) && (arguments2 == null || arguments2.Count == 0);

1
Tapeti/IDependencyResolver.cs
View File

@ -23,7 +23,6 @@ namespace Tapeti
}
/// <inheritdoc />
/// <summary>
/// Allows registering controller classes into the IoC container. Also registers default implementations,
/// so that the calling application may override these.

6
Tapeti/ILogger.cs
View File

@ -1,6 +1,6 @@
using System;
using System.Collections.Generic;
using Tapeti.Config;
using Tapeti.Connection;
// ReSharper disable UnusedMember.Global
// ReSharper disable UnusedMemberInSuper.Global
@ -24,7 +24,6 @@ namespace Tapeti
}
/// <inheritdoc />
/// <summary>
/// Contains information about the failed connection.
/// </summary>
@ -37,7 +36,6 @@ namespace Tapeti
}
/// <inheritdoc />
/// <summary>
/// Contains information about the established connection.
/// </summary>
@ -140,7 +138,7 @@ namespace Tapeti
/// <param name="queueName">The name of the queue that is declared</param>
/// <param name="existingArguments">The x-arguments of the existing queue</param>
/// <param name="arguments">The x-arguments of the queue that would be declared</param>
void QueueExistsWarning(string queueName, IReadOnlyDictionary<string, string> existingArguments, IReadOnlyDictionary<string, string> arguments);
void QueueExistsWarning(string queueName, IRabbitMQArguments existingArguments, IRabbitMQArguments arguments);
/// <summary>
/// Called before a binding is added to a queue.

1
Tapeti/IPublisher.cs
View File

@ -54,7 +54,6 @@ namespace Tapeti
}
/// <inheritdoc />
/// <summary>
/// Low-level publisher for Tapeti internal use.
/// </summary>

4
Tapeti/Tapeti.csproj
View File

@ -23,8 +23,8 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
<PackageReference Include="RabbitMQ.Client" Version="6.4.0" />
<PackageReference Include="Newtonsoft.Json" Version="13.*" />
<PackageReference Include="RabbitMQ.Client" Version="6.*" />
<PackageReference Include="System.Configuration.ConfigurationManager" Version="7.0.0" />
</ItemGroup>

89
Tapeti/TapetiAppSettingsConnectionParams.cs
View File

@ -1,89 +0,0 @@
using System.Configuration;
using System.Linq;
// ReSharper disable UnusedMember.Global
namespace Tapeti
{
/// <inheritdoc />
/// <summary>
/// Implementation of TapetiConnectionParams which reads the values from the AppSettings.
/// </summary>
/// <list type="table">
/// <listheader>
/// <description>AppSettings keys</description>
/// </listheader>
/// <item><description>rabbitmq:hostname</description></item>
/// <item><description>rabbitmq:port</description></item>
/// <item><description>rabbitmq:virtualhost</description></item>
/// <item><description>rabbitmq:username</description></item>
/// <item><description>rabbitmq:password</description></item>
/// <item><description>rabbitmq:prefetchcount</description></item>
/// <item><description>rabbitmq:managementport</description></item>
/// <item><description>rabbitmq:clientproperty:*</description></item>
/// </list>
public class TapetiAppSettingsConnectionParams : TapetiConnectionParams
{
private const string DefaultPrefix = "rabbitmq:";
// ReSharper disable InconsistentNaming
private const string KeyHostname = "hostname";
private const string KeyPort = "port";
private const string KeyVirtualHost = "virtualhost";
private const string KeyUsername = "username";
private const string KeyPassword = "password";
private const string KeyPrefetchCount = "prefetchcount";
private const string KeyManagementPort = "managementport";
private const string KeyClientProperty = "clientproperty:";
// ReSharper restore InconsistentNaming
private readonly struct AppSettingsKey
{
public readonly string Entry;
public readonly string Parameter;
public AppSettingsKey(string entry, string parameter)
{
Entry = entry;
Parameter = parameter;
}
}
/// <inheritdoc />
/// <summary></summary>
/// <param name="prefix">The prefix to apply to the keys. Defaults to "rabbitmq:"</param>
public TapetiAppSettingsConnectionParams(string prefix = DefaultPrefix)
{
var keys = !string.IsNullOrEmpty(prefix)
? ConfigurationManager.AppSettings.AllKeys.Where(k => k.StartsWith(prefix)).Select(k => new AppSettingsKey(k, k.Substring(prefix.Length)))
: ConfigurationManager.AppSettings.AllKeys.Select(k => new AppSettingsKey(k, k));
foreach (var key in keys)
{
var value = ConfigurationManager.AppSettings[key.Entry];
if (key.Parameter.StartsWith(KeyClientProperty))
{
ClientProperties.Add(key.Parameter.Substring(KeyClientProperty.Length), value);
}
else
{
// ReSharper disable once SwitchStatementMissingSomeCases - don't fail if we encounter an unknown value
switch (key.Parameter)
{
case KeyHostname: HostName = value; break;
case KeyPort: Port = int.Parse(value); break;
case KeyVirtualHost: VirtualHost = value; break;
case KeyUsername: Username = value; break;
case KeyPassword: Password = value; break;
case KeyPrefetchCount: PrefetchCount = ushort.Parse(value); break;
case KeyManagementPort: ManagementPort = int.Parse(value); break;
}
}
}
}
}
}

2
Tapeti/TapetiConfig.cs
View File

@ -189,7 +189,7 @@ namespace Tapeti
/// </summary>
protected void RegisterDefaults()
{
if (!(DependencyResolver is IDependencyContainer container))
if (DependencyResolver is not IDependencyContainer container)
return;
if (ConsoleHelper.IsAvailable())

27
Tapeti/TapetiConfigControllers.cs
View File

@ -1,16 +1,15 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using Tapeti.Annotations;
using Tapeti.Config;
using Tapeti.Connection;
using Tapeti.Default;
// ReSharper disable UnusedMember.Global
namespace Tapeti
{
/// <inheritdoc />
/// <summary>
/// Thrown when an issue is detected in a controller configuration.
/// </summary>
@ -171,40 +170,34 @@ namespace Tapeti
}
private static IReadOnlyDictionary<string, string> GetQueueArguments(QueueArgumentsAttribute queueArgumentsAttribute)
private static IRabbitMQArguments GetQueueArguments(QueueArgumentsAttribute queueArgumentsAttribute)
{
if (queueArgumentsAttribute == null)
return null;
#if NETSTANDARD2_1_OR_GREATER
var arguments = new Dictionary<string, string>(queueArgumentsAttribute.CustomArguments);
#else
var arguments = new Dictionary<string, string>();
foreach (var pair in queueArgumentsAttribute.CustomArguments)
arguments.Add(pair.Key, pair.Value);
#endif
var arguments = new RabbitMQArguments(queueArgumentsAttribute.CustomArguments);
if (queueArgumentsAttribute.MaxLength > 0)
arguments.Add(@"x-max-length", queueArgumentsAttribute.MaxLength.ToString());
arguments.Add(@"x-max-length", queueArgumentsAttribute.MaxLength);
if (queueArgumentsAttribute.MaxLengthBytes > 0)
arguments.Add(@"x-max-length-bytes", queueArgumentsAttribute.MaxLengthBytes.ToString());
arguments.Add(@"x-max-length-bytes", queueArgumentsAttribute.MaxLengthBytes);
if (queueArgumentsAttribute.MessageTTL > 0)
arguments.Add(@"x-message-ttl", queueArgumentsAttribute.MessageTTL.ToString());
arguments.Add(@"x-message-ttl", queueArgumentsAttribute.MessageTTL);
switch (queueArgumentsAttribute.Overflow)
{
case RabbitMQOverflow.NotSpecified:
break;
case RabbitMQOverflow.DropHead:
arguments.Add(@"x-overflow", @"drop-head");
arguments.AddUTF8(@"x-overflow", @"drop-head");
break;
case RabbitMQOverflow.RejectPublish:
arguments.Add(@"x-overflow", @"reject-publish");
arguments.AddUTF8(@"x-overflow", @"reject-publish");
break;
case RabbitMQOverflow.RejectPublishDeadletter:
arguments.Add(@"x-overflow", @"reject-publish-dlx");
arguments.AddUTF8(@"x-overflow", @"reject-publish-dlx");
break;
default:
throw new ArgumentOutOfRangeException(nameof(queueArgumentsAttribute.Overflow), queueArgumentsAttribute.Overflow, "Unsupported Overflow value");

1
Tapeti/TapetiConnection.cs
View File

@ -9,7 +9,6 @@ using Tapeti.Connection;
namespace Tapeti
{
/// <inheritdoc />
/// <summary>
/// Creates a connection to RabbitMQ based on the provided Tapeti config.
/// </summary>

1
Tapeti/Tasks/SingleThreadTaskQueue.cs
View File

@ -6,7 +6,6 @@ using System.Threading.Tasks;
namespace Tapeti.Tasks
{
/// <inheritdoc />
/// <summary>
/// An implementation of a queue which runs tasks on a single thread.
/// </summary>