GitHub-Mirror/Tapeti
1
0
Fork 0
mirror of synced 2024-11-14 06:29:17 +00:00
Code Issues Projects Releases Wiki Activity

Merge branch 'release/3.1.3'

Browse Source
This commit is contained in:
Mark van Renswoude 2023-04-26 11:16:49 +02:00
commit c533f4866e
50 changed files with 588 additions and 272 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Examples/01-PublishSubscribe/ExampleMessageController.cs
View File

@ -1,7 +1,7 @@
using System;
using ExampleLib;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _01_PublishSubscribe
{

2
Examples/02-DeclareDurableQueues/ExampleMessageController.cs
View File

@ -1,7 +1,7 @@
using System;
using ExampleLib;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _02_DeclareDurableQueues
{

2
Examples/03-FlowRequestResponse/ParallelFlowController.cs
View File

@ -2,7 +2,7 @@
using System.Threading.Tasks;
using ExampleLib;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
using Tapeti.Flow;
using Tapeti.Flow.Annotations;

2
Examples/03-FlowRequestResponse/ReceivingMessageController.cs
View File

@ -1,6 +1,6 @@
using System.Threading.Tasks;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _03_FlowRequestResponse
{

2
Examples/03-FlowRequestResponse/SimpleFlowController.cs
View File

@ -1,7 +1,7 @@
using System;
using ExampleLib;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
using Tapeti.Flow;
using Tapeti.Flow.Annotations;

2
Examples/04-Transient/UsersMessageController.cs
View File

@ -1,7 +1,7 @@
using System;
using System.Threading.Tasks;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _04_Transient
{

2
Examples/05-SpeedTest/SpeedMessageController.cs
View File

@ -1,5 +1,5 @@
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _05_SpeedTest
{

2
Examples/06-StatelessRequestResponse/ExampleMessageController.cs
View File

@ -1,7 +1,7 @@
using System;
using ExampleLib;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _06_StatelessRequestResponse
{

2
Examples/06-StatelessRequestResponse/ReceivingMessageController.cs
View File

@ -1,5 +1,5 @@
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _06_StatelessRequestResponse
{

2
Examples/07-ParallelizationTest/ParallelizationMessageController.cs
View File

@ -1,6 +1,6 @@
using System.Threading.Tasks;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _07_ParallelizationTest
{

2
Examples/08-MessageHandlerLogging/SlowMessageController.cs
View File

@ -2,7 +2,7 @@
using System.Threading.Tasks;
using ExampleLib;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
namespace _08_MessageHandlerLogging
{

2
Examples/08-MessageHandlerLogging/SpeedyMessageController.cs
View File

@ -1,6 +1,6 @@
using System;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
using Tapeti.Serilog;
namespace _08_MessageHandlerLogging

1
Tapeti.Autofac/Tapeti.Autofac.csproj
View File

@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.SimpleInjector.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>

1
Tapeti.CastleWindsor/Tapeti.CastleWindsor.csproj
View File

@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.SimpleInjector.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>

2
Tapeti.CastleWindsor/WindsorDependencyResolver.cs
View File

@ -1,12 +1,14 @@
using System;
using Castle.MicroKernel.Registration;
using Castle.Windsor;
using JetBrains.Annotations;
namespace Tapeti.CastleWindsor
{
/// <summary>
/// Dependency resolver and container implementation for Castle Windsor.
/// </summary>
[PublicAPI]
public class WindsorDependencyResolver : IDependencyContainer
{
private readonly IWindsorContainer container;

1
Tapeti.DataAnnotations/Tapeti.DataAnnotations.csproj
View File

@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.DataAnnotations.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>

8
Tapeti.Flow.SQL/Tapeti.Flow.SQL.csproj
View File

@ -1,4 +1,4 @@
<Project Sdk="Microsoft.NET.Sdk">
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net6.0;net7.0</TargetFrameworks>
@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.Flow.SQL.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>
@ -19,11 +18,6 @@
<NoWarn>1701;1702</NoWarn>
</PropertyGroup>
<PropertyGroup Condition="'$(TargetFramework)'!='netstandard2.0'">
<!-- Suppress 'using statement can be simplified' which requires language version 8 not available in .NET Standard 2.0 -->
<NoWarn>IDE0063</NoWarn>
</PropertyGroup>
<ItemGroup>
<None Remove="scripts\Flow table.sql" />
</ItemGroup>

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

@ -154,7 +154,9 @@ namespace Tapeti.Flow.Default
var flowHandler = context.Config.DependencyResolver.Resolve<IFlowHandler>();
return flowHandler.Execute(new FlowHandlerContext(context), new DelegateYieldPoint(async flowContext =>
{
await flowContext.Store(context.Binding.QueueType == QueueType.Durable);
// IFlowParallelRequest.AddRequest will store the flow immediately
if (!flowPayload.FlowContext.IsStoredOrDeleted())
await flowContext.Store(context.Binding.QueueType == QueueType.Durable);
}));
}

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

@ -1,7 +1,7 @@
using System;
using System.Threading.Tasks;
using Tapeti.Config;
using Tapeti.Flow.FlowHelpers;
using Tapeti.Helpers;
namespace Tapeti.Flow.Default
{

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

@ -1,4 +1,4 @@
using System;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
@ -8,7 +8,7 @@ using Tapeti.Annotations;
using Tapeti.Config;
using Tapeti.Default;
using Tapeti.Flow.Annotations;
using Tapeti.Flow.FlowHelpers;
using Tapeti.Helpers;
namespace Tapeti.Flow.Default
{
@ -55,7 +55,7 @@ namespace Tapeti.Flow.Default
/// <inheritdoc />
public IFlowParallelRequestBuilder YieldWithParallelRequest()
{
return new ParallelRequestBuilder(config, this);
return new ParallelRequestBuilder(config, this, publisher);
}
/// <inheritdoc />
@ -71,8 +71,8 @@ namespace Tapeti.Flow.Default
}
internal async Task SendRequest(FlowContext context, object message, ResponseHandlerInfo responseHandlerInfo,
string? convergeMethodName = null, bool convergeMethodTaskSync = false, bool store = true)
internal async Task<MessageProperties> PrepareRequest(FlowContext context, ResponseHandlerInfo responseHandlerInfo,
string convergeMethodName = null, bool convergeMethodTaskSync = false)
{
if (!context.HasFlowStateAndLock)
{
@ -96,8 +96,15 @@ namespace Tapeti.Flow.Default
ReplyTo = responseHandlerInfo.ReplyToQueue
};
if (store)
await context.Store(responseHandlerInfo.IsDurableQueue);
return properties;
}
internal async Task SendRequest(FlowContext context, object message, ResponseHandlerInfo responseHandlerInfo,
string convergeMethodName = null, bool convergeMethodTaskSync = false)
{
var properties = await PrepareRequest(context, responseHandlerInfo, convergeMethodName, convergeMethodTaskSync);
await context.Store(responseHandlerInfo.IsDurableQueue);
await publisher.Publish(message, properties, true);
}
@ -134,7 +141,7 @@ namespace Tapeti.Flow.Default
{
await context.Delete();
if (context.HasFlowStateAndLock && context.FlowState.Metadata.Reply != null)
if (context is { HasFlowStateAndLock: true, FlowState.Metadata.Reply: { } })
throw new YieldPointException($"Flow must end with a response message of type {context.FlowState.Metadata.Reply.ResponseTypeName}");
}
@ -200,7 +207,7 @@ namespace Tapeti.Flow.Default
flowContext.SetFlowState(flowState, flowStateLock);
}
/// <inheritdoc />
public async ValueTask Execute(IFlowHandlerContext context, IYieldPoint yieldPoint)
{
@ -222,7 +229,7 @@ namespace Tapeti.Flow.Default
}
else
flowContext = flowPayload.FlowContext;
try
{
await executableYieldPoint.Execute(flowContext);
@ -327,13 +334,15 @@ namespace Tapeti.Flow.Default
private readonly ITapetiConfig config;
private readonly FlowProvider flowProvider;
private readonly IInternalPublisher publisher;
private readonly List<RequestInfo> requests = new();
public ParallelRequestBuilder(ITapetiConfig config, FlowProvider flowProvider)
public ParallelRequestBuilder(ITapetiConfig config, FlowProvider flowProvider, IInternalPublisher publisher)
{
this.config = config;
this.flowProvider = flowProvider;
this.publisher = publisher;
}
@ -407,18 +416,21 @@ namespace Tapeti.Flow.Default
if (convergeMethod.Method.DeclaringType != context.HandlerContext.Controller?.GetType())
throw new YieldPointException("Converge method must be in the same controller class");
var preparedRequests = new List<PreparedRequest>();
foreach (var requestInfo in requests)
{
await flowProvider.SendRequest(
var properties = await flowProvider.PrepareRequest(
context,
requestInfo.Message,
requestInfo.ResponseHandlerInfo,
convergeMethod.Method.Name,
convergeMethodSync,
false);
convergeMethodSync);
preparedRequests.Add(new PreparedRequest(requestInfo.Message, properties));
}
await context.Store(requests.Any(i => i.ResponseHandlerInfo.IsDurableQueue));
await Task.WhenAll(preparedRequests.Select(r => publisher.Publish(r.Message, r.Properties, true)));
});
}
}
@ -465,12 +477,11 @@ namespace Tapeti.Flow.Default
throw new InvalidOperationException("No ContinuationMetadata in FlowContext");
return flowProvider.SendRequest(
flowContext,
message,
responseHandlerInfo,
flowContext.ContinuationMetadata.ConvergeMethodName,
flowContext.ContinuationMetadata.ConvergeMethodSync,
false);
flowContext,
message,
responseHandlerInfo,
flowContext.ContinuationMetadata.ConvergeMethodName,
flowContext.ContinuationMetadata.ConvergeMethodSync);
}
}
@ -489,5 +500,19 @@ namespace Tapeti.Flow.Default
IsDurableQueue = isDurableQueue;
}
}
internal class PreparedRequest
{
public object Message { get; }
public MessageProperties Properties { get; }
public PreparedRequest(object message, MessageProperties properties)
{
Message = message;
Properties = properties;
}
}
}
}

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

