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

Merge branch 'hotfix/1.2.1'

Browse Source
This commit is contained in:
Mark van Renswoude 2019-05-20 15:24:56 +02:00
commit 978f2b73f4
8 changed files with 444 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Tapeti.Annotations/DurableQueueAttribute.cs
View File

@ -1,4 +1,5 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Annotations
{
@ -13,6 +14,7 @@ namespace Tapeti.Annotations
/// for deploy-time management of durable queues (shameless plug intended).
/// </remarks>
[AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)]
[MeansImplicitUse]
public class DurableQueueAttribute : Attribute
{
public string Name { get; set; }

2
Tapeti.Annotations/DynamicQueueAttribute.cs
View File

@ -1,4 +1,5 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Annotations
{
@ -8,6 +9,7 @@ namespace Tapeti.Annotations
/// on an entire MessageController class or on individual methods.
/// </summary>
[AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)]
[MeansImplicitUse]
public class DynamicQueueAttribute : Attribute
{
public string Prefix { get; set; }

2
Tapeti.Annotations/MessageControllerAttribute.cs
View File

@ -1,4 +1,5 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Annotations
{
@ -8,6 +9,7 @@ namespace Tapeti.Annotations
/// when using the RegisterAllControllers method. It is not required when manually registering a controller.
/// </summary>
[AttributeUsage(AttributeTargets.Class)]
[MeansImplicitUse]
public class MessageControllerAttribute : Attribute
{
}

17
Tapeti.Annotations/MessageHandlerAttribute.cs Normal file
View File

@ -0,0 +1,17 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Annotations
{
/// <inheritdoc />
/// <summary>
/// This attribute does nothing in runtime and is not required. It is only used as
/// a hint to ReSharper, and maybe developers as well, to indicate the method is
/// indeed used.
/// </summary>
[AttributeUsage(AttributeTargets.Method)]
[MeansImplicitUse]
public class MessageHandlerAttribute : Attribute
{
}
}

179
Tapeti.Annotations/ReSharper/JetBrains.Annotations.cs Normal file
View File

@ -0,0 +1,179 @@
/*
* 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 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; private set; }
public ImplicitUseTargetFlags TargetFlags { get; private set; }
}
/// <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([NotNull] string comment)
{
Comment = comment;
}
[CanBeNull] public string Comment { get; private set; }
}
}

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

@ -1,8 +1,10 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Flow.Annotations
{
[AttributeUsage(AttributeTargets.Method)]
[MeansImplicitUse]
public class StartAttribute : Attribute
{
}

179
Tapeti.Flow/ReSharper/JetBrains.Annotations.cs Normal file
View File

@ -0,0 +1,179 @@
/*
* 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 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; private set; }
public ImplicitUseTargetFlags TargetFlags { get; private set; }
}
/// <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([NotNull] string comment)
{
Comment = comment;
}
[CanBeNull] public string Comment { get; private set; }
}
}

80
Tapeti/Connection/TapetiWorker.cs
View File

@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
@ -37,6 +38,9 @@ namespace Tapeti.Connection
private IModel channelInstance;
private ulong lastDeliveryTag;
private DateTime connectedDateTime;
// These fields must be locked, since the callbacks for BasicAck/BasicReturn can run in a different thread
private readonly object confirmLock = new Object();
private readonly Dictionary<ulong, ConfirmMessageInfo> confirmMessages = new Dictionary<ulong, ConfirmMessageInfo>();
private readonly Dictionary<string, ReturnInfo> returnRoutingKeys = new Dictionary<string, ReturnInfo>();
@ -208,6 +212,9 @@ namespace Tapeti.Connection
async (handler, next) => await handler.Handle(context, next),
() => taskQueue.Value.Add(async () =>
{
if (Thread.CurrentThread.ManagedThreadId != 3)
Debug.WriteLine(Thread.CurrentThread.ManagedThreadId);
var body = messageSerializer.Serialize(context.Message, context.Properties);
Task<int> publishResultTask = null;
@ -225,7 +232,16 @@ namespace Tapeti.Connection
{
lastDeliveryTag++;
confirmMessages.Add(lastDeliveryTag, messageInfo);
Monitor.Enter(confirmLock);
try
{
confirmMessages.Add(lastDeliveryTag, messageInfo);
}
finally
{
Monitor.Exit(confirmLock);
}
publishResultTask = messageInfo.CompletionSource.Task;
}
else
@ -326,7 +342,17 @@ namespace Tapeti.Connection
if (config.UsePublisherConfirms)
{
lastDeliveryTag = 0;
confirmMessages.Clear();
Monitor.Enter(confirmLock);
try
{
confirmMessages.Clear();
}
finally
{
Monitor.Exit(confirmLock);
}
channelInstance.ConfirmSelect();
}
@ -403,35 +429,51 @@ namespace Tapeti.Connection
private void HandleBasicAck(object sender, BasicAckEventArgs e)
{
foreach (var deliveryTag in GetDeliveryTags(e))
Monitor.Enter(confirmLock);
try
{
if (!confirmMessages.TryGetValue(deliveryTag, out var messageInfo))
continue;
if (returnRoutingKeys.TryGetValue(messageInfo.ReturnKey, out var returnInfo))
foreach (var deliveryTag in GetDeliveryTags(e))
{
messageInfo.CompletionSource.SetResult(returnInfo.FirstReplyCode);
if (!confirmMessages.TryGetValue(deliveryTag, out var messageInfo))
continue;
returnInfo.RefCount--;
if (returnInfo.RefCount == 0)
returnRoutingKeys.Remove(messageInfo.ReturnKey);
if (returnRoutingKeys.TryGetValue(messageInfo.ReturnKey, out var returnInfo))
{
messageInfo.CompletionSource.SetResult(returnInfo.FirstReplyCode);
returnInfo.RefCount--;
if (returnInfo.RefCount == 0)
returnRoutingKeys.Remove(messageInfo.ReturnKey);
}
messageInfo.CompletionSource.SetResult(0);
confirmMessages.Remove(deliveryTag);
}
messageInfo.CompletionSource.SetResult(0);
confirmMessages.Remove(deliveryTag);
}
finally
{
Monitor.Exit(confirmLock);
}
}
private void HandleBasicNack(object sender, BasicNackEventArgs e)
{
foreach (var deliveryTag in GetDeliveryTags(e))
Monitor.Enter(confirmLock);
try
{
if (!confirmMessages.TryGetValue(deliveryTag, out var messageInfo))
continue;
foreach (var deliveryTag in GetDeliveryTags(e))
{
if (!confirmMessages.TryGetValue(deliveryTag, out var messageInfo))
continue;
messageInfo.CompletionSource.SetCanceled();
confirmMessages.Remove(e.DeliveryTag);
messageInfo.CompletionSource.SetCanceled();
confirmMessages.Remove(e.DeliveryTag);
}
}
finally
{
Monitor.Exit(confirmLock);
}
}