@ -6,6 +6,7 @@ using System.Linq;
using System.Threading.Tasks;
using Tapeti.Config;
using Tapeti.Flow.FlowHelpers;
using Tapeti.Helpers;
namespace Tapeti.Flow.Default
{

180
Tapeti.Flow/ReSharper/JetBrains.Annotations.cs
View File

@ -1,180 +0,0 @@
/*
* Stripped version of the ReSharper Annotations source. Enables hinting without referencing the
* ReSharper.Annotations NuGet package.
*
* If you need more annotations, this code was generated using
* ReSharper - Options - Code Annotations - Copy C# implementation to clipboard
*/
/* MIT License
Copyright (c) 2016 JetBrains http://www.jetbrains.com
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE. */
using System;
#pragma warning disable 1591
// ReSharper disable UnusedMember.Global
// ReSharper disable MemberCanBePrivate.Global
// ReSharper disable UnusedAutoPropertyAccessor.Global
// ReSharper disable IntroduceOptionalParameters.Global
// ReSharper disable MemberCanBeProtected.Global
// ReSharper disable InconsistentNaming
// ReSharper disable InheritdocConsiderUsage
// ReSharper disable once CheckNamespace
namespace JetBrains.Annotations
{
/// <summary>
/// Indicates that the value of the marked element could be <c>null</c> sometimes,
/// so the check for <c>null</c> is necessary before its usage.
/// </summary>
/// <example><code>
/// [CanBeNull] object Test() => null;
///
/// void UseTest() {
/// var p = Test();
/// var s = p.ToString(); // Warning: Possible 'System.NullReferenceException'
/// }
/// </code></example>
[AttributeUsage(
AttributeTargets.Method | AttributeTargets.Parameter | AttributeTargets.Property |
AttributeTargets.Delegate | AttributeTargets.Field | AttributeTargets.Event |
AttributeTargets.Class | AttributeTargets.Interface | AttributeTargets.GenericParameter)]
internal sealed class CanBeNullAttribute : Attribute { }
/// <summary>
/// Indicates that the value of the marked element could never be <c>null</c>.
/// </summary>
/// <example><code>
/// [NotNull] object Foo() {
/// return null; // Warning: Possible 'null' assignment
/// }
/// </code></example>
[AttributeUsage(
AttributeTargets.Method | AttributeTargets.Parameter | AttributeTargets.Property |
AttributeTargets.Delegate | AttributeTargets.Field | AttributeTargets.Event |
AttributeTargets.Class | AttributeTargets.Interface | AttributeTargets.GenericParameter)]
internal sealed class NotNullAttribute : Attribute { }
/// <summary>
/// Indicates that the marked symbol is used implicitly (e.g. via reflection, in external library),
/// so this symbol will not be marked as unused (as well as by other usage inspections).
/// </summary>
[AttributeUsage(AttributeTargets.All)]
internal sealed class UsedImplicitlyAttribute : Attribute
{
public UsedImplicitlyAttribute()
: this(ImplicitUseKindFlags.Default, ImplicitUseTargetFlags.Default) { }
public UsedImplicitlyAttribute(ImplicitUseKindFlags useKindFlags)
: this(useKindFlags, ImplicitUseTargetFlags.Default) { }
public UsedImplicitlyAttribute(ImplicitUseTargetFlags targetFlags)
: this(ImplicitUseKindFlags.Default, targetFlags) { }
public UsedImplicitlyAttribute(ImplicitUseKindFlags useKindFlags, ImplicitUseTargetFlags targetFlags)
{
UseKindFlags = useKindFlags;
TargetFlags = targetFlags;
}
public ImplicitUseKindFlags UseKindFlags { get; }
public ImplicitUseTargetFlags TargetFlags { get; }
}
/// <summary>
/// Should be used on attributes and causes ReSharper to not mark symbols marked with such attributes
/// as unused (as well as by other usage inspections)
/// </summary>
[AttributeUsage(AttributeTargets.Class | AttributeTargets.GenericParameter)]
internal sealed class MeansImplicitUseAttribute : Attribute
{
public MeansImplicitUseAttribute()
: this(ImplicitUseKindFlags.Default, ImplicitUseTargetFlags.Default) { }
public MeansImplicitUseAttribute(ImplicitUseKindFlags useKindFlags)
: this(useKindFlags, ImplicitUseTargetFlags.Default) { }
public MeansImplicitUseAttribute(ImplicitUseTargetFlags targetFlags)
: this(ImplicitUseKindFlags.Default, targetFlags) { }
public MeansImplicitUseAttribute(ImplicitUseKindFlags useKindFlags, ImplicitUseTargetFlags targetFlags)
{
UseKindFlags = useKindFlags;
TargetFlags = targetFlags;
}
[UsedImplicitly] public ImplicitUseKindFlags UseKindFlags { get; private set; }
[UsedImplicitly] public ImplicitUseTargetFlags TargetFlags { get; private set; }
}
[Flags]
internal enum ImplicitUseKindFlags
{
Default = Access | Assign | InstantiatedWithFixedConstructorSignature,
/// <summary>Only entity marked with attribute considered used.</summary>
Access = 1,
/// <summary>Indicates implicit assignment to a member.</summary>
Assign = 2,
/// <summary>
/// Indicates implicit instantiation of a type with fixed constructor signature.
/// That means any unused constructor parameters won't be reported as such.
/// </summary>
InstantiatedWithFixedConstructorSignature = 4,
/// <summary>Indicates implicit instantiation of a type.</summary>
InstantiatedNoFixedConstructorSignature = 8
}
/// <summary>
/// Specify what is considered used implicitly when marked
/// with <see cref="MeansImplicitUseAttribute"/> or <see cref="UsedImplicitlyAttribute"/>.
/// </summary>
[Flags]
internal enum ImplicitUseTargetFlags
{
Default = Itself,
Itself = 1,
/// <summary>Members of entity marked with attribute are considered used.</summary>
Members = 2,
/// <summary>Entity marked with attribute and all its members considered used.</summary>
WithMembers = Itself | Members
}
/// <summary>
/// This attribute is intended to mark publicly available API
/// which should not be removed and so is treated as used.
/// </summary>
[MeansImplicitUse(ImplicitUseTargetFlags.WithMembers)]
internal sealed class PublicAPIAttribute : Attribute
{
public PublicAPIAttribute() { }
public PublicAPIAttribute(string comment)
{
Comment = comment;
}
public string? Comment { get; }
}
}

6
Tapeti.Flow/Tapeti.Flow.csproj
View File

@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.Flow.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>
@ -19,11 +18,6 @@
<NoWarn>1701;1702</NoWarn>
</PropertyGroup>
<PropertyGroup Condition="'$(TargetFramework)'!='netstandard2.0'">
<!-- Suppress 'Use switch expression' which requires language version 8 not available in .NET Standard 2.0 -->
<NoWarn>IDE0066</NoWarn>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\Tapeti\Tapeti.csproj" />
</ItemGroup>

1
Tapeti.Ninject/Tapeti.Ninject.csproj
View File

@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.SimpleInjector.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>

1
Tapeti.Serilog/Tapeti.Serilog.csproj
View File

@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.Serilog.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>

1
Tapeti.SimpleInjector/Tapeti.SimpleInjector.csproj
View File

@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.SimpleInjector.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>

65
Tapeti.Tests/Client/Controller/RequestResponseFilterController.cs Normal file
View File

@ -0,0 +1,65 @@
using System.Threading.Tasks;
using Tapeti.Config.Annotations;
namespace Tapeti.Tests.Client.Controller
{
[Annotations.Request(Response = typeof(FilteredResponseMessage))]
public class FilteredRequestMessage
{
public int ExpectedHandler { get; set; }
}
public class FilteredResponseMessage
{
public int ExpectedHandler { get; set; }
}
#pragma warning disable CA1822 // Mark members as static
[MessageController]
[DurableQueue("request.response.filter")]
public class RequestResponseFilterController
{
public static TaskCompletionSource<int> ValidResponse { get; private set; } = new();
public static TaskCompletionSource<int> InvalidResponse { get; private set; } = new();
public FilteredResponseMessage EchoRequest(FilteredRequestMessage message)
{
return new FilteredResponseMessage
{
ExpectedHandler = message.ExpectedHandler
};
}
[NoBinding]
public static void ResetCompletionSource()
{
ValidResponse = new TaskCompletionSource<int>();
InvalidResponse = new TaskCompletionSource<int>();
}
[ResponseHandler]
public void Handler1(FilteredResponseMessage message)
{
if (message.ExpectedHandler != 1)
InvalidResponse.TrySetResult(1);
else
ValidResponse.SetResult(1);
}
[ResponseHandler]
public void Handler2(FilteredResponseMessage message)
{
if (message.ExpectedHandler != 2)
InvalidResponse.TrySetResult(2);
else
ValidResponse.SetResult(2);
}
}
#pragma warning restore CA1822
}

87
Tapeti.Tests/Client/ControllerTests.cs Normal file
View File

@ -0,0 +1,87 @@
using System.Threading.Tasks;
using FluentAssertions;
using SimpleInjector;
using Tapeti.Config;
using Tapeti.SimpleInjector;
using Tapeti.Tests.Client.Controller;
using Tapeti.Tests.Mock;
using Xunit;
using Xunit.Abstractions;
namespace Tapeti.Tests.Client
{
[Collection(RabbitMQCollection.Name)]
[Trait("Category", "Requires Docker")]
public class ControllerTests : IAsyncLifetime
{
private readonly RabbitMQFixture fixture;
private readonly Container container = new();
private TapetiConnection? connection;
public ControllerTests(RabbitMQFixture fixture, ITestOutputHelper testOutputHelper)
{
this.fixture = fixture;
container.RegisterInstance<ILogger>(new MockLogger(testOutputHelper));
}
public Task InitializeAsync()
{
return Task.CompletedTask;
}
public async Task DisposeAsync()
{
if (connection != null)
await connection.DisposeAsync();
}
[Fact]
public async Task RequestResponseFilter()
{
var config = new TapetiConfig(new SimpleInjectorDependencyResolver(container))
.EnableDeclareDurableQueues()
.RegisterController<RequestResponseFilterController>()
.Build();
connection = CreateConnection(config);
await connection!.Subscribe();
await connection.GetPublisher().PublishRequest<RequestResponseFilterController, FilteredRequestMessage, FilteredResponseMessage>(new FilteredRequestMessage
{
ExpectedHandler = 2
}, c => c.Handler2);
var handler = await RequestResponseFilterController.ValidResponse.Task;
handler.Should().Be(2);
var invalidHandler = await Task.WhenAny(RequestResponseFilterController.InvalidResponse.Task, Task.Delay(1000));
invalidHandler.Should().NotBe(RequestResponseFilterController.InvalidResponse.Task);
}
private TapetiConnection CreateConnection(ITapetiConfig config)
{
return new TapetiConnection(config)
{
Params = new TapetiConnectionParams
{
HostName = "127.0.0.1",
Port = fixture.RabbitMQPort,
ManagementPort = fixture.RabbitMQManagementPort,
Username = RabbitMQFixture.RabbitMQUsername,
Password = RabbitMQFixture.RabbitMQPassword,
PrefetchCount = 1
}
};
}
}
}

2
Tapeti.Tests/Client/RabbitMQFixture.cs
View File

@ -63,7 +63,7 @@ namespace Tapeti.Tests.Client
testcontainers = testcontainersBuilder.Build();
await testcontainers.StartAsync();
await testcontainers!.StartAsync();
RabbitMQPort = testcontainers.GetMappedPublicPort(DefaultRabbitMQPort);
RabbitMQManagementPort = testcontainers.GetMappedPublicPort(DefaultRabbitMQManagementPort);

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

@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

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

@ -7,7 +7,7 @@ using System.Threading.Tasks;
using FluentAssertions;
using FluentAssertions.Execution;
using Moq;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
using Tapeti.Config;
using Tapeti.Connection;
using Xunit;
@ -84,10 +84,10 @@ namespace Tapeti.Tests.Config
{
var config = GetControllerConfig<TestController>();
var binding1 = config.Bindings.Single(b => b is IControllerMethodBinding cmb && cmb.Method.Name == "HandleMessage1");
var binding1 = config.Bindings.Single(b => b is IControllerMethodBinding { Method.Name: "HandleMessage1" });
binding1.Should().NotBeNull();
var binding2 = config.Bindings.Single(b => b is IControllerMethodBinding cmb && cmb.Method.Name == "HandleMessage2");
var binding2 = config.Bindings.Single(b => b is IControllerMethodBinding { Method.Name: "HandleMessage2" });
binding2.Should().NotBeNull();

2
Tapeti.Tests/Config/SimpleControllerTest.cs
View File

@ -1,6 +1,6 @@
using System.Linq;
using FluentAssertions;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
using Tapeti.Config;
using Xunit;

4
Tapeti.Tests/Tapeti.Tests.csproj
View File

@ -11,9 +11,10 @@
<ItemGroup>
<PackageReference Include="FluentAssertions" Version="6.8.0" />
<PackageReference Include="JetBrains.Annotations" Version="2022.3.1" />
<PackageReference Include="JetBrains.Annotations" Version="2022.*" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.4.0" />
<PackageReference Include="Moq" Version="4.18.2" />
<PackageReference Include="SimpleInjector" Version="5.4.1" />
<PackageReference Include="Testcontainers" Version="2.2.0" />
<PackageReference Include="xunit" Version="2.4.2" />
<PackageReference Include="xunit.runner.visualstudio" Version="2.4.5">
@ -23,6 +24,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Tapeti.SimpleInjector\Tapeti.SimpleInjector.csproj" />
<ProjectReference Include="..\Tapeti\Tapeti.csproj" />
</ItemGroup>

1
Tapeti.Transient/Tapeti.Transient.csproj
View File

@ -11,7 +11,6 @@
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.Flow.png</PackageIcon>
<Version>2.0.0</Version>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>

77
Tapeti/Config/Annotations/BackwardsCompatibilityHelpers.cs Normal file
View File

@ -0,0 +1,77 @@
using System;
using System.Reflection;
#pragma warning disable CS0618 // Obsolete
#pragma warning disable CS1591 // Missing documentation
namespace Tapeti.Config.Annotations
{
/// <summary>
/// Provides extensions methods to support moved (marked obsolete) attributes from Tapeti.Annotations.
/// </summary>
public static class BackwardsCompatibilityHelpers
{
public static DurableQueueAttribute? GetDurableQueueAttribute(this MemberInfo member)
{
return member.GetCustomAttribute<DurableQueueAttribute>() ?? Upgrade(member.GetCustomAttribute<Tapeti.Annotations.DurableQueueAttribute>());
}
public static DynamicQueueAttribute? GetDynamicQueueAttribute(this MemberInfo member)
{
return member.GetCustomAttribute<DynamicQueueAttribute>() ?? Upgrade(member.GetCustomAttribute<Tapeti.Annotations.DynamicQueueAttribute>());
}
public static QueueArgumentsAttribute? GetQueueArgumentsAttribute(this MemberInfo member)
{
return member.GetCustomAttribute<QueueArgumentsAttribute>() ?? Upgrade(member.GetCustomAttribute<Tapeti.Annotations.QueueArgumentsAttribute>());
}
public static ResponseHandlerAttribute? GetResponseHandlerAttribute(this MemberInfo member)
{
return member.GetCustomAttribute<ResponseHandlerAttribute>() ?? Upgrade(member.GetCustomAttribute<Tapeti.Annotations.ResponseHandlerAttribute>());
}
public static bool HasMessageControllerAttribute(this MemberInfo member)
{
return member.IsDefined(typeof(MessageControllerAttribute)) || member.IsDefined(typeof(Tapeti.Annotations.MessageControllerAttribute));
}
private static DurableQueueAttribute? Upgrade(Tapeti.Annotations.DurableQueueAttribute? attribute)
{
return attribute == null ? null : new DurableQueueAttribute(attribute.Name);
}
private static DynamicQueueAttribute? Upgrade(Tapeti.Annotations.DynamicQueueAttribute? attribute)
{
return attribute == null ? null : new DynamicQueueAttribute(attribute.Prefix);
}
private static QueueArgumentsAttribute? Upgrade(Tapeti.Annotations.QueueArgumentsAttribute? attribute)
{
return attribute == null
? null
: new QueueArgumentsAttribute(attribute.CustomArguments)
{
MaxLength = attribute.MaxLength,
MaxLengthBytes = attribute.MaxLengthBytes,
Overflow = attribute.Overflow switch
{
Tapeti.Annotations.RabbitMQOverflow.NotSpecified => RabbitMQOverflow.NotSpecified,
Tapeti.Annotations.RabbitMQOverflow.DropHead => RabbitMQOverflow.DropHead,
Tapeti.Annotations.RabbitMQOverflow.RejectPublish => RabbitMQOverflow.RejectPublish,
Tapeti.Annotations.RabbitMQOverflow.RejectPublishDeadletter => RabbitMQOverflow.RejectPublishDeadletter,
_ => throw new ArgumentOutOfRangeException(nameof(attribute.Overflow))
},
MessageTTL = attribute.MessageTTL
};
}
private static ResponseHandlerAttribute? Upgrade(Tapeti.Annotations.ResponseHandlerAttribute? attribute)
{
return attribute == null ? null : new ResponseHandlerAttribute();
}
}
}

28
Tapeti/Config/Annotations/DurableQueueAttribute.cs Normal file
View File

@ -0,0 +1,28 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Config.Annotations
{
/// <summary>
/// Binds to an existing durable queue to receive messages. Can be used
/// on an entire MessageController class or on individual methods.
/// </summary>
[AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)]
[MeansImplicitUse(ImplicitUseTargetFlags.WithMembers)]
[PublicAPI]
public class DurableQueueAttribute : Attribute
{
/// <summary>
/// Specifies the name of the durable queue (must already be declared).
/// </summary>
public string Name { get; set; }
/// <inheritdoc />
/// <param name="name">The name of the durable queue</param>
public DurableQueueAttribute(string name)
{
Name = name;
}
}
}

32
Tapeti/Config/Annotations/DynamicQueueAttribute.cs Normal file
View File

@ -0,0 +1,32 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Config.Annotations
{
/// <summary>
/// Creates a non-durable auto-delete queue to receive messages. Can be used
/// on an entire MessageController class or on individual methods.
/// </summary>
[AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)]
[MeansImplicitUse]
[PublicAPI]
public class DynamicQueueAttribute : Attribute
{
/// <summary>
/// An optional prefix. If specified, Tapeti will compose the queue name using the
/// prefix and a unique ID. If not specified, an empty queue name will be passed
/// to RabbitMQ thus letting it create a unique queue name.
/// </summary>
public string? Prefix { get; set; }
/// <inheritdoc />
/// <param name="prefix">An optional prefix. If specified, Tapeti will compose the queue name using the
/// prefix and a unique ID. If not specified, an empty queue name will be passed
/// to RabbitMQ thus letting it create a unique queue name.</param>
public DynamicQueueAttribute(string? prefix = null)
{
Prefix = prefix;
}
}
}

16
Tapeti/Config/Annotations/MessageControllerAttribute.cs Normal file
View File

@ -0,0 +1,16 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Config.Annotations
{
/// <summary>
/// Attaching this attribute to a class includes it in the auto-discovery of message controllers
/// when using the RegisterAllControllers method. It is not required when manually registering a controller.
/// </summary>
[AttributeUsage(AttributeTargets.Class)]
[MeansImplicitUse(ImplicitUseTargetFlags.WithMembers)]
[PublicAPI]
public class MessageControllerAttribute : Attribute
{
}
}

14
Tapeti/Config/Annotations/NoBindingAttribute.cs Normal file
View File

@ -0,0 +1,14 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Config.Annotations
{
/// <summary>
/// Indicates that the method is not a message handler and should not be bound by Tapeti.
/// </summary>
[AttributeUsage(AttributeTargets.Method)]
[PublicAPI]
public class NoBindingAttribute : Attribute
{
}
}

106
Tapeti/Config/Annotations/QueueArgumentsAttribute.cs Normal file
View File

@ -0,0 +1,106 @@
using System;
using System.Collections.Generic;
using JetBrains.Annotations;
namespace Tapeti.Config.Annotations
{
/// <summary>
/// Determines the overflow behaviour of a queue that has reached it's maximum as set by <see cref="QueueArgumentsAttribute.MaxLength"/> or <see cref="QueueArgumentsAttribute.MaxLengthBytes"/>.
/// </summary>
[PublicAPI]
public enum RabbitMQOverflow
{
/// <summary>
/// The argument will not be explicitly specified and use the RabbitMQ default, which is equivalent to <see cref="DropHead"/>.
/// </summary>
NotSpecified,
/// <summary>
/// Discards or dead-letters the oldest published message. This is the default value.
/// </summary>
DropHead,
/// <summary>
/// Discards the most recently published messages and nacks the message.
/// </summary>
RejectPublish,
/// <summary>
/// Dead-letters the most recently published messages and nacks the message.
/// </summary>
RejectPublishDeadletter
}
/// <summary>
/// Specifies the optional queue arguments (also known as 'x-arguments') used when declaring
/// the queue.
/// </summary>
/// <remarks>
/// The QueueArguments attribute can be applied to any controller or method and will affect the queue
/// that is used in that context. For durable queues, at most one QueueArguments attribute can be specified
/// per unique queue name.
/// <br/><br/>
/// Also note that queue arguments can not be changed after a queue is declared. You should declare a new queue
/// and make the old one Obsolete to have Tapeti automatically removed it once it is empty. Tapeti will use the
/// existing queue, but log a warning at startup time.
/// </remarks>
[AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)]
[PublicAPI]
public class QueueArgumentsAttribute : Attribute
{
/// <summary>
/// The maximum number of messages in the queue. Set <see cref="Overflow"/> to determine the overflow behaviour.
/// </summary>
/// <remarks>
/// Corresponds to 'max-length'. See <see href="https://www.rabbitmq.com/maxlength.html"/>
/// </remarks>
public int MaxLength { get; set; }
/// <summary>
/// The maximum number of bytes in the queue (counting only the message bodies). Set <see cref="Overflow"/> to determine the overflow behaviour.
/// </summary>
/// <remarks>
/// Corresponds to 'x-max-length-bytes'. See <see href="https://www.rabbitmq.com/maxlength.html"/>
/// </remarks>
public int MaxLengthBytes { get; set; }
/// <inheritdoc cref="RabbitMQOverflow"/>
/// <remarks>
/// Corresponds to 'x-overflow'. Default is to drop or deadletter the oldest messages in the queue. See <see href="https://www.rabbitmq.com/maxlength.html"/>
/// </remarks>
public RabbitMQOverflow Overflow { get; set; } = RabbitMQOverflow.NotSpecified;
/// <summary>
/// Specifies the maximum Time-to-Live for messages in the queue, in milliseconds.
/// </summary>
/// <remarks>
/// Corresponds to 'x-message-ttl'. See <see href="https://www.rabbitmq.com/ttl.html" />
/// </remarks>
public int MessageTTL { get; set; }
/// <summary>
/// Any arguments to add which are not supported by properties of QueueArguments.
/// </summary>
public IReadOnlyDictionary<string, object> CustomArguments { get; private set; }
/// <inheritdoc cref="QueueArgumentsAttribute"/>
/// <param name="customArguments">Any arguments to add which are not supported by properties of QueueArguments. Must be a multiple of 2, specify each key followed by the value.</param>
public QueueArgumentsAttribute(params object[] customArguments)
{
if (customArguments.Length % 2 != 0)
throw new ArgumentException("customArguments must be a multiple of 2 to specify each key-value combination", nameof(customArguments));
var customArgumentsPairs = new Dictionary<string, object>();
for (var i = 0; i < customArguments.Length; i += 2)
customArgumentsPairs[(string)customArguments[i]] = customArguments[i + 1];
CustomArguments = customArgumentsPairs;
}
}
}

15
Tapeti/Config/Annotations/ResponseHandlerAttribute.cs Normal file
View File

@ -0,0 +1,15 @@
using JetBrains.Annotations;
using System;
namespace Tapeti.Config.Annotations
{
/// <summary>
/// Indicates that the method only handles response messages which are sent directly
/// to the queue. No binding will be created.
/// </summary>
[AttributeUsage(AttributeTargets.Method)]
[PublicAPI]
public class ResponseHandlerAttribute : Attribute
{
}
}

2
Tapeti/Config/ITapetiConfigBuilder.cs
View File

@ -1,4 +1,5 @@
using System;
using JetBrains.Annotations;
// ReSharper disable UnusedMember.Global
@ -8,6 +9,7 @@ namespace Tapeti.Config
/// Configures Tapeti. Every method other than Build returns the builder instance
/// for method chaining.
/// </summary>
[PublicAPI]
public interface ITapetiConfigBuilder
{
/// <summary>

4
Tapeti/Connection/TapetiPublisher.cs
View File

@ -4,6 +4,7 @@ using System.Reflection;
using System.Threading.Tasks;
using Tapeti.Annotations;
using Tapeti.Config;
using Tapeti.Config.Annotations;
using Tapeti.Default;
using Tapeti.Helpers;
@ -72,7 +73,7 @@ namespace Tapeti.Connection
if (!binding.Accept(requestAttribute.Response))
throw new ArgumentException($"responseHandler must accept message of type {requestAttribute.Response}", nameof(responseHandler));
var responseHandleAttribute = binding.Method.GetCustomAttribute<ResponseHandlerAttribute>();
var responseHandleAttribute = binding.Method.GetResponseHandlerAttribute();
if (responseHandleAttribute == null)
throw new ArgumentException("responseHandler must be marked with the ResponseHandler attribute", nameof(responseHandler));
@ -82,6 +83,7 @@ namespace Tapeti.Connection
var properties = new MessageProperties
{
CorrelationId = ResponseFilterMiddleware.CorrelationIdRequestPrefix + MethodSerializer.Serialize(responseHandler),
ReplyTo = binding.QueueName
};

10
Tapeti/Default/ControllerMethodBinding.cs
View File

@ -120,7 +120,7 @@ namespace Tapeti.Default
QueueName = await target.BindDynamic(bindingInfo.MessageClass, bindingInfo.QueueInfo.Name, bindingInfo.QueueInfo.QueueArguments);
else
{
await target.BindDurable(bindingInfo.MessageClass, bindingInfo.QueueInfo.Name, bindingInfo.QueueInfo.QueueArguments);
await target.BindDurable(bindingInfo.MessageClass, bindingInfo.QueueInfo.Name!, bindingInfo.QueueInfo.QueueArguments);
QueueName = bindingInfo.QueueInfo.Name;
}
@ -131,7 +131,7 @@ namespace Tapeti.Default
QueueName = await target.BindDynamicDirect(bindingInfo.MessageClass, bindingInfo.QueueInfo.Name, bindingInfo.QueueInfo.QueueArguments);
else
{
await target.BindDurableDirect(bindingInfo.QueueInfo.Name, bindingInfo.QueueInfo.QueueArguments);
await target.BindDurableDirect(bindingInfo.QueueInfo.Name!, bindingInfo.QueueInfo.QueueArguments);
QueueName = bindingInfo.QueueInfo.Name;
}
@ -143,7 +143,7 @@ namespace Tapeti.Default
}
else if (bindingInfo.QueueInfo.QueueType == Config.QueueType.Durable)
{
await target.BindDurableObsolete(bindingInfo.QueueInfo.Name);
await target.BindDurableObsolete(bindingInfo.QueueInfo.Name!);
QueueName = bindingInfo.QueueInfo.Name;
}
}
@ -317,7 +317,7 @@ namespace Tapeti.Default
/// <summary>
/// The name of the durable queue, or optional prefix of the dynamic queue.
/// </summary>
public string Name { get; set; }
public string? Name { get; set; }
/// <summary>
/// Optional arguments (x-arguments) passed when declaring the queue.
@ -330,7 +330,7 @@ namespace Tapeti.Default
public bool IsValid => QueueType == QueueType.Dynamic || !string.IsNullOrEmpty(Name);
public QueueInfo(QueueType queueType, string name)
public QueueInfo(QueueType queueType, string? name)
{
QueueType = queueType;
Name = name;

2
Tapeti/Default/DependencyResolverBinding.cs
View File

@ -15,7 +15,7 @@ namespace Tapeti.Default
{
next();
foreach (var parameter in context.Parameters.Where(p => !p.HasBinding && p.Info.ParameterType.IsClass))
foreach (var parameter in context.Parameters.Where(p => p is { HasBinding: false, Info.ParameterType.IsClass: true }))
parameter.SetBinding(messageContext => messageContext.Config.DependencyResolver.Resolve(parameter.Info.ParameterType));
}
}

37
Tapeti/Default/ResponseFilterMiddleware.cs Normal file
View File

@ -0,0 +1,37 @@
using System;
using System.Threading.Tasks;
using Tapeti.Config;
using Tapeti.Helpers;
namespace Tapeti.Default
{
/// <inheritdoc cref="IControllerMessageMiddleware"/> />
/// <summary>
/// Handles methods marked with the ResponseHandler attribute.
/// </summary>
internal class ResponseFilterMiddleware : IControllerFilterMiddleware//, IControllerMessageMiddleware
{
internal const string CorrelationIdRequestPrefix = "request|";
public async ValueTask Filter(IMessageContext context, Func<ValueTask> next)
{
if (!context.TryGet<ControllerMessageContextPayload>(out var controllerPayload))
return;
// If no CorrelationId is present, this could be a request-response in flight from a previous version of
// Tapeti so we should not filter the response handler.
if (!string.IsNullOrEmpty(context.Properties.CorrelationId))
{
if (!context.Properties.CorrelationId.StartsWith(CorrelationIdRequestPrefix))
return;
var methodName = context.Properties.CorrelationId[CorrelationIdRequestPrefix.Length..];
if (methodName != MethodSerializer.Serialize(controllerPayload.Binding.Method))
return;
}
await next();
}
}
}

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

@ -1,7 +1,7 @@
using System.Reflection;
using System.Text.RegularExpressions;
namespace Tapeti.Flow.FlowHelpers
namespace Tapeti.Helpers
{
/// <summary>
/// Converts a method into a unique string representation.

2
Tapeti/MessageController.cs
View File

@ -1,4 +1,4 @@
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
// ReSharper disable UnusedMember.Global

11
Tapeti/Tapeti.csproj
View File

@ -11,7 +11,6 @@
<PackageLicenseExpression>Unlicense</PackageLicenseExpression>
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.png</PackageIcon>
<LangVersion>9</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>
@ -19,20 +18,12 @@
<NoWarn>1701;1702</NoWarn>
</PropertyGroup>
<PropertyGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
<DefineConstants></DefineConstants>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="JetBrains.Annotations" Version="2022.*" />
<PackageReference Include="Newtonsoft.Json" Version="13.*" />
<PackageReference Include="RabbitMQ.Client" Version="[6.5]" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
<PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="5.0.0" />
<PackageReference Include="System.Threading.Tasks.Extensions" Version="4.5.4" />
</ItemGroup>
<ItemGroup>
<None Include="..\resources\icons\Tapeti.png">
<Pack>True</Pack>

32
Tapeti/TapetiConfigControllers.cs
View File

@ -2,7 +2,7 @@ using System;
using System.Linq;
using System.Reflection;
using System.Text;
using Tapeti.Annotations;
using Tapeti.Config.Annotations;
using Tapeti.Config;
using Tapeti.Connection;
using Tapeti.Default;
@ -48,12 +48,18 @@ namespace Tapeti
.Where(m => m.MemberType == MemberTypes.Method && m.DeclaringType != typeof(object) && (m as MethodInfo)?.IsSpecialName == false)
.Select(m => (MethodInfo)m))
{
if (method.GetCustomAttributes<NoBindingAttribute>().Any())
continue;
var methodIsObsolete = controllerIsObsolete || method.GetCustomAttribute<ObsoleteAttribute>() != null;
var context = new ControllerBindingContext(controller, method, method.GetParameters(), method.ReturnParameter);
if (method.GetCustomAttribute<ResponseHandlerAttribute>() != null)
if (method.GetResponseHandlerAttribute() != null)
{
context.SetBindingTargetMode(BindingTargetMode.Direct);
context.Use(new ResponseFilterMiddleware());
}
var allowBinding = false;
@ -100,6 +106,14 @@ namespace Tapeti
}
/// <inheritdoc cref="RegisterController"/>
public static ITapetiConfigBuilder RegisterController<TController>(this ITapetiConfigBuilder builder) where TController : class
{
return RegisterController(builder, typeof(TController));
}
/// <summary>
/// Registers all controllers in the specified assembly which are marked with the MessageController attribute.
/// </summary>
@ -107,7 +121,7 @@ namespace Tapeti
/// <param name="assembly">The assembly to scan for controllers.</param>
public static ITapetiConfigBuilder RegisterAllControllers(this ITapetiConfigBuilder builder, Assembly assembly)
{
foreach (var type in assembly.GetTypes().Where(t => t.IsDefined(typeof(MessageControllerAttribute))))
foreach (var type in assembly.GetTypes().Where(t => t.HasMessageControllerAttribute()))
RegisterController(builder, type);
return builder;
@ -130,9 +144,9 @@ namespace Tapeti
private static ControllerMethodBinding.QueueInfo? GetQueueInfo(MemberInfo member, ControllerMethodBinding.QueueInfo? fallbackQueueInfo)
{
var dynamicQueueAttribute = member.GetCustomAttribute<DynamicQueueAttribute>();
var durableQueueAttribute = member.GetCustomAttribute<DurableQueueAttribute>();
var queueArgumentsAttribute = member.GetCustomAttribute<QueueArgumentsAttribute>();
var dynamicQueueAttribute = member.GetDynamicQueueAttribute();
var durableQueueAttribute = member.GetDurableQueueAttribute();
var queueArgumentsAttribute = member.GetQueueArgumentsAttribute();
if (dynamicQueueAttribute != null && durableQueueAttribute != null)
throw new TopologyConfigurationException($"Cannot combine static and dynamic queue attributes on controller {member.DeclaringType?.Name} method {member.Name}");
@ -142,7 +156,7 @@ namespace Tapeti
QueueType queueType;
string name;
string? name;
if (dynamicQueueAttribute != null)
@ -180,10 +194,8 @@ namespace Tapeti
string stringValue => Encoding.UTF8.GetBytes(stringValue),
_ => p.Value
}
))
{
));
};
if (queueArgumentsAttribute.MaxLength > 0)
arguments.Add(@"x-max-length", queueArgumentsAttribute.MaxLength);