GitHub-Mirror/Tapeti
1
0
Fork 0
mirror of synced 2024-11-05 02:59:16 +00:00
Code Issues Projects Releases Wiki Activity

Merge branch 'release/2.8'

Browse Source
This commit is contained in:
Mark van Renswoude 2021-10-07 15:59:59 +02:00
commit 5703062054
63 changed files with 646 additions and 3510 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
Examples/08-MessageHandlerLogging/08-MessageHandlerLogging.csproj Normal file
View File

@ -0,0 +1,22 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>netcoreapp3.1</TargetFramework>
<RootNamespace>_08_MessageHandlerLogging</RootNamespace>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Serilog.Sinks.Console" Version="4.0.0" />
<PackageReference Include="SimpleInjector" Version="5.3.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Tapeti.Serilog\Tapeti.Serilog.csproj" />
<ProjectReference Include="..\ExampleLib\ExampleLib.csproj" />
<ProjectReference Include="..\Messaging.TapetiExample\Messaging.TapetiExample.csproj" />
<ProjectReference Include="..\..\Tapeti.SimpleInjector\Tapeti.SimpleInjector.csproj" />
<ProjectReference Include="..\..\Tapeti\Tapeti.csproj" />
</ItemGroup>
</Project>

62
Examples/08-MessageHandlerLogging/Program.cs Normal file
View File

@ -0,0 +1,62 @@
using System;
using System.Threading.Tasks;
using ExampleLib;
using Messaging.TapetiExample;
using Serilog;
using Serilog.Events;
using SimpleInjector;
using Tapeti;
using Tapeti.Serilog;
using Tapeti.SimpleInjector;
using ILogger = Tapeti.ILogger;
namespace _08_MessageHandlerLogging
{
public class Program
{
public static void Main()
{
var container = new Container();
var dependencyResolver = new SimpleInjectorDependencyResolver(container);
var seriLogger = new LoggerConfiguration()
.MinimumLevel.Debug()
// Include {Properties} or specific keys in the output template to see properties added to the diagnostic context
.WriteTo.Console(outputTemplate: "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj} {Properties}{NewLine}{Exception}")
.CreateLogger();
container.RegisterInstance((Serilog.ILogger)seriLogger);
container.Register<ILogger, TapetiSeriLogger.WithMessageLogging>();
var helper = new ExampleConsoleApp(dependencyResolver);
helper.Run(MainAsync);
seriLogger.Dispose();
}
internal static async Task MainAsync(IDependencyResolver dependencyResolver, Func<Task> waitForDone)
{
var config = new TapetiConfig(dependencyResolver)
.WithMessageHandlerLogging()
.RegisterAllControllers()
.Build();
await using var connection = new TapetiConnection(config);
var subscriber = await connection.Subscribe(false);
var publisher = dependencyResolver.Resolve<IPublisher>();
await publisher.Publish(new PublishSubscribeMessage
{
Greeting = "Hello message handler logging!"
});
await subscriber.Resume();
await waitForDone();
}
}
}

31
Examples/08-MessageHandlerLogging/SlowMessageController.cs Normal file
View File

@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
using ExampleLib;
using Messaging.TapetiExample;
using Tapeti.Annotations;
namespace _08_MessageHandlerLogging
{
[MessageController]
[DynamicQueue("tapeti.example.08.slow")]
public class SlowMessageController
{
private readonly IExampleState exampleState;
public SlowMessageController(IExampleState exampleState)
{
this.exampleState = exampleState;
}
public async Task GimmeASec(PublishSubscribeMessage message)
{
Console.WriteLine("Received message (in slow controller): " + message.Greeting);
await Task.Delay(1000);
exampleState.Done();
}
}
}

20
Examples/08-MessageHandlerLogging/SpeedyMessageController.cs Normal file
View File

@ -0,0 +1,20 @@
using System;
using Messaging.TapetiExample;
using Tapeti.Annotations;
using Tapeti.Serilog;
namespace _08_MessageHandlerLogging
{
[MessageController]
[DynamicQueue("tapeti.example.08.speedy")]
public class SpeedyMessageController
{
// ReSharper disable once InconsistentNaming
public void IAmSpeed(PublishSubscribeMessage message, IDiagnosticContext diagnosticContext)
{
diagnosticContext.Set("PropertySetByMessageHandler", 42);
Console.WriteLine("Received message (in speedy controller): " + message.Greeting);
}
}
}

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

@ -6,10 +6,7 @@
<ItemGroup>
<PackageReference Include="System.ComponentModel.Annotations" Version="5.0.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Tapeti.Annotations\Tapeti.Annotations.csproj" />
<PackageReference Include="Tapeti.Annotations" Version="3.0.0" />
</ItemGroup>
</Project>

12
README.md
View File

@ -1,3 +1,4 @@
## Introduction
Tapeti is a wrapper for the RabbitMQ .NET Client designed for long-running microservices. It’s main goal is to minimize the amount of messaging code required, and instead focus on the higher-level flow.
@ -47,6 +48,17 @@ The documentation for Tapeti is available on Read the Docs:
[![Documentation Status](https://readthedocs.org/projects/tapeti/badge/?version=latest)](http://tapeti.readthedocs.io/en/latest/introduction.html?badge=latest)
## Related repositories
Parts of Tapeti have been split into their own repository. This allows them to have their own version numbers which increases compatibility between shared message packages when services use different Tapeti versions.
- [Tapeti.Annotations](https://github.com/MvRens/Tapeti.Annotations)
annotations used in message classes, like [Request]
- [Tapeti.DataAnnotations.Extensions](https://github.com/MvRens/Tapet.DataAnnotations.Extensions)
generic data annotation attributes useful in messages, like the [RequiredGuid] attribute
- [Tapeti.Cmd](https://github.com/MvRens/Tapeti.Cmd)
the standalone tool to manage RabbitMQ messages and queues, focused on Tapeti-compatible messages
## Builds
Builds are automatically run using AppVeyor, with the resulting packages being pushed to NuGet.

28
Tapeti.Annotations/DurableQueueAttribute.cs
View File

@ -1,28 +0,0 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Annotations
{
/// <inheritdoc />
/// <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)]
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.Annotations/DynamicQueueAttribute.cs
View File

@ -1,32 +0,0 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Annotations
{
/// <inheritdoc />
/// <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]
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;
}
}
}

15
Tapeti.Annotations/MandatoryAttribute.cs
View File

@ -1,15 +0,0 @@
using System;
namespace Tapeti.Annotations
{
/// <inheritdoc />
/// <summary>
/// Can be attached to a message class to specify that delivery to a queue must be guaranteed.
/// Publish will fail if no queues bind to the routing key. Publisher confirms must be enabled
/// on the connection for this to work (enabled by default).
/// </summary>
[AttributeUsage(AttributeTargets.Class)]
public class MandatoryAttribute : Attribute
{
}
}

16
Tapeti.Annotations/MessageControllerAttribute.cs
View File

@ -1,16 +0,0 @@
using System;
using JetBrains.Annotations;
namespace Tapeti.Annotations
{
/// <inheritdoc />
/// <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)]
public class MessageControllerAttribute : Attribute
{
}
}

180
Tapeti.Annotations/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;
// ReSharper disable InheritdocConsiderUsage
#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; }
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([NotNull] string comment)
{
Comment = comment;
}
[CanBeNull] public string Comment { get; }
}
}

21
Tapeti.Annotations/RequestAttribute.cs
View File

@ -1,21 +0,0 @@
using System;
namespace Tapeti.Annotations
{
/// <inheritdoc />
/// <summary>
/// Can be attached to a message class to specify that the receiver of the message must
/// provide a response message of the type specified in the Response attribute. This response
/// must be sent by either returning it from the message handler method or using
/// EndWithResponse when using Tapeti Flow. These methods will respond directly
/// to the queue specified in the reply-to header automatically added by Tapeti.
/// </summary>
[AttributeUsage(AttributeTargets.Class)]
public class RequestAttribute : Attribute
{
/// <summary>
/// The type of the message class which must be returned as the response.
/// </summary>
public Type Response { get; set; }
}
}

14
Tapeti.Annotations/ResponseHandlerAttribute.cs
View File

@ -1,14 +0,0 @@
using System;
namespace Tapeti.Annotations
{
/// <inheritdoc />
/// <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)]
public class ResponseHandlerAttribute : Attribute
{
}
}

52
Tapeti.Annotations/RoutingKeyAttribute.cs
View File

@ -1,52 +0,0 @@
using System;
namespace Tapeti.Annotations
{
/// <inheritdoc />
/// <summary>
/// Can be attached to a message class to override or extend the generated routing key.
/// One of the intended scenarios is for versioning messages, where you want to add for example a
/// ".v2" postfix but keep the class name itself the same (in a different namespace of course).
/// </summary>
/// <remarks>
/// The implementation of IRoutingKeyStrategy must explicitly add support for this attribute for it to have effect.
/// The default implementation (TypeNameRoutingKeyStrategy) does, of course. Custom implementations can use
/// the Tapeti.Helpers.RoutingKeyHelper class to add support and keep up-to-date automatically.
///
/// When using EnableDeclareDurableQueues to automatically generate the queue bindings, both the sender and receiver must
/// use a Tapeti version >= 2.5 to support this attribute or the binding will differ from the sent routing key.
///
/// The routing keys are no longer used by Tapeti once the message is in the queue, the message handler
/// is instead based on the full message class name (thus including namespace), so if the binding is generated in any
/// other way this remark does not apply and prior versions of Tapeti are compatible.
/// </remarks>
[AttributeUsage(AttributeTargets.Class)]
public class RoutingKeyAttribute : Attribute
{
/// <summary>
/// If specified, the routing key strategy is skipped altogether and this value is used instead.
/// </summary>
/// <remarks>
/// The Prefix and Postfix properties will not have any effect if the Full property is specified.
/// </remarks>
public string Full { get; set; }
/// <summary>
/// If specified, the value generated by the default routing key strategy is prefixed with this value.
/// No dot will be added after this prefix, if you want to include it add it as part of the value.
/// </summary>
/// <remarks>
/// This property will not have any effect if the Full property is specified. Can be used in combination with Postfix.
/// </remarks>
public string Prefix { get; set; }
/// <summary>
/// If specified, the value generated by the default routing key strategy is postfixed with this value.
/// No dot will be added before this postfix, if you want to include it add it as part of the value.
/// </summary>
/// <remarks>
/// This property will not have any effect if the Full property is specified. Can be used in combination with Prefix.
/// </remarks>
public string Postfix { get; set; }
}
}

30
Tapeti.Annotations/Tapeti.Annotations.csproj
View File

@ -1,30 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<Authors>Mark van Renswoude</Authors>
<Company />
<Description>Annotations for Tapeti</Description>
<PackageTags>rabbitmq tapeti</PackageTags>
<PackageLicenseExpression>Unlicense</PackageLicenseExpression>
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.Annotations.png</PackageIcon>
<Version>2.0.0</Version>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'">
<NoWarn>1701;1702</NoWarn>
</PropertyGroup>
<ItemGroup>
<None Include="..\resources\icons\Tapeti.Annotations.png">
<Pack>True</Pack>
<PackagePath></PackagePath>
</None>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
</ItemGroup>
</Project>

324
Tapeti.Cmd/ConsoleHelper/ConsoleWrapper.cs
View File

@ -1,324 +0,0 @@
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using Console = System.Console;
namespace Tapeti.Cmd.ConsoleHelper
{
public class ConsoleWrapper : IConsole
{
private readonly List<TemporaryWriter> temporaryWriters = new();
private bool temporaryActive;
private int temporaryCursorTop;
public ConsoleWrapper()
{
temporaryCursorTop = Console.CursorTop;
Console.CancelKeyPress += (_, args) =>
{
if (Cancelled)
return;
using var consoleWriter = GetPermanentWriter();
consoleWriter.WriteLine("Cancelling...");
args.Cancel = true;
Cancelled = true;
};
}
public void Dispose()
{
foreach (var writer in temporaryWriters)
writer.Dispose();
Console.CursorVisible = true;
GC.SuppressFinalize(this);
}
public bool Cancelled { get; private set; }
public IConsoleWriter GetPermanentWriter()
{
return new PermanentWriter(this);
}
public IConsoleWriter GetTemporaryWriter()
{
var writer = new TemporaryWriter(this, temporaryWriters.Count);
temporaryWriters.Add(writer);
return writer;
}
private void AcquirePermanent()
{
if (!temporaryActive)
return;
foreach (var writer in temporaryWriters)
{
Console.SetCursorPosition(0, temporaryCursorTop + writer.RelativePosition);
writer.Clear();
}
Console.SetCursorPosition(0, temporaryCursorTop);
Console.CursorVisible = true;
temporaryActive = false;
}
private void ReleasePermanent()
{
if (temporaryWriters.Count == 0)
{
temporaryCursorTop = Console.CursorTop;
return;
}
foreach (var writer in temporaryWriters)
{
writer.Restore();
Console.WriteLine();
}
// Store the cursor position afterwards to account for buffer scrolling
temporaryCursorTop = Console.CursorTop - temporaryWriters.Count;
Console.CursorVisible = false;
temporaryActive = true;
}
private void AcquireTemporary(TemporaryWriter writer)
{
Console.SetCursorPosition(0, temporaryCursorTop + writer.RelativePosition);
if (temporaryActive)
return;
Console.CursorVisible = false;
temporaryActive = true;
}
private void DisposeWriter(BaseWriter writer)
{
if (writer is not TemporaryWriter temporaryWriter)
return;
Console.SetCursorPosition(0, temporaryCursorTop + temporaryWriter.RelativePosition);
temporaryWriter.Clear();
temporaryWriters.Remove(temporaryWriter);
}
private abstract class BaseWriter : IConsoleWriter
{
protected readonly ConsoleWrapper Owner;
protected BaseWriter(ConsoleWrapper owner)
{
Owner = owner;
}
public virtual void Dispose()
{
Owner.DisposeWriter(this);
GC.SuppressFinalize(this);
}
public abstract bool Enabled { get; }
public abstract void WriteCaptured(string value, Action processInput);
public abstract void WriteLine(string value);
public void Confirm(string message)
{
WriteLine(message);
// Clear any previous key entered before this confirmation
while (!Owner.Cancelled && Console.KeyAvailable)
Console.ReadKey(true);
while (!Owner.Cancelled && !Console.KeyAvailable)
Thread.Sleep(50);
if (Owner.Cancelled)
return;
Console.ReadKey(true);
}
public bool ConfirmYesNo(string message)
{
var confirmed = false;
WriteCaptured($"{message} (Y/N) ", () =>
{
// Clear any previous key entered before this confirmation
while (!Owner.Cancelled && Console.KeyAvailable)
Console.ReadKey(true);
var input = new StringBuilder();
while (!Owner.Cancelled)
{
if (!Console.KeyAvailable)
{
Thread.Sleep(50);
continue;
}
var keyInfo = Console.ReadKey(false);
// ReSharper disable once SwitchStatementHandlesSomeKnownEnumValuesWithDefault - by design
switch (keyInfo.Key)
{
case ConsoleKey.Enter:
Console.WriteLine();
confirmed = input.ToString().Equals("Y", StringComparison.CurrentCultureIgnoreCase);
return;
case ConsoleKey.Backspace:
if (input.Length > 0)
{
input.Remove(input.Length - 1, 1);
// We need to handle erasing the character ourselves, as we want to use ReadKey so that we can monitor Cancelled
Console.Write(" \b");
}
break;
default:
if (keyInfo.KeyChar != -1)
input.Append(keyInfo.KeyChar);
break;
}
}
});
return confirmed;
}
}
private class PermanentWriter : BaseWriter
{
public PermanentWriter(ConsoleWrapper owner) : base(owner)
{
}
public override bool Enabled => true;
public override void WriteCaptured(string value, Action waitForInput)
{
Owner.AcquirePermanent();
try
{
Console.Write(value);
waitForInput();
}
finally
{
Owner.ReleasePermanent();
}
}
public override void WriteLine(string value)
{
Owner.AcquirePermanent();
try
{
Console.WriteLine(value);
}
finally
{
Owner.ReleasePermanent();
}
}
}
private class TemporaryWriter : BaseWriter
{
public int RelativePosition { get; }
private bool isActive;
private string storedValue;
public TemporaryWriter(ConsoleWrapper owner, int relativePosition) : base(owner)
{
RelativePosition = relativePosition;
}
public override bool Enabled => !Console.IsOutputRedirected;
public override void WriteCaptured(string value, Action waitForInput)
{
WriteLine(value);
waitForInput();
}
public override void WriteLine(string value)
{
if (!Enabled)
return;
Owner.AcquireTemporary(this);
Console.Write(value);
if (!string.IsNullOrEmpty(storedValue) && storedValue.Length > value.Length)
// Clear characters remaining from the previous value
Console.Write(new string(' ', storedValue.Length - value.Length));
storedValue = value;
isActive = true;
}
public void Clear()
{
if (!isActive)
return;
if (!string.IsNullOrEmpty(storedValue))
Console.Write(new string(' ', storedValue.Length));
isActive = false;
}
public void Restore()
{
if (string.IsNullOrEmpty(storedValue))
return;
Console.Write(storedValue);
isActive = true;
}
}
}
}

44
Tapeti.Cmd/ConsoleHelper/IConsole.cs
View File

@ -1,44 +0,0 @@
using System;
namespace Tapeti.Cmd.ConsoleHelper
{
/// <summary>
/// Wraps access to the console to provide cooperation between temporary outputs like the
/// progress bar and batch confirmations. Temporary outputs hide the cursor and will be
/// automatically be erased and restored when a permanent writer is called.
/// </summary>
/// <remarks>
/// Temporary outputs are automatically supressed when the console output is redirected.
/// The Enabled property will reflect this.
/// </remarks>
public interface IConsole : IDisposable
{
bool Cancelled { get; }
IConsoleWriter GetPermanentWriter();
IConsoleWriter GetTemporaryWriter();
}
/// <summary>
/// For simplicity outputs only support one line of text.
/// For temporary writers, each call to WriteLine will overwrite the previous and clear any
/// extra characters if the previous value was longer.
/// </summary>
public interface IConsoleWriter : IDisposable
{
bool Enabled { get; }
void WriteLine(string value);
/// <summary>
/// Waits for any user input.
/// </summary>
void Confirm(string message);
/// <summary>
/// Waits for user confirmation (Y/N).
/// </summary>
bool ConfirmYesNo(string message);
}
}

90
Tapeti.Cmd/ConsoleHelper/ProgressBar.cs
View File

@ -1,90 +0,0 @@
using System;
using System.Text;
namespace Tapeti.Cmd.ConsoleHelper
{
public class ProgressBar : IDisposable, IProgress<int>
{
private static readonly TimeSpan UpdateInterval = TimeSpan.FromMilliseconds(20);
private readonly IConsoleWriter consoleWriter;
private readonly int max;
private readonly int width;
private readonly bool showPosition;
private int position;
private readonly bool enabled;
private DateTime lastUpdate = DateTime.MinValue;
public ProgressBar(IConsole console, int max, int width = 10, bool showPosition = true)
{
if (width <= 0)
throw new ArgumentOutOfRangeException(nameof(width), "Width must be greater than zero");
if (max <= 0)
throw new ArgumentOutOfRangeException(nameof(max), "Max must be greater than zero");
consoleWriter = console.GetTemporaryWriter();
this.max = max;
this.width = width;
this.showPosition = showPosition;
enabled = consoleWriter.Enabled;
if (!enabled)
return;
Redraw();
}
public void Dispose()
{
consoleWriter.Dispose();
}
public void Report(int value)
{
if (!enabled)
return;
value = Math.Max(0, Math.Min(max, value));
position = value;
var now = DateTime.Now;
if (now - lastUpdate < UpdateInterval)
return;
lastUpdate = now;
Redraw();
}
private void Redraw()
{
var output = new StringBuilder("[");
var blockCount = (int)Math.Truncate((decimal)position / max * width);
if (blockCount > 0)
output.Append(new string('#', blockCount));
if (blockCount < width)
output.Append(new string('.', width - blockCount));
output.Append("] ");
if (showPosition)
{
output
.Append(position.ToString("N0")).Append(" / ").Append(max.ToString("N0"))
.Append(" (").Append((int) Math.Truncate((decimal) position / max * 100)).Append("%)");
}
else
output.Append(" ").Append((int)Math.Truncate((decimal)position / max * 100)).Append("%");
consoleWriter.WriteLine(output.ToString());
}
}
}

166
Tapeti.Cmd/Mock/MockBasicProperties.cs
View File

@ -1,166 +0,0 @@
using System;
using System.Collections.Generic;
using RabbitMQ.Client;
namespace Tapeti.Cmd.Mock
{
public class MockBasicProperties : IBasicProperties
{
public ushort ProtocolClassId { get; set; }
public string ProtocolClassName { get; set; }
public void ClearAppId()
{
throw new NotImplementedException();
}
public void ClearClusterId()
{
throw new NotImplementedException();
}
public void ClearContentEncoding()
{
throw new NotImplementedException();
}
public void ClearContentType()
{
throw new NotImplementedException();
}
public void ClearCorrelationId()
{
throw new NotImplementedException();
}
public void ClearDeliveryMode()
{
throw new NotImplementedException();
}
public void ClearExpiration()
{
throw new NotImplementedException();
}
public void ClearHeaders()
{
throw new NotImplementedException();
}
public void ClearMessageId()
{
throw new NotImplementedException();
}
public void ClearPriority()
{
throw new NotImplementedException();
}
public void ClearReplyTo()
{
throw new NotImplementedException();
}
public void ClearTimestamp()
{
throw new NotImplementedException();
}
public void ClearType()
{
throw new NotImplementedException();
}
public void ClearUserId()
{
throw new NotImplementedException();
}
public bool IsAppIdPresent() => appIdPresent;
public bool IsClusterIdPresent() => clusterIdPresent;
public bool IsContentEncodingPresent() => contentEncodingPresent;
public bool IsContentTypePresent() => contentTypePresent;
public bool IsCorrelationIdPresent() => correlationIdPresent;
public bool IsDeliveryModePresent() => deliveryModePresent;
public bool IsExpirationPresent() => expirationPresent;
public bool IsHeadersPresent() => headersPresent;
public bool IsMessageIdPresent() => messageIdPresent;
public bool IsPriorityPresent() => priorityPresent;
public bool IsReplyToPresent() => replyToPresent;
public bool IsTimestampPresent() => timestampPresent;
public bool IsTypePresent() => typePresent;
public bool IsUserIdPresent() => userIdPresent;
private bool appIdPresent;
private string appId;
private bool clusterIdPresent;
private string clusterId;
private bool contentEncodingPresent;
private string contentEncoding;
private bool contentTypePresent;
private string contentType;
private bool correlationIdPresent;
private string correlationId;
private bool deliveryModePresent;
private byte deliveryMode;
private bool expirationPresent;
private string expiration;
private bool headersPresent;
private IDictionary<string, object> headers;
private bool messageIdPresent;
private string messageId;
private bool priorityPresent;
private byte priority;
private bool replyToPresent;
private string replyTo;
private bool timestampPresent;
private AmqpTimestamp timestamp;
private bool typePresent;
private string type;
private bool userIdPresent;
private string userId;
public string AppId { get => appId; set => SetValue(out appId, out appIdPresent, value); }
public string ClusterId { get => clusterId; set => SetValue(out clusterId, out clusterIdPresent, value); }
public string ContentEncoding { get => contentEncoding; set => SetValue(out contentEncoding, out contentEncodingPresent, value); }
public string ContentType { get => contentType; set => SetValue(out contentType, out contentTypePresent, value); }
public string CorrelationId { get => correlationId; set => SetValue(out correlationId, out correlationIdPresent, value); }
public byte DeliveryMode { get => deliveryMode; set => SetValue(out deliveryMode, out deliveryModePresent, value); }
public string Expiration { get => expiration; set => SetValue(out expiration, out expirationPresent, value); }
public IDictionary<string, object> Headers { get => headers; set => SetValue(out headers, out headersPresent, value); }
public string MessageId { get => messageId; set => SetValue(out messageId, out messageIdPresent, value); }
public bool Persistent { get; set; }
public byte Priority { get => priority; set => SetValue(out priority, out priorityPresent, value); }
public string ReplyTo { get => replyTo; set => SetValue(out replyTo, out replyToPresent, value); }
public PublicationAddress ReplyToAddress { get; set; }
public AmqpTimestamp Timestamp { get => timestamp; set => SetValue(out timestamp, out timestampPresent, value); }
public string Type { get => type; set => SetValue(out type, out typePresent, value); }
public string UserId { get => userId; set => SetValue(out userId, out userIdPresent, value); }
private static void SetValue<T>(out T field, out bool present, T value)
{
field = value;
present = true;
}
}
}

23
Tapeti.Cmd/Parser/BindingParser.cs
View File

@ -1,23 +0,0 @@
using System;
using System.Collections.Generic;
using System.Linq;
namespace Tapeti.Cmd.Parser
{
public static class BindingParser
{
public static Tuple<string, string>[] Parse(IEnumerable<string> bindings)
{
return bindings
.Select(b =>
{
var parts = b.Split(':');
if (parts.Length != 2)
throw new InvalidOperationException($"Invalid binding format: {b}");
return new Tuple<string, string>(parts[0], parts[1]);
})
.ToArray();
}
}
}

63
Tapeti.Cmd/Program.cs
View File

@ -1,63 +0,0 @@
using System;
using System.Diagnostics;
using System.Linq;
using System.Reflection;
using CommandLine;
using Tapeti.Cmd.ConsoleHelper;
using Tapeti.Cmd.Verbs;
namespace Tapeti.Cmd
{
public class Program
{
public static int Main(string[] args)
{
var exitCode = 1;
var verbTypes = Assembly.GetExecutingAssembly().GetTypes()
.Where(t => t.GetCustomAttribute<ExecutableVerbAttribute>() != null)
.ToArray();
using var consoleWrapper = new ConsoleWrapper();
// ReSharper disable AccessToDisposedClosure
CommandLine.Parser.Default.ParseArguments(args, verbTypes.ToArray())
.WithParsed(o =>
{
try
{
var executableVerbAttribute = o.GetType().GetCustomAttribute<ExecutableVerbAttribute>();
var executer = Activator.CreateInstance(executableVerbAttribute.VerbExecuter, o) as IVerbExecuter;
// Should have been validated by the ExecutableVerbAttribute
Debug.Assert(executer != null, nameof(executer) + " != null");
executer.Execute(consoleWrapper);
exitCode = 0;
}
catch (Exception e)
{
using var consoleWriter = consoleWrapper.GetPermanentWriter();
consoleWriter.WriteLine(e.Message);
DebugConfirmClose(consoleWrapper);
}
})
.WithNotParsed(_ =>
{
DebugConfirmClose(consoleWrapper);
});
// ReSharper restore AccessToDisposedClosure
return exitCode;
}
private static void DebugConfirmClose(IConsole console)
{
if (!Debugger.IsAttached)
return;
using var consoleWriter = console.GetPermanentWriter();
consoleWriter.Confirm("Press any key to continue...");
}
}
}

16
Tapeti.Cmd/Properties/PublishProfiles/FolderProfile.pubxml
View File

@ -1,16 +0,0 @@
<?xml version="1.0" encoding="utf-8"?>
<!--
https://go.microsoft.com/fwlink/?LinkID=208121.
-->
<Project ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<PropertyGroup>
<PublishProtocol>FileSystem</PublishProtocol>
<Configuration>Release</Configuration>
<Platform>Any CPU</Platform>
<TargetFramework>netcoreapp2.1</TargetFramework>
<PublishDir>bin\Release\netcoreapp2.1\publish\</PublishDir>
<RuntimeIdentifier>win-x64</RuntimeIdentifier>
<SelfContained>false</SelfContained>
<_IsPortable>true</_IsPortable>
</PropertyGroup>
</Project>

80
Tapeti.Cmd/RateLimiter/BatchSizeRateLimiter.cs
View File

@ -1,80 +0,0 @@
using System;
using System.Threading;
using Tapeti.Cmd.ConsoleHelper;
namespace Tapeti.Cmd.RateLimiter
{
public abstract class BaseBatchSizeRateLimiter : IRateLimiter
{
private readonly IConsole console;
private readonly IRateLimiter decoratedRateLimiter;
private readonly int batchSize;
private int batchCount;
protected BaseBatchSizeRateLimiter(IConsole console, IRateLimiter decoratedRateLimiter, int batchSize)
{
this.console = console;
this.decoratedRateLimiter = decoratedRateLimiter;
this.batchSize = batchSize;
}
public void Execute(Action action)
{
batchCount++;
if (batchCount > batchSize)
{
Pause(console);
batchCount = 1;
}
decoratedRateLimiter.Execute(action);
}
protected abstract void Pause(IConsole console);
}
public class ManualBatchSizeRateLimiter : BaseBatchSizeRateLimiter
{
public ManualBatchSizeRateLimiter(IConsole console, IRateLimiter decoratedRateLimiter, int batchSize) : base(console, decoratedRateLimiter, batchSize)
{
}
protected override void Pause(IConsole console)
{
using var consoleWriter = console.GetTemporaryWriter();
consoleWriter.Confirm("Press any key to continue with the next batch...");
}
}
public class TimedBatchSizeRateLimiter : BaseBatchSizeRateLimiter
{
private readonly int batchPauseTime;
public TimedBatchSizeRateLimiter(IConsole console, IRateLimiter decoratedRateLimiter, int batchSize, int batchPauseTime) : base(console, decoratedRateLimiter, batchSize)
{
this.batchPauseTime = batchPauseTime;
}
protected override void Pause(IConsole console)
{
using var consoleWriter = console.GetTemporaryWriter();
var remaining = batchPauseTime;
while (remaining > 0 && !console.Cancelled)
{
consoleWriter.WriteLine($"Next batch in {remaining} second{(remaining != 1 ? "s" : "")}...");
Thread.Sleep(1000);
remaining--;
}
}
}
}

9
Tapeti.Cmd/RateLimiter/IRateLimiter.cs
View File

@ -1,9 +0,0 @@
using System;
namespace Tapeti.Cmd.RateLimiter
{
public interface IRateLimiter
{
void Execute(Action action);
}
}

12
Tapeti.Cmd/RateLimiter/NoRateLimiter.cs
View File

@ -1,12 +0,0 @@
using System;
namespace Tapeti.Cmd.RateLimiter
{
public class NoRateLimiter : IRateLimiter
{
public void Execute(Action action)
{
action();
}
}
}

29
Tapeti.Cmd/RateLimiter/RateLimiterFactory.cs
View File

@ -1,29 +0,0 @@
using System;
using Tapeti.Cmd.ConsoleHelper;
namespace Tapeti.Cmd.RateLimiter
{
public static class RateLimiterFactory
{
public static IRateLimiter Create(IConsole console, int? maxRate, int? batchSize, int? batchPauseTime)
{
IRateLimiter rateLimiter;
if (maxRate > 0)
rateLimiter = new SpreadRateLimiter(maxRate.Value, TimeSpan.FromSeconds(1));
else
rateLimiter = new NoRateLimiter();
// ReSharper disable once InvertIf - I don't like the readability of that flow
if (batchSize > 0)
{
if (batchPauseTime > 0)
rateLimiter = new TimedBatchSizeRateLimiter(console, rateLimiter, batchSize.Value, batchPauseTime.Value);
else
rateLimiter = new ManualBatchSizeRateLimiter(console, rateLimiter, batchSize.Value);
}
return rateLimiter;
}
}
}

30
Tapeti.Cmd/RateLimiter/SpreadRateLimiter.cs
View File

@ -1,30 +0,0 @@
using System;
using System.Threading;
namespace Tapeti.Cmd.RateLimiter
{
public class SpreadRateLimiter : IRateLimiter
{
private readonly TimeSpan delay;
private DateTime lastExecute = DateTime.MinValue;
public SpreadRateLimiter(int amount, TimeSpan perTimespan)
{
delay = TimeSpan.FromMilliseconds(perTimespan.TotalMilliseconds / amount);
}
public void Execute(Action action)
{
// Very simple implementation; the time between actions must be at least the delay.
// This prevents bursts followed by nothing which are common with normal rate limiter implementations.
var remainingWaitTime = delay - (DateTime.Now - lastExecute);
if (remainingWaitTime.TotalMilliseconds > 0)
Thread.Sleep(remainingWaitTime);
action();
lastExecute = DateTime.Now;
}
}
}

333
Tapeti.Cmd/Serialization/EasyNetQMessageSerializer.cs
View File

@ -1,333 +0,0 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Text.RegularExpressions;
using Newtonsoft.Json;
using RabbitMQ.Client;
namespace Tapeti.Cmd.Serialization
{
public class EasyNetQMessageSerializer : IMessageSerializer
{
private static readonly Regex InvalidCharRegex = new(@"[\\\/:\*\?\""\<\>|]", RegexOptions.Compiled);
private readonly Lazy<string> writablePath;
private int messageCount;
private readonly Lazy<string[]> files;
public EasyNetQMessageSerializer(string path)
{
writablePath = new Lazy<string>(() =>
{
Directory.CreateDirectory(path);
return path;
});
files = new Lazy<string[]>(() => Directory.GetFiles(path, "*.*.message.txt"));
}
public static bool OutputExists(string path)
{
return Directory.Exists(path) && Directory.GetFiles(path, "*.message.txt").Length > 0;
}
public void Dispose()
{
}
public void Serialize(Message message)
{
var uniqueFileName = SanitiseQueueName(message.Queue) + "." + messageCount;
var bodyPath = Path.Combine(writablePath.Value, uniqueFileName + ".message.txt");
var propertiesPath = Path.Combine(writablePath.Value, uniqueFileName + ".properties.txt");
var infoPath = Path.Combine(writablePath.Value, uniqueFileName + ".info.txt");
var properties = new EasyNetQMessageProperties(message.Properties);
var info = new EasyNetQMessageReceivedInfo(message);
File.WriteAllText(bodyPath, Encoding.UTF8.GetString(message.Body));
File.WriteAllText(propertiesPath, JsonConvert.SerializeObject(properties));
File.WriteAllText(infoPath, JsonConvert.SerializeObject(info));
messageCount++;
}
private static string SanitiseQueueName(string queueName)
{
return InvalidCharRegex.Replace(queueName, "_");
}
public int GetMessageCount()
{
return files.Value.Length;
}
public IEnumerable<Message> Deserialize(IModel channel)
{
foreach (var file in files.Value)
{
const string messageTag = ".message.";
var directoryName = Path.GetDirectoryName(file);
var fileName = Path.GetFileName(file);
var propertiesFileName = Path.Combine(directoryName, fileName.Replace(messageTag, ".properties."));
var infoFileName = Path.Combine(directoryName, fileName.Replace(messageTag, ".info."));
var body = File.ReadAllText(file);
var propertiesJson = File.ReadAllText(propertiesFileName);
var properties = JsonConvert.DeserializeObject<EasyNetQMessageProperties>(propertiesJson);
var infoJson = File.ReadAllText(infoFileName);
var info = JsonConvert.DeserializeObject<EasyNetQMessageReceivedInfo>(infoJson);
if (info == null)
continue;
var message = info.ToMessage();
if (properties != null)
message.Properties = properties.ToBasicProperties(channel);
message.Body = Encoding.UTF8.GetBytes(body);
yield return message;
}
}
// ReSharper disable MemberCanBePrivate.Local - used by JSON deserialization
// ReSharper disable AutoPropertyCanBeMadeGetOnly.Local
private class EasyNetQMessageProperties
{
// ReSharper disable once MemberCanBePrivate.Local - used by JSON deserialization
public EasyNetQMessageProperties()
{
}
public EasyNetQMessageProperties(IBasicProperties basicProperties) : this()
{
if (basicProperties.IsContentTypePresent()) ContentType = basicProperties.ContentType;
if (basicProperties.IsContentEncodingPresent()) ContentEncoding = basicProperties.ContentEncoding;
if (basicProperties.IsDeliveryModePresent()) DeliveryMode = basicProperties.DeliveryMode;
if (basicProperties.IsPriorityPresent()) Priority = basicProperties.Priority;
if (basicProperties.IsCorrelationIdPresent()) CorrelationId = basicProperties.CorrelationId;
if (basicProperties.IsReplyToPresent()) ReplyTo = basicProperties.ReplyTo;
if (basicProperties.IsExpirationPresent()) Expiration = basicProperties.Expiration;
if (basicProperties.IsMessageIdPresent()) MessageId = basicProperties.MessageId;
if (basicProperties.IsTimestampPresent()) Timestamp = basicProperties.Timestamp.UnixTime;
if (basicProperties.IsTypePresent()) Type = basicProperties.Type;
if (basicProperties.IsUserIdPresent()) UserId = basicProperties.UserId;
if (basicProperties.IsAppIdPresent()) AppId = basicProperties.AppId;
if (basicProperties.IsClusterIdPresent()) ClusterId = basicProperties.ClusterId;
if (!basicProperties.IsHeadersPresent())
return;
foreach (var (key, value) in basicProperties.Headers)
Headers.Add(key, (byte[])value);
}
public IBasicProperties ToBasicProperties(IModel channel)
{
var basicProperties = channel.CreateBasicProperties();
if (ContentTypePresent) basicProperties.ContentType = ContentType;
if (ContentEncodingPresent) basicProperties.ContentEncoding = ContentEncoding;
if (DeliveryModePresent) basicProperties.DeliveryMode = DeliveryMode;
if (PriorityPresent) basicProperties.Priority = Priority;
if (CorrelationIdPresent) basicProperties.CorrelationId = CorrelationId;
if (ReplyToPresent) basicProperties.ReplyTo = ReplyTo;
if (ExpirationPresent) basicProperties.Expiration = Expiration;
if (MessageIdPresent) basicProperties.MessageId = MessageId;
if (TimestampPresent) basicProperties.Timestamp = new AmqpTimestamp(Timestamp);
if (TypePresent) basicProperties.Type = Type;
if (UserIdPresent) basicProperties.UserId = UserId;
if (AppIdPresent) basicProperties.AppId = AppId;
if (ClusterIdPresent) basicProperties.ClusterId = ClusterId;
if (HeadersPresent)
{
basicProperties.Headers = new Dictionary<string, object>(Headers.ToDictionary(p => p.Key, p => (object)p.Value));
}
return basicProperties;
}
private string contentType;
public string ContentType
{
get => contentType;
set { contentType = value; ContentTypePresent = true; }
}
private string contentEncoding;
public string ContentEncoding
{
get => contentEncoding;
set { contentEncoding = value; ContentEncodingPresent = true; }
}
// The original EasyNetQ.Hosepipe defines this as an IDictionary<string, object>. This causes UTF-8 headers
// to be serialized as Base64, and deserialized as string, corrupting the republished message.
// This may cause incompatibilities, but fixes it for dumped Tapeti messages.
private IDictionary<string, byte[]> headers = new Dictionary<string, byte[]>();
public IDictionary<string, byte[]> Headers
{
get => headers;
// ReSharper disable once UnusedMember.Local
set { headers = value; HeadersPresent = true; }
}
private byte deliveryMode;
public byte DeliveryMode
{
get => deliveryMode;
set { deliveryMode = value; DeliveryModePresent = true; }
}
private byte priority;
public byte Priority
{
get => priority;
set { priority = value; PriorityPresent = true; }
}
private string correlationId;
public string CorrelationId
{
get => correlationId;
set { correlationId = value; CorrelationIdPresent = true; }
}
private string replyTo;
public string ReplyTo
{
get => replyTo;
set { replyTo = value; ReplyToPresent = true; }
}
private string expiration;
public string Expiration
{
get => expiration;
set { expiration = value; ExpirationPresent = true; }
}
private string messageId;
public string MessageId
{
get => messageId;
set { messageId = value; MessageIdPresent = true; }
}
private long timestamp;
public long Timestamp
{
get => timestamp;
set { timestamp = value; TimestampPresent = true; }
}
private string type;
public string Type
{
get => type;
set { type = value; TypePresent = true; }
}
private string userId;
public string UserId
{
get => userId;
set { userId = value; UserIdPresent = true; }
}
private string appId;
public string AppId
{
get => appId;
set { appId = value; AppIdPresent = true; }
}
private string clusterId;
public string ClusterId
{
get => clusterId;
set { clusterId = value; ClusterIdPresent = true; }
}
public bool ContentTypePresent { get; set; }
public bool ContentEncodingPresent { get; set; }
public bool HeadersPresent { get; set; } = true;
public bool DeliveryModePresent { get; set; }
public bool PriorityPresent { get; set; }
public bool CorrelationIdPresent { get; set; }
public bool ReplyToPresent { get; set; }
public bool ExpirationPresent { get; set; }
public bool MessageIdPresent { get; set; }
public bool TimestampPresent { get; set; }
public bool TypePresent { get; set; }
public bool UserIdPresent { get; set; }
public bool AppIdPresent { get; set; }
public bool ClusterIdPresent { get; set; }
}
private class EasyNetQMessageReceivedInfo
{
// ReSharper disable once UnusedAutoPropertyAccessor.Local - used by JSON deserialization
public string ConsumerTag { get; set; }
public ulong DeliverTag { get; set; }
public bool Redelivered { get; set; }
public string Exchange { get; set; }
public string RoutingKey { get; set; }
public string Queue { get; set; }
// ReSharper disable once MemberCanBePrivate.Local - used by JSON deserialization
// ReSharper disable once UnusedMember.Local
// ReSharper disable once UnusedMember.Global
public EasyNetQMessageReceivedInfo()
{
}
public EasyNetQMessageReceivedInfo(Message fromMessage)
{
ConsumerTag = "hosepipe";
DeliverTag = fromMessage.DeliveryTag;
Redelivered = fromMessage.Redelivered;
Exchange = fromMessage.Exchange;
RoutingKey = fromMessage.RoutingKey;
Queue = fromMessage.Queue;
}
public Message ToMessage()
{
return new()
{
//ConsumerTag =
DeliveryTag = DeliverTag,
Redelivered = Redelivered,
Exchange = Exchange,
RoutingKey = RoutingKey,
Queue = Queue
};
}
}
// ReSharper restore AutoPropertyCanBeMadeGetOnly.Local
// ReSharper restore MemberCanBePrivate.Local
}
}

26
Tapeti.Cmd/Serialization/IMessageSerializer.cs
View File

@ -1,26 +0,0 @@
using System;
using System.Collections.Generic;
using RabbitMQ.Client;
namespace Tapeti.Cmd.Serialization
{
public class Message
{
public ulong DeliveryTag;
public bool Redelivered;
public string Exchange;
public string RoutingKey;
public string Queue;
public IBasicProperties Properties;
public byte[] Body;
}
public interface IMessageSerializer : IDisposable
{
void Serialize(Message message);
int GetMessageCount();
IEnumerable<Message> Deserialize(IModel channel);
}
}

270
Tapeti.Cmd/Serialization/SingleFileJSONMessageSerializer.cs
View File

@ -1,270 +0,0 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
namespace Tapeti.Cmd.Serialization
{
public class SingleFileJSONMessageSerializer : IMessageSerializer
{
private readonly Stream stream;
private readonly bool disposeStream;
private readonly Encoding encoding;
// StreamReader.DefaultBufferSize is private :-/
private const int DefaultBufferSize = 1024;
private static readonly JsonSerializerSettings SerializerSettings = new()
{
NullValueHandling = NullValueHandling.Ignore
};
private readonly Lazy<StreamWriter> exportFile;
public SingleFileJSONMessageSerializer(Stream stream, bool disposeStream, Encoding encoding)
{
this.stream = stream;
this.disposeStream = disposeStream;
this.encoding = encoding;
exportFile = new Lazy<StreamWriter>(() => new StreamWriter(stream, encoding));
}
public void Serialize(Message message)
{
var serializableMessage = new SerializableMessage(message);
var serialized = JsonConvert.SerializeObject(serializableMessage, SerializerSettings);
exportFile.Value.WriteLine(serialized);
}
public int GetMessageCount()
{
if (!stream.CanSeek)
return 0;
var position = stream.Position;
try
{
var lineCount = 0;
using var reader = new StreamReader(stream, encoding, true, DefaultBufferSize, true);
while (!reader.EndOfStream)
{
if (!string.IsNullOrEmpty(reader.ReadLine()))
lineCount++;
}
return lineCount;
}
finally
{
stream.Position = position;
}
}
public IEnumerable<Message> Deserialize(IModel channel)
{
using var reader = new StreamReader(stream, encoding, true, DefaultBufferSize, true);
while (!reader.EndOfStream)
{
var serialized = reader.ReadLine();
if (string.IsNullOrEmpty(serialized))
continue;
var serializableMessage = JsonConvert.DeserializeObject<SerializableMessage>(serialized);
if (serializableMessage == null)
continue;
yield return serializableMessage.ToMessage(channel);
}
}
public void Dispose()
{
if (exportFile.IsValueCreated)
exportFile.Value.Dispose();
if (disposeStream)
stream.Dispose();
}
// ReSharper disable MemberCanBePrivate.Local - used for JSON serialization
// ReSharper disable NotAccessedField.Local
// ReSharper disable FieldCanBeMadeReadOnly.Local
private class SerializableMessage
{
public ulong DeliveryTag;
public bool Redelivered;
public string Exchange;
public string RoutingKey;
public string Queue;
// ReSharper disable once FieldCanBeMadeReadOnly.Local - must be settable by JSON deserialization
public SerializableMessageProperties Properties;
public JObject Body;
public byte[] RawBody;
// ReSharper disable once UnusedMember.Global - used by JSON deserialization
// ReSharper disable once UnusedMember.Local
public SerializableMessage()
{
Properties = new SerializableMessageProperties();
}
public SerializableMessage(Message fromMessage)
{
DeliveryTag = fromMessage.DeliveryTag;
Redelivered = fromMessage.Redelivered;
Exchange = fromMessage.Exchange;
RoutingKey = fromMessage.RoutingKey;
Queue = fromMessage.Queue;
Properties = new SerializableMessageProperties(fromMessage.Properties);
// If this is detected as a JSON message, include the object directly in the JSON line so that it is easier
// to read and process in the output file. Otherwise simply include the raw data and let Newtonsoft encode it.
// This does mean the message will be rewritten. If this is an issue, feel free to add a "raw" option to this tool
// that forces the RawBody to be used. It is open-source after all :-).
if (Properties.ContentType == "application/json")
{
try
{
Body = JObject.Parse(Encoding.UTF8.GetString(fromMessage.Body));
RawBody = null;
}
catch
{
// Fall back to using the raw body
Body = null;
RawBody = fromMessage.Body;
}
}
else
{
Body = null;
RawBody = fromMessage.Body;
}
}
public Message ToMessage(IModel channel)
{
return new()
{
DeliveryTag = DeliveryTag,
Redelivered = Redelivered,
Exchange = Exchange,
RoutingKey = RoutingKey,
Queue = Queue,
Properties = Properties.ToBasicProperties(channel),
Body = Body != null
? Encoding.UTF8.GetBytes(Body.ToString(Formatting.None))
: RawBody
};
}
}
// IBasicProperties is finicky when it comes to writing it's properties,
// so we need this normalized class to read and write it from and to JSON
private class SerializableMessageProperties
{
public string AppId;
public string ClusterId;
public string ContentEncoding;
public string ContentType;
public string CorrelationId;
public byte? DeliveryMode;
public string Expiration;
public IDictionary<string, string> Headers;
public string MessageId;
public byte? Priority;
public string ReplyTo;
public long? Timestamp;
public string Type;
public string UserId;
public SerializableMessageProperties()
{
}
public SerializableMessageProperties(IBasicProperties fromProperties)
{
AppId = fromProperties.AppId;
ClusterId = fromProperties.ClusterId;
ContentEncoding = fromProperties.ContentEncoding;
ContentType = fromProperties.ContentType;
CorrelationId = fromProperties.CorrelationId;
DeliveryMode = fromProperties.IsDeliveryModePresent() ? (byte?)fromProperties.DeliveryMode : null;
Expiration = fromProperties.Expiration;
MessageId = fromProperties.MessageId;
Priority = fromProperties.IsPriorityPresent() ? (byte?) fromProperties.Priority : null;
ReplyTo = fromProperties.ReplyTo;
Timestamp = fromProperties.IsTimestampPresent() ? (long?)fromProperties.Timestamp.UnixTime : null;
Type = fromProperties.Type;
UserId = fromProperties.UserId;
if (fromProperties.IsHeadersPresent())
{
Headers = new Dictionary<string, string>();
// This assumes header values are UTF-8 encoded strings. This is true for Tapeti.
foreach (var (key, value) in fromProperties.Headers)
Headers.Add(key, Encoding.UTF8.GetString((byte[])value));
}
else
Headers = null;
}
public IBasicProperties ToBasicProperties(IModel channel)
{
var properties = channel.CreateBasicProperties();
if (!string.IsNullOrEmpty(AppId)) properties.AppId = AppId;
if (!string.IsNullOrEmpty(ClusterId)) properties.ClusterId = ClusterId;
if (!string.IsNullOrEmpty(ContentEncoding)) properties.ContentEncoding = ContentEncoding;
if (!string.IsNullOrEmpty(ContentType)) properties.ContentType = ContentType;
if (!string.IsNullOrEmpty(CorrelationId)) properties.CorrelationId = CorrelationId;
if (DeliveryMode.HasValue) properties.DeliveryMode = DeliveryMode.Value;
if (!string.IsNullOrEmpty(Expiration)) properties.Expiration = Expiration;
if (!string.IsNullOrEmpty(MessageId)) properties.MessageId = MessageId;
if (Priority.HasValue) properties.Priority = Priority.Value;
if (!string.IsNullOrEmpty(ReplyTo)) properties.ReplyTo = ReplyTo;
if (Timestamp.HasValue) properties.Timestamp = new AmqpTimestamp(Timestamp.Value);
if (!string.IsNullOrEmpty(Type)) properties.Type = Type;
if (!string.IsNullOrEmpty(UserId)) properties.UserId = UserId;
// ReSharper disable once InvertIf
if (Headers != null)
{
properties.Headers = new Dictionary<string, object>();
foreach (var (key, value) in Headers)
properties.Headers.Add(key, Encoding.UTF8.GetBytes(value));
}
return properties;
}
}
// ReSharper restore FieldCanBeMadeReadOnly.Local
// ReSharper restore NotAccessedField.Local
// ReSharper restore MemberCanBePrivate.Local
}
}

24
Tapeti.Cmd/Tapeti.Cmd.csproj
View File

@ -1,24 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>netcoreapp2.1</TargetFramework>
<Version>2.0.0</Version>
<Authors>Mark van Renswoude</Authors>
<Company />
<Description></Description>
<PackageTags>rabbitmq tapeti</PackageTags>
<PackageLicenseExpression>Unlicense</PackageLicenseExpression>
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<Version>2.0.0</Version>
<Product>Tapeti Command-line Utility</Product>
<LangVersion>latest</LangVersion>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="CommandLineParser" Version="2.8.0" />
<PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
<PackageReference Include="RabbitMQ.Client" Version="6.2.1" />
</ItemGroup>
</Project>

93
Tapeti.Cmd/Verbs/BaseConnectionOptions.cs
View File

@ -1,93 +0,0 @@
using System;
using CommandLine;
using RabbitMQ.Client;
using Tapeti.Cmd.ConsoleHelper;
namespace Tapeti.Cmd.Verbs
{
public class BaseConnectionOptions
{
[Option('h', "host", HelpText = "(Default: localhost) Hostname of the RabbitMQ server. Can also be set using the TAPETI_HOST environment variable.")]
public string Host { get; set; }
[Option("port", HelpText = "(Default: 5672) AMQP port of the RabbitMQ server. Can also be set using the TAPETI_PORT environment variable.")]
public int? Port { get; set; }
[Option('v', "virtualhost", HelpText = "(Default: /) Virtual host used for the RabbitMQ connection. Can also be set using the TAPETI_VIRTUALHOST environment variable.")]
public string VirtualHost { get; set; }
[Option('u', "username", HelpText = "(Default: guest) Username used to connect to the RabbitMQ server. Can also be set using the TAPETI_USERNAME environment variable.")]
public string Username { get; set; }
[Option('p', "password", HelpText = "(Default: guest) Password used to connect to the RabbitMQ server. Can also be set using the TAPETI_PASSWORD environment variable.")]
public string Password { get; set; }
public ConnectionFactory CreateConnectionFactory(IConsole console)
{
var consoleWriter = console.GetPermanentWriter();
consoleWriter.WriteLine("Using connection parameters:");
var factory = new ConnectionFactory
{
HostName = GetOptionOrEnvironmentValue(consoleWriter, " Host : ", Host, "TAPETI_HOST", "localhost"),
Port = GetOptionOrEnvironmentValue(consoleWriter, " Port : ", Port, "TAPETI_PORT", 5672),
VirtualHost = GetOptionOrEnvironmentValue(consoleWriter, " Virtual host : ", VirtualHost, "TAPETI_VIRTUALHOST", "/"),
UserName = GetOptionOrEnvironmentValue(consoleWriter, " Username : ", Username, "TAPETI_USERNAME", "guest"),
Password = GetOptionOrEnvironmentValue(consoleWriter, " Password : ", Password, "TAPETI_PASSWORD", "guest", true)
};
consoleWriter.WriteLine("");
return factory;
}
private static string GetOptionOrEnvironmentValue(IConsoleWriter consoleWriter, string consoleDisplayName, string optionValue, string environmentName, string defaultValue, bool hideValue = false)
{
string GetDisplayValue(string value)
{
return hideValue
? "<hidden>"
: value;
}
if (!string.IsNullOrEmpty(optionValue))
{
consoleWriter.WriteLine($"{consoleDisplayName}{GetDisplayValue(optionValue)} (from command-line)");
return optionValue;
}
var environmentValue = Environment.GetEnvironmentVariable(environmentName);
if (!string.IsNullOrEmpty(environmentValue))
{
consoleWriter.WriteLine($"{consoleDisplayName}{GetDisplayValue(environmentValue)} (from environment variable)");
return environmentValue;
}
consoleWriter.WriteLine($"{consoleDisplayName}{GetDisplayValue(defaultValue)} (default)");
return defaultValue;
}
private static int GetOptionOrEnvironmentValue(IConsoleWriter consoleWriter, string consoleDisplayName, int? optionValue, string environmentName, int defaultValue)
{
if (optionValue.HasValue)
{
consoleWriter.WriteLine($"{consoleDisplayName}{optionValue} (from command-line)");
return optionValue.Value;
}
var environmentValue = Environment.GetEnvironmentVariable(environmentName);
if (!string.IsNullOrEmpty(environmentValue) && int.TryParse(environmentValue, out var environmentIntValue))
{
consoleWriter.WriteLine($"{consoleDisplayName}{environmentIntValue} (from environment variable)");
return environmentIntValue;
}
consoleWriter.WriteLine($"{consoleDisplayName}{defaultValue} (default)");
return defaultValue;
}
}
}

17
Tapeti.Cmd/Verbs/BaseMessageSerializerOptions.cs
View File

@ -1,17 +0,0 @@
using CommandLine;
namespace Tapeti.Cmd.Verbs
{
public enum SerializationMethod
{
SingleFileJSON,
EasyNetQHosepipe
}
public class BaseMessageSerializerOptions : BaseConnectionOptions
{
[Option('s', "serialization", HelpText = "The method used to serialize the message for import or export. Valid options: SingleFileJSON, EasyNetQHosepipe.", Default = SerializationMethod.SingleFileJSON)]
public SerializationMethod SerializationMethod { get; set; }
}
}

48
Tapeti.Cmd/Verbs/BindQueueVerb.cs
View File

@ -1,48 +0,0 @@
using System;
using System.Collections.Generic;
using CommandLine;
using RabbitMQ.Client;
using Tapeti.Cmd.ConsoleHelper;
using Tapeti.Cmd.Parser;
namespace Tapeti.Cmd.Verbs
{
[Verb("bindqueue", HelpText = "Add a binding to an existing queue.")]
[ExecutableVerb(typeof(BindQueueVerb))]
public class BindQueueOptions : BaseConnectionOptions
{
[Option('q', "queue", Required = true, HelpText = "The name of the queue to add the binding(s) to.")]
public string QueueName { get; set; }
[Option('b', "bindings", Required = false, HelpText = "One or more bindings to add to the queue. Format: <exchange>:<routingKey>")]
public IEnumerable<string> Bindings { get; set; }
}
public class BindQueueVerb : IVerbExecuter
{
private readonly BindQueueOptions options;
public BindQueueVerb(BindQueueOptions options)
{
this.options = options;
}
public void Execute(IConsole console)
{
var consoleWriter = console.GetPermanentWriter();
var bindings = BindingParser.Parse(options.Bindings);
var factory = options.CreateConnectionFactory(console);
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
foreach (var (exchange, routingKey) in bindings)
channel.QueueBind(options.QueueName, exchange, routingKey);
consoleWriter.WriteLine($"{bindings.Length} binding{(bindings.Length != 1 ? "s" : "")} added to queue {options.QueueName}.");
}
}
}

52
Tapeti.Cmd/Verbs/DeclareQueueVerb.cs
View File

@ -1,52 +0,0 @@
using System;
using System.Collections.Generic;
using CommandLine;
using RabbitMQ.Client;
using Tapeti.Cmd.ConsoleHelper;
using Tapeti.Cmd.Parser;
namespace Tapeti.Cmd.Verbs
{
[Verb("declarequeue", HelpText = "Declares a durable queue without arguments.")]
[ExecutableVerb(typeof(DeclareQueueVerb))]
public class DeclareQueueOptions : BaseConnectionOptions
{
[Option('q', "queue", Required = true, HelpText = "The name of the queue to declare.")]
public string QueueName { get; set; }
[Option('b', "bindings", Required = false, HelpText = "One or more bindings to add to the queue. Format: <exchange>:<routingKey>")]
public IEnumerable<string> Bindings { get; set; }
}
public class DeclareQueueVerb : IVerbExecuter
{
private readonly DeclareQueueOptions options;
public DeclareQueueVerb(DeclareQueueOptions options)
{
this.options = options;
}
public void Execute(IConsole console)
{
var consoleWriter = console.GetPermanentWriter();
// Parse early to fail early
var bindings = BindingParser.Parse(options.Bindings);
var factory = options.CreateConnectionFactory(console);
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
channel.QueueDeclare(options.QueueName, true, false, false);
foreach (var (exchange, routingKey) in bindings)
channel.QueueBind(options.QueueName, exchange, routingKey);
consoleWriter.WriteLine($"Queue {options.QueueName} declared with {bindings.Length} binding{(bindings.Length != 1 ? "s" : "")}.");
}
}
}

54
Tapeti.Cmd/Verbs/ExampleVerb.cs
View File

@ -1,54 +0,0 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using CommandLine;
using RabbitMQ.Client;
using Tapeti.Cmd.ConsoleHelper;
using Tapeti.Cmd.Mock;
using Tapeti.Cmd.Serialization;
namespace Tapeti.Cmd.Verbs
{
[Verb("example", HelpText = "Output an example SingleFileJSON formatted message.")]
[ExecutableVerb(typeof(ExampleVerb))]
public class ExampleOptions
{
}
public class ExampleVerb : IVerbExecuter
{
public ExampleVerb(ExampleOptions options)
{
// Prevent compiler warnings, the parameter is expected by the Activator
Debug.Assert(options != null);
}
public void Execute(IConsole console)
{
using var messageSerializer = new SingleFileJSONMessageSerializer(Console.OpenStandardOutput(), false, new UTF8Encoding(false));
messageSerializer.Serialize(new Message
{
Exchange = "example",
Queue = "example.queue",
RoutingKey = "example.routing.key",
DeliveryTag = 42,
Properties = new MockBasicProperties
{
ContentType = "application/json",
DeliveryMode = 2,
Headers = new Dictionary<string, object>
{
{ "classType", Encoding.UTF8.GetBytes("Tapeti.Cmd.Example:Tapeti.Cmd") }
},
ReplyTo = "reply.queue",
Timestamp = new AmqpTimestamp(new DateTimeOffset(DateTime.UtcNow).ToUnixTimeSeconds())
},
Body = Encoding.UTF8.GetBytes("{ \"Hello\": \"world!\" }")
});
}
}
}

31
Tapeti.Cmd/Verbs/ExecutableVerbAttribute.cs
View File

@ -1,31 +0,0 @@
using System;
using Tapeti.Cmd.ConsoleHelper;
namespace Tapeti.Cmd.Verbs
{
/// <remarks>
/// Implementations are expected to have a constructor which accepts the options class
/// associated with the ExecutableVerb attribute.
/// </remarks>
public interface IVerbExecuter
{
void Execute(IConsole console);
}
[AttributeUsage(AttributeTargets.Class)]
public class ExecutableVerbAttribute : Attribute
{
public Type VerbExecuter { get; }
public ExecutableVerbAttribute(Type verbExecuter)
{
if (!typeof(IVerbExecuter).IsAssignableFrom(verbExecuter))
throw new InvalidCastException("Type must support IVerbExecuter");
VerbExecuter = verbExecuter;
}
}
}

136
Tapeti.Cmd/Verbs/ExportVerb.cs
View File

@ -1,136 +0,0 @@
using System;
using System.IO;
using System.Text;
using CommandLine;
using Tapeti.Cmd.ConsoleHelper;
using Tapeti.Cmd.Serialization;
namespace Tapeti.Cmd.Verbs
{
[Verb("export", HelpText = "Fetch messages from a queue and write it to disk.")]
[ExecutableVerb(typeof(ExportVerb))]
public class ExportOptions : BaseMessageSerializerOptions
{
[Option('q', "queue", Required = true, HelpText = "The queue to read the messages from.")]
public string QueueName { get; set; }
[Option('o', "output", Required = true, HelpText = "Path or filename (depending on the chosen serialization method) where the messages will be output to.")]
public string OutputPath { get; set; }
[Option('y', "overwrite", HelpText = "If the output exists, do not ask to overwrite.")]
public bool Overwrite { get; set; }
[Option('r', "remove", HelpText = "If specified messages are acknowledged and removed from the queue. If not messages are kept.")]
public bool RemoveMessages { get; set; }
[Option("skip", HelpText = "(Default: 0) Number of messages in the queue to skip. Useful if a previous non-removing export was interrupted.", Default = 0)]
public int Skip { get; set; }
[Option('n', "maxcount", HelpText = "(Default: all) Maximum number of messages to retrieve from the queue.")]
public int? MaxCount { get; set; }
}
public class ExportVerb : IVerbExecuter
{
private readonly ExportOptions options;
public ExportVerb(ExportOptions options)
{
this.options = options;
}
public void Execute(IConsole console)
{
var consoleWriter = console.GetPermanentWriter();
using var messageSerializer = GetMessageSerializer(options, consoleWriter);
if (messageSerializer == null)
return;
var factory = options.CreateConnectionFactory(console);
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
var totalCount = (int)channel.MessageCount(options.QueueName);
var skip = Math.Max(options.Skip, 0);
if (skip > 0)
totalCount -= Math.Min(skip, totalCount);
if (options.MaxCount.HasValue && options.MaxCount.Value < totalCount)
totalCount = options.MaxCount.Value;
consoleWriter.WriteLine($"Exporting {totalCount} message{(totalCount != 1 ? "s" : "")} (actual number may differ if queue has active consumers or publishers)");
var messageCount = 0;
using (var progressBar = new ProgressBar(console, totalCount))
{
while (!console.Cancelled && (!options.MaxCount.HasValue || messageCount < options.MaxCount.Value))
{
var result = channel.BasicGet(options.QueueName, false);
if (result == null)
// No more messages on the queue
break;
if (skip > 0)
skip--;
else
{
messageCount++;
messageSerializer.Serialize(new Message
{
DeliveryTag = result.DeliveryTag,
Redelivered = result.Redelivered,
Exchange = result.Exchange,
RoutingKey = result.RoutingKey,
Queue = options.QueueName,
Properties = result.BasicProperties,
Body = result.Body.ToArray()
});
if (options.RemoveMessages)
channel.BasicAck(result.DeliveryTag, false);
progressBar.Report(messageCount);
}
}
}
consoleWriter.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} exported.");
}
private static IMessageSerializer GetMessageSerializer(ExportOptions options, IConsoleWriter consoleWriter)
{
switch (options.SerializationMethod)
{
case SerializationMethod.SingleFileJSON:
// ReSharper disable once InvertIf - causes two lines of "new SingleFileJSONMessageSerializer". DRY ReSharper.
if (!options.Overwrite && File.Exists(options.OutputPath))
{
if (!consoleWriter.ConfirmYesNo($"The output file '{options.OutputPath}' already exists, do you want to overwrite it?"))
return null;
}
return new SingleFileJSONMessageSerializer(new FileStream(options.OutputPath, FileMode.Create, FileAccess.Write, FileShare.Read), true, Encoding.UTF8);
case SerializationMethod.EasyNetQHosepipe:
// ReSharper disable once InvertIf - causes two lines of "new SingleFileJSONMessageSerializer". DRY ReSharper.
if (!options.Overwrite && EasyNetQMessageSerializer.OutputExists(options.OutputPath))
{
if (!consoleWriter.ConfirmYesNo($"The output path '{options.OutputPath}' already contains a previous export, do you want to overwrite it?"))
return null;
}
return new EasyNetQMessageSerializer(options.OutputPath);
default:
throw new ArgumentOutOfRangeException(nameof(options.SerializationMethod), options.SerializationMethod, "Invalid SerializationMethod");
}
}
}
}

153
Tapeti.Cmd/Verbs/ImportVerb.cs
View File

@ -1,153 +0,0 @@
using System;
using System.IO;
using System.Net;
using System.Text;
using CommandLine;
using RabbitMQ.Client;
using Tapeti.Cmd.ConsoleHelper;
using Tapeti.Cmd.RateLimiter;
using Tapeti.Cmd.Serialization;
namespace Tapeti.Cmd.Verbs
{
[Verb("import", HelpText = "Read messages from disk as previously exported and publish them to a queue.")]
[ExecutableVerb(typeof(ImportVerb))]
public class ImportOptions : BaseMessageSerializerOptions
{
[Option('i', "input", Group = "Input", HelpText = "Path or filename (depending on the chosen serialization method) where the messages will be read from.")]
public string InputFile { get; set; }
[Option('m', "message", Group = "Input", HelpText = "Single message to be sent, in the same format as used for SingleFileJSON. Serialization argument has no effect when using this input.")]
public string InputMessage { get; set; }
[Option('c', "pipe", Group = "Input", HelpText = "Messages are read from the standard input pipe, in the same format as used for SingleFileJSON. Serialization argument has no effect when using this input.")]
public bool InputPipe { get; set; }
[Option("urlencoded", HelpText = "Indicates the message is URL encoded. Only applies to messages passed directly with --message as quotes are very quirky on the command-line, even more so in PowerShell.")]
public bool UrlEncoded { get; set; }
[Option('e', "exchange", HelpText = "If specified publishes to the originating exchange using the original routing key. By default these are ignored and the message is published directly to the originating queue.")]
public bool PublishToExchange { get; set; }
[Option("skip", HelpText = "(Default: 0) Number of messages in the input to skip. Useful if a previous import was interrupted.", Default = 0)]
public int Skip { get; set; }
[Option('n', "maxcount", HelpText = "(Default: all) Maximum number of messages to import.")]
public int? MaxCount { get; set; }
[Option("maxrate", HelpText = "The maximum amount of messages per second to import.")]
public int? MaxRate { get; set; }
[Option("batchsize", HelpText = "How many messages to import before pausing. Will wait for manual confirmation unless batchpausetime is specified.")]
public int? BatchSize { get; set; }
[Option("batchpausetime", HelpText = "How many seconds to wait before starting the next batch if batchsize is specified.")]
public int? BatchPauseTime { get; set; }
}
public class ImportVerb : IVerbExecuter
{
private readonly ImportOptions options;
public ImportVerb(ImportOptions options)
{
this.options = options;
}
public void Execute(IConsole console)
{
var consoleWriter = console.GetPermanentWriter();
var factory = options.CreateConnectionFactory(console); using var messageSerializer = GetMessageSerializer(options);
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
var rateLimiter = RateLimiterFactory.Create(console, options.MaxRate, options.BatchSize, options.BatchPauseTime);
var totalCount = messageSerializer.GetMessageCount();
var messageCount = 0;
var skip = Math.Max(options.Skip, 0);
ProgressBar progress = null;
if (totalCount > 0)
progress = new ProgressBar(console, totalCount);
try
{
foreach (var message in messageSerializer.Deserialize(channel))
{
if (console.Cancelled || (options.MaxCount.HasValue && messageCount >= options.MaxCount.Value))
break;
if (skip > 0)
skip--;
else
rateLimiter.Execute(() =>
{
if (console.Cancelled)
return;
var exchange = options.PublishToExchange ? message.Exchange : "";
var routingKey = options.PublishToExchange ? message.RoutingKey : message.Queue;
// ReSharper disable AccessToDisposedClosure
channel.BasicPublish(exchange, routingKey, message.Properties, message.Body);
messageCount++;
progress?.Report(messageCount);
// ReSharper restore AccessToDisposedClosure
});
}
}
finally
{
progress?.Dispose();
}
consoleWriter.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} published.");
}
private static IMessageSerializer GetMessageSerializer(ImportOptions options)
{
switch (options.SerializationMethod)
{
case SerializationMethod.SingleFileJSON:
return new SingleFileJSONMessageSerializer(GetInputStream(options, out var disposeStream), disposeStream, Encoding.UTF8);
case SerializationMethod.EasyNetQHosepipe:
if (string.IsNullOrEmpty(options.InputFile))
throw new ArgumentException("An input path must be provided when using EasyNetQHosepipe serialization");
return new EasyNetQMessageSerializer(options.InputFile);
default:
throw new ArgumentOutOfRangeException(nameof(options.SerializationMethod), options.SerializationMethod, "Invalid SerializationMethod");
}
}
private static Stream GetInputStream(ImportOptions options, out bool disposeStream)
{
if (options.InputPipe)
{
disposeStream = false;
return Console.OpenStandardInput();
}
if (!string.IsNullOrEmpty(options.InputMessage))
{
var inputMessage = options.UrlEncoded
? WebUtility.UrlDecode(options.InputMessage)
: options.InputMessage;
disposeStream = true;
return new MemoryStream(Encoding.UTF8.GetBytes(inputMessage));
}
disposeStream = true;
return new FileStream(options.InputFile, FileMode.Open, FileAccess.Read, FileShare.ReadWrite);
}
}
}

48
Tapeti.Cmd/Verbs/PurgeVerb.cs
View File

@ -1,48 +0,0 @@
using CommandLine;
using RabbitMQ.Client;
using Tapeti.Cmd.ConsoleHelper;
namespace Tapeti.Cmd.Verbs
{
[Verb("purge", HelpText = "Removes all messages from a queue destructively.")]
[ExecutableVerb(typeof(PurgeVerb))]
public class PurgeOptions : BaseConnectionOptions
{
[Option('q', "queue", Required = true, HelpText = "The queue to purge.")]
public string QueueName { get; set; }
[Option("confirm", HelpText = "Confirms the purging of the specified queue. If not provided, an interactive prompt will ask for confirmation.", Default = false)]
public bool Confirm { get; set; }
}
public class PurgeVerb : IVerbExecuter
{
private readonly PurgeOptions options;
public PurgeVerb(PurgeOptions options)
{
this.options = options;
}
public void Execute(IConsole console)
{
var consoleWriter = console.GetPermanentWriter();
if (!options.Confirm)
{
if (!consoleWriter.ConfirmYesNo($"Do you want to purge the queue '{options.QueueName}'?"))
return;
}
var factory = options.CreateConnectionFactory(console); using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
var messageCount = channel.QueuePurge(options.QueueName);
consoleWriter.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} purged from '{options.QueueName}'.");
}
}
}

78
Tapeti.Cmd/Verbs/RemoveQueueVerb.cs
View File

@ -1,78 +0,0 @@
using CommandLine;
using RabbitMQ.Client;
using RabbitMQ.Client.Exceptions;
using Tapeti.Cmd.ConsoleHelper;
namespace Tapeti.Cmd.Verbs
{
[Verb("removequeue", HelpText = "Removes a durable queue.")]
[ExecutableVerb(typeof(RemoveQueueVerb))]
public class RemoveQueueOptions : BaseConnectionOptions
{
[Option('q', "queue", Required = true, HelpText = "The name of the queue to remove.")]
public string QueueName { get; set; }
[Option("confirm", HelpText = "Confirms the removal of the specified queue. If not provided, an interactive prompt will ask for confirmation.", Default = false)]
public bool Confirm { get; set; }
[Option("confirmpurge", HelpText = "Confirms the removal of the specified queue even if there still are messages in the queue. If not provided, an interactive prompt will ask for confirmation.", Default = false)]
public bool ConfirmPurge { get; set; }
}
public class RemoveQueueVerb : IVerbExecuter
{
private readonly RemoveQueueOptions options;
public RemoveQueueVerb(RemoveQueueOptions options)
{
this.options = options;
}
public void Execute(IConsole console)
{
var consoleWriter = console.GetPermanentWriter();
if (!options.Confirm)
{
if (!consoleWriter.ConfirmYesNo($"Do you want to remove the queue '{options.QueueName}'?"))
return;
}
var factory = options.CreateConnectionFactory(console);
uint messageCount;
try
{
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
messageCount = channel.QueueDelete(options.QueueName, true, true);
}
catch (OperationInterruptedException e)
{
if (e.ShutdownReason.ReplyCode == 406)
{
if (!options.ConfirmPurge)
{
if (!consoleWriter.ConfirmYesNo($"There are messages remaining. Do you want to purge the queue '{options.QueueName}'?"))
return;
}
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
messageCount = channel.QueueDelete(options.QueueName, true, false);
}
else
throw;
}
consoleWriter.WriteLine(messageCount == 0
? $"Empty or non-existent queue '{options.QueueName}' removed."
: $"{messageCount} message{(messageCount != 1 ? "s" : "")} purged while removing '{options.QueueName}'.");
}
}
}

191
Tapeti.Cmd/Verbs/ShovelVerb.cs
View File

@ -1,191 +0,0 @@
using System;
using CommandLine;
using RabbitMQ.Client;
using Tapeti.Cmd.ConsoleHelper;
using Tapeti.Cmd.RateLimiter;
namespace Tapeti.Cmd.Verbs
{
[Verb("shovel", HelpText = "Reads messages from a queue and publishes them to another queue, optionally to another RabbitMQ server.")]
[ExecutableVerb(typeof(ShovelVerb))]
public class ShovelOptions : BaseConnectionOptions
{
[Option('q', "queue", Required = true, HelpText = "The queue to read the messages from.")]
public string QueueName { get; set; }
[Option('t', "targetqueue", HelpText = "The target queue to publish the messages to. Defaults to the source queue if a different target host, port or virtualhost is specified. Otherwise it must be different from the source queue.")]
public string TargetQueueName { get; set; }
[Option('r', "remove", HelpText = "If specified messages are acknowledged and removed from the source queue. If not messages are kept.")]
public bool RemoveMessages { get; set; }
[Option('n', "maxcount", HelpText = "(Default: all) Maximum number of messages to retrieve from the queue.")]
public int? MaxCount { get; set; }
[Option("targethost", HelpText = "Hostname of the target RabbitMQ server. Defaults to the source host. Note that you may still specify a different targetusername for example.")]
public string TargetHost { get; set; }
[Option("targetport", HelpText = "AMQP port of the target RabbitMQ server. Defaults to the source port.")]
public int? TargetPort { get; set; }
[Option("targetvirtualhost", HelpText = "Virtual host used for the target RabbitMQ connection. Defaults to the source virtualhost.")]
public string TargetVirtualHost { get; set; }
[Option("targetusername", HelpText = "Username used to connect to the target RabbitMQ server. Defaults to the source username.")]
public string TargetUsername { get; set; }
[Option("targetpassword", HelpText = "Password used to connect to the target RabbitMQ server. Defaults to the source password.")]
public string TargetPassword { get; set; }
[Option("skip", HelpText = "(Default: 0) Number of messages in the queue to skip. Useful if a previous non-removing shovel was interrupted.", Default = 0)]
public int Skip { get; set; }
[Option("maxrate", HelpText = "The maximum amount of messages per second to shovel.")]
public int? MaxRate { get; set; }
[Option("batchsize", HelpText = "How many messages to shovel before pausing. Will wait for manual confirmation unless batchpausetime is specified.")]
public int? BatchSize { get; set; }
[Option("batchpausetime", HelpText = "How many seconds to wait before starting the next batch if batchsize is specified.")]
public int? BatchPauseTime { get; set; }
}
public class ShovelVerb : IVerbExecuter
{
private readonly ShovelOptions options;
public ShovelVerb(ShovelOptions options)
{
this.options = options;
}
public void Execute(IConsole console)
{
var sourceFactory = options.CreateConnectionFactory(console);
using var sourceConnection = sourceFactory.CreateConnection();
using var sourceChannel = sourceConnection.CreateModel();
if (RequiresSecondConnection(options))
{
var targetFactory = new ConnectionFactory
{
HostName = !string.IsNullOrEmpty(options.TargetHost) ? options.TargetHost : options.Host,
Port = options.TargetPort ?? options.Port ?? 5672,
VirtualHost = !string.IsNullOrEmpty(options.TargetVirtualHost) ? options.TargetVirtualHost : options.VirtualHost,
UserName = !string.IsNullOrEmpty(options.TargetUsername) ? options.TargetUsername : options.Username,
Password = !string.IsNullOrEmpty(options.TargetPassword) ? options.TargetPassword : options.Password
};
using var targetConnection = targetFactory.CreateConnection();
using var targetChannel = targetConnection.CreateModel();
Shovel(console, options, sourceChannel, targetChannel);
}
else
Shovel(console, options, sourceChannel, sourceChannel);
}
private static void Shovel(IConsole console, ShovelOptions options, IModel sourceChannel, IModel targetChannel)
{
var consoleWriter = console.GetPermanentWriter();
var rateLimiter = RateLimiterFactory.Create(console, options.MaxRate, options.BatchSize, options.BatchPauseTime);
var targetQueueName = !string.IsNullOrEmpty(options.TargetQueueName) ? options.TargetQueueName : options.QueueName;
var totalCount = (int)sourceChannel.MessageCount(options.QueueName);
var skip = Math.Max(options.Skip, 0);
if (skip > 0)
totalCount -= Math.Min(skip, totalCount);
if (options.MaxCount.HasValue && options.MaxCount.Value < totalCount)
totalCount = options.MaxCount.Value;
consoleWriter.WriteLine($"Shoveling {totalCount} message{(totalCount != 1 ? "s" : "")} (actual number may differ if queue has active consumers or publishers)");
var messageCount = 0;
using (var progressBar = new ProgressBar(console, totalCount))
{
// Perform the skips outside of the rate limiter
while (skip > 0 && !console.Cancelled)
{
var result = sourceChannel.BasicGet(options.QueueName, false);
if (result == null)
// No more messages on the queue
return;
skip--;
}
var hasMessage = true;
while (!console.Cancelled && hasMessage && (!options.MaxCount.HasValue || messageCount < options.MaxCount.Value))
{
rateLimiter.Execute(() =>
{
if (console.Cancelled)
return;
var result = sourceChannel.BasicGet(options.QueueName, false);
if (result == null)
{
// No more messages on the queue
hasMessage = false;
return;
}
// Since RabbitMQ client 6 we need to copy the body before calling another channel method
// like BasicPublish, or the published body will be corrupted if sourceChannel and targetChannel are the same
var bodyCopy = result.Body.ToArray();
targetChannel.BasicPublish("", targetQueueName, result.BasicProperties, bodyCopy);
messageCount++;
if (options.RemoveMessages)
sourceChannel.BasicAck(result.DeliveryTag, false);
// ReSharper disable once AccessToDisposedClosure
progressBar.Report(messageCount);
});
}
}
consoleWriter.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} shoveled.");
}
private static bool RequiresSecondConnection(ShovelOptions options)
{
if (!string.IsNullOrEmpty(options.TargetHost) && options.TargetHost != options.Host)
return true;
if (options.TargetPort.HasValue && options.TargetPort.Value != options.Port)
return true;
if (!string.IsNullOrEmpty(options.TargetVirtualHost) && options.TargetVirtualHost != options.VirtualHost)
return true;
// All relevant target host parameters are either omitted or the same. This means the queue must be different
// to prevent an infinite loop.
if (string.IsNullOrEmpty(options.TargetQueueName) || options.TargetQueueName == options.QueueName)
throw new ArgumentException("Target queue must be different from the source queue when shoveling within the same (virtual) host");
if (!string.IsNullOrEmpty(options.TargetUsername) && options.TargetUsername != options.Username)
return true;
// ReSharper disable once ConvertIfStatementToReturnStatement
if (!string.IsNullOrEmpty(options.TargetPassword) && options.TargetPassword != options.Password)
return true;
// Everything's the same, we can use the same channel
return false;
}
}
}

48
Tapeti.Cmd/Verbs/UnbindQueueVerb.cs
View File

@ -1,48 +0,0 @@
using System;
using System.Collections.Generic;
using CommandLine;
using RabbitMQ.Client;
using Tapeti.Cmd.ConsoleHelper;
using Tapeti.Cmd.Parser;
namespace Tapeti.Cmd.Verbs
{
[Verb("unbindqueue", HelpText = "Remove a binding from a queue.")]
[ExecutableVerb(typeof(UnbindQueueVerb))]
public class UnbindQueueOptions : BaseConnectionOptions
{
[Option('q', "queue", Required = true, HelpText = "The name of the queue to remove the binding(s) from.")]
public string QueueName { get; set; }
[Option('b', "bindings", Required = false, HelpText = "One or more bindings to remove from the queue. Format: <exchange>:<routingKey>")]
public IEnumerable<string> Bindings { get; set; }
}
public class UnbindQueueVerb : IVerbExecuter
{
private readonly UnbindQueueOptions options;
public UnbindQueueVerb(UnbindQueueOptions options)
{
this.options = options;
}
public void Execute(IConsole console)
{
var consoleWriter = console.GetPermanentWriter();
var bindings = BindingParser.Parse(options.Bindings);
var factory = options.CreateConnectionFactory(console);
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
foreach (var (exchange, routingKey) in bindings)
channel.QueueUnbind(options.QueueName, exchange, routingKey);
consoleWriter.WriteLine($"{bindings.Length} binding{(bindings.Length != 1 ? "s" : "")} removed from queue {options.QueueName}.");
}
}
}

10
Tapeti.Cmd/build-release.bat
View File

@ -1,10 +0,0 @@
@Echo Off
mkdir publish
REM Executable is generated using self-contained=true, which is just a wrapper for "dotnet Tapeti.Cmd.dll".
REM We don't need all the other DLL's so we'll build it twice and borrow the wrapper executable for a proper
REM framework-dependant build.
dotnet publish -c Release -r win-x64 --self-contained=true -o .\publish\selfcontained
dotnet publish -c Release -r win-x64 --self-contained=false -o .\publish
copy .\publish\selfcontained\Tapeti.Cmd.exe .\publish\

38
Tapeti.DataAnnotations.Extensions/RequiredGuidAttribute.cs
View File

@ -1,38 +0,0 @@
using System;
using System.ComponentModel.DataAnnotations;
// ReSharper disable UnusedMember.Global
namespace Tapeti.DataAnnotations.Extensions
{
/// <inheritdoc />
/// <summary>
/// Can be used on Guid fields which are supposed to be Required, as the Required attribute does
/// not work for Guids and making them Nullable is counter-intuitive.
/// </summary>
public class RequiredGuidAttribute : ValidationAttribute
{
private const string DefaultErrorMessage = "'{0}' does not contain a valid guid";
private const string InvalidTypeErrorMessage = "'{0}' is not of type Guid";
/// <inheritdoc />
public RequiredGuidAttribute() : base(DefaultErrorMessage)
{
}
/// <inheritdoc />
protected override ValidationResult IsValid(object value, ValidationContext validationContext)
{
if (value == null)
return new ValidationResult(FormatErrorMessage(validationContext.DisplayName));
if (value.GetType() != typeof(Guid))
return new ValidationResult(string.Format(InvalidTypeErrorMessage, validationContext.DisplayName));
var guid = (Guid)value;
return guid == Guid.Empty
? new ValidationResult(FormatErrorMessage(validationContext.DisplayName))
: null;
}
}
}

34
Tapeti.DataAnnotations.Extensions/Tapeti.DataAnnotations.Extensions.csproj
View File

@ -1,34 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
<Authors>Mark van Renswoude</Authors>
<Company />
<Description>Additional DataAnnotations attributes. Not specific to Tapeti, but useful for annotating message classes.</Description>
<PackageTags>rabbitmq tapeti dataannotations</PackageTags>
<PackageLicenseExpression>Unlicense</PackageLicenseExpression>
<PackageProjectUrl>https://github.com/MvRens/Tapeti</PackageProjectUrl>
<PackageIcon>Tapeti.DataAnnotations.png</PackageIcon>
<Version>2.0.0</Version>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'">
<NoWarn>1701;1702</NoWarn>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="System.ComponentModel.Annotations" Version="5.0.0" />
</ItemGroup>
<ItemGroup>
<None Include="..\resources\icons\Tapeti.DataAnnotations.png">
<Pack>True</Pack>
<PackagePath></PackagePath>
</None>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
</ItemGroup>
</Project>

4
Tapeti.Flow/Tapeti.Flow.csproj
View File

@ -18,7 +18,6 @@
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\Tapeti.Annotations\Tapeti.Annotations.csproj" />
<ProjectReference Include="..\Tapeti\Tapeti.csproj" />
</ItemGroup>
@ -30,6 +29,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
<PackageReference Include="Tapeti.Annotations" Version="3.0.0" />
</ItemGroup>
</Project>

28
Tapeti.Serilog/ConfigExtensions.cs Normal file
View File

@ -0,0 +1,28 @@
using Serilog.Events;
using Tapeti.Config;
using Tapeti.Serilog.Middleware;
// ReSharper disable UnusedMember.Global - public API
namespace Tapeti.Serilog
{
/// <summary>
/// ITapetiConfigBuilder extension for enabling message handler logging.
/// </summary>
public static class ConfigExtensions
{
/// <summary>
/// Enables message handler logging.
/// </summary>
/// <param name="config"></param>
/// <param name="elapsedWarningTreshold">The time (in milliseconds) a message handler is allowed to run without a warning being logged</param>
/// <param name="defaultLevel">The default log level when a message handler completes within the elapsedWarningTreshold</param>
/// <returns></returns>
public static ITapetiConfigBuilder WithMessageHandlerLogging(this ITapetiConfigBuilder config,
double elapsedWarningTreshold = 500, LogEventLevel defaultLevel = LogEventLevel.Debug)
{
config.Use(new MessageHandlerLoggingBindingMiddleware(elapsedWarningTreshold, defaultLevel));
return config;
}
}
}

62
Tapeti.Serilog/Default/DiagnosticContext.cs Normal file
View File

@ -0,0 +1,62 @@
using System.Collections.Generic;
using Serilog.Core;
using Serilog.Events;
namespace Tapeti.Serilog.Default
{
/// <summary>
/// Implements the IDiagnosticContext interface for a Serilog ILogger.
/// </summary>
public class DiagnosticContext : IDiagnosticContext
{
private readonly global::Serilog.ILogger logger;
private readonly List<LogEventProperty> properties = new List<LogEventProperty>();
/// <summary>
/// Creates a new instance of a DiagnosticContext
/// </summary>
/// <param name="logger">The Serilog ILogger which will be enriched</param>
public DiagnosticContext(global::Serilog.ILogger logger)
{
this.logger = logger;
}
/// <inheritdoc />
public void Set(string propertyName, object value, bool destructureObjects = false)
{
if (logger.BindProperty(propertyName, value, destructureObjects, out var logEventProperty))
properties.Add(logEventProperty);
}
/// <summary>
/// Returns a Serilog ILogger which is enriched with the properties set by this DiagnosticContext
/// </summary>
public global::Serilog.ILogger GetEnrichedLogger()
{
return properties.Count > 0
? logger.ForContext(new LogEventPropertiesEnricher(properties))
: logger;
}
private class LogEventPropertiesEnricher : ILogEventEnricher
{
private readonly IEnumerable<LogEventProperty> properties;
public LogEventPropertiesEnricher(IEnumerable<LogEventProperty> properties)
{
this.properties = properties;
}
public void Enrich(LogEvent logEvent, ILogEventPropertyFactory propertyFactory)
{
foreach (var property in properties)
logEvent.AddOrUpdateProperty(property);
}
}
}
}

23
Tapeti.Serilog/DiagnosticContextPayload.cs Normal file
View File

@ -0,0 +1,23 @@
using Tapeti.Config;
namespace Tapeti.Serilog
{
/// <summary>
/// Stores the IDiagnosticContext for a message handler.
/// </summary>
public class DiagnosticContextPayload : IMessageContextPayload
{
/// <summary>
/// Initializes a DiagnosticContext payload.
/// </summary>
public DiagnosticContextPayload(IDiagnosticContext diagnosticContext)
{
DiagnosticContext = diagnosticContext;
}
/// <summary>
/// The diagnostic context for the current message handler.
/// </summary>
public IDiagnosticContext DiagnosticContext { get; }
}
}

23
Tapeti.Serilog/IDiagnosticContext.cs Normal file
View File

@ -0,0 +1,23 @@
namespace Tapeti.Serilog
{
/// <summary>
/// Collects diagnostic information for message handler logging when using the
/// MessageHandlerLogging middleware.
/// </summary>
/// <remarks>
/// This is a one-to-one copy of the IDiagnosticContext in Serilog.Extensions.Hosting which
/// saves a reference to that package while allowing similar usage within Tapeti message handlers.
/// </remarks>
public interface IDiagnosticContext
{
/// <summary>
/// Set the specified property on the current diagnostic context. The property will be collected
/// and attached to the event emitted at the completion of the context.
/// </summary>
/// <param name="propertyName">The name of the property. Must be non-empty.</param>
/// <param name="value">The property value.</param>
/// <param name="destructureObjects">If true, the value will be serialized as structured
/// data if possible; if false, the object will be recorded as a scalar or simple array.</param>
void Set(string propertyName, object value, bool destructureObjects = false);
}
}

53
Tapeti.Serilog/Middleware/MessageHandlerLoggingBindingMiddleware.cs Normal file
View File

@ -0,0 +1,53 @@
using System;
using System.Linq;
using Serilog.Events;
using Tapeti.Config;
namespace Tapeti.Serilog.Middleware
{
/// <summary>
/// Implements the middleware which binds any IDiagnosticContext parameter in message handlers.
/// </summary>
public class MessageHandlerLoggingBindingMiddleware : IControllerBindingMiddleware
{
private readonly IControllerMessageMiddleware controllerMessageMiddleware;
/// <summary>
/// Creates a new instance of the middleware which binds any IDiagnosticContext parameter in message handlers.
/// </summary>
/// <param name="elapsedWarningTreshold"></param>
/// <param name="defaultLevel"></param>
public MessageHandlerLoggingBindingMiddleware(double elapsedWarningTreshold = 500, LogEventLevel defaultLevel = LogEventLevel.Debug)
{
controllerMessageMiddleware = new MessageHandlerLoggingMessageMiddleware(elapsedWarningTreshold, defaultLevel);
}
/// <inheritdoc />
public void Handle(IControllerBindingContext context, Action next)
{
RegisterDiagnosticContextParameter(context);
// All state is contained within the message context, using a single middleware instance is safe
context.Use(controllerMessageMiddleware);
next();
}
private static void RegisterDiagnosticContextParameter(IControllerBindingContext context)
{
foreach (var parameter in context.Parameters.Where(p => !p.HasBinding && p.Info.ParameterType == typeof(IDiagnosticContext)))
parameter.SetBinding(DiagnosticContextFactory);
}
private static object DiagnosticContextFactory(IMessageContext context)
{
return context.TryGet<DiagnosticContextPayload>(out var diagnosticContextPayload)
? diagnosticContextPayload.DiagnosticContext
: null;
}
}
}

63
Tapeti.Serilog/Middleware/MessageHandlerLoggingMessageMiddleware.cs Normal file
View File

@ -0,0 +1,63 @@
using System;
using System.Diagnostics;
using System.Threading.Tasks;
using Serilog.Events;
using Tapeti.Config;
using Tapeti.Serilog.Default;
namespace Tapeti.Serilog.Middleware
{
/// <summary>
/// Implements the message handler logging middleware which provides an IDiagnosticContext for
/// the message handler and logs the result.
/// </summary>
public class MessageHandlerLoggingMessageMiddleware : IControllerMessageMiddleware
{
private readonly double elapsedWarningTreshold;
private readonly LogEventLevel defaultLevel;
/// <summary>
/// Creates a new instance of the message handler logging middleware which provides an IDiagnosticContext
/// for the message handler and logs the result.
/// </summary>
/// <param name="elapsedWarningTreshold">The time (in milliseconds) a message handler is allowed to run without a warning being logged</param>
/// <param name="defaultLevel">The default log level when a message handler completes within the elapsedWarningTreshold</param>
public MessageHandlerLoggingMessageMiddleware(double elapsedWarningTreshold = 500, LogEventLevel defaultLevel = LogEventLevel.Debug)
{
this.elapsedWarningTreshold = elapsedWarningTreshold;
this.defaultLevel = defaultLevel;
}
/// <inheritdoc />
public async Task Handle(IMessageContext context, Func<Task> next)
{
var logger = context.Config.DependencyResolver.Resolve<global::Serilog.ILogger>();
var diagnosticContext = new DiagnosticContext(logger);
context.Store(new DiagnosticContextPayload(diagnosticContext));
var stopwatch = new Stopwatch();
stopwatch.Start();
await next();
stopwatch.Stop();
var controllerName = "Unknown";
var methodName = "Unknown";
if (context.TryGet<ControllerMessageContextPayload>(out var controllerMessageContextPayload))
{
controllerName = controllerMessageContextPayload.Binding.Controller.Name;
methodName = controllerMessageContextPayload.Binding.Method.Name;
}
var level = stopwatch.ElapsedMilliseconds > elapsedWarningTreshold ? LogEventLevel.Warning : defaultLevel;
var enrichedLogger = diagnosticContext.GetEnrichedLogger();
enrichedLogger.Write(level, "Tapeti: {controller}.{method} completed in {elapsedMilliseconds} ms",
controllerName, methodName, stopwatch.ElapsedMilliseconds);
}
}
}

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

@ -33,6 +33,6 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
</ItemGroup>
</Project>

32
Tapeti.sln
View File

@ -3,8 +3,6 @@ Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 16
VisualStudioVersion = 16.0.31005.135
MinimumVisualStudioVersion = 10.0.40219.1
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Annotations", "Tapeti.Annotations\Tapeti.Annotations.csproj", "{4B742AB2-59DD-4792-8E0F-D80B5366B844}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti", "Tapeti\Tapeti.csproj", "{2952B141-C54D-4E6F-8108-CAD735B0279F}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.DataAnnotations", "Tapeti.DataAnnotations\Tapeti.DataAnnotations.csproj", "{6504D430-AB4A-4DE3-AE76-0384591BEEE7}"
@ -21,8 +19,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Serilog", "Tapeti.Se
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Transient", "Tapeti.Transient\Tapeti.Transient.csproj", "{A6355E63-19AB-47EA-91FA-49B5E9B41F88}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.DataAnnotations.Extensions", "Tapeti.DataAnnotations.Extensions\Tapeti.DataAnnotations.Extensions.csproj", "{1AAA5A2C-EAA8-4C49-96A6-673EA1EEE831}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Examples", "Examples", "{266B9B94-A4D2-41C2-860C-24A7C3B63B56}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "01-PublishSubscribe", "Examples\01-PublishSubscribe\01-PublishSubscribe.csproj", "{8350A0AB-F0EE-48CF-9CA6-6019467101CF}"
@ -53,13 +49,11 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.UnityContainer", "Ta
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Ninject", "Tapeti.Ninject\Tapeti.Ninject.csproj", "{29478B10-FC53-4E93-ADEF-A775D9408131}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Tools", "Tools", "{62002327-46B0-4B72-B95A-594CE7F8C80D}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Tapeti.Cmd", "Tapeti.Cmd\Tapeti.Cmd.csproj", "{C8728BFC-7F97-41BC-956B-690A57B634EC}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "06-StatelessRequestResponse", "Examples\06-StatelessRequestResponse\06-StatelessRequestResponse.csproj", "{152227AA-3165-4550-8997-6EA80C84516E}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "07-ParallelizationTest", "Examples\07-ParallelizationTest\07-ParallelizationTest.csproj", "{E69E6BA5-68E7-4A4D-A38C-B2526AA66E96}"
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "07-ParallelizationTest", "Examples\07-ParallelizationTest\07-ParallelizationTest.csproj", "{E69E6BA5-68E7-4A4D-A38C-B2526AA66E96}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "08-MessageHandlerLogging", "Examples\08-MessageHandlerLogging\08-MessageHandlerLogging.csproj", "{906605A6-2CAB-4B29-B0DD-B735BF265E39}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
@ -67,10 +61,6 @@ Global
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{4B742AB2-59DD-4792-8E0F-D80B5366B844}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{4B742AB2-59DD-4792-8E0F-D80B5366B844}.Debug|Any CPU.Build.0 = Debug|Any CPU
{4B742AB2-59DD-4792-8E0F-D80B5366B844}.Release|Any CPU.ActiveCfg = Release|Any CPU
{4B742AB2-59DD-4792-8E0F-D80B5366B844}.Release|Any CPU.Build.0 = Release|Any CPU
{2952B141-C54D-4E6F-8108-CAD735B0279F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{2952B141-C54D-4E6F-8108-CAD735B0279F}.Debug|Any CPU.Build.0 = Debug|Any CPU
{2952B141-C54D-4E6F-8108-CAD735B0279F}.Release|Any CPU.ActiveCfg = Release|Any CPU
@ -103,10 +93,6 @@ Global
{A6355E63-19AB-47EA-91FA-49B5E9B41F88}.Debug|Any CPU.Build.0 = Debug|Any CPU
{A6355E63-19AB-47EA-91FA-49B5E9B41F88}.Release|Any CPU.ActiveCfg = Release|Any CPU
{A6355E63-19AB-47EA-91FA-49B5E9B41F88}.Release|Any CPU.Build.0 = Release|Any CPU
{1AAA5A2C-EAA8-4C49-96A6-673EA1EEE831}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{1AAA5A2C-EAA8-4C49-96A6-673EA1EEE831}.Debug|Any CPU.Build.0 = Debug|Any CPU
{1AAA5A2C-EAA8-4C49-96A6-673EA1EEE831}.Release|Any CPU.ActiveCfg = Release|Any CPU
{1AAA5A2C-EAA8-4C49-96A6-673EA1EEE831}.Release|Any CPU.Build.0 = Release|Any CPU
{8350A0AB-F0EE-48CF-9CA6-6019467101CF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{8350A0AB-F0EE-48CF-9CA6-6019467101CF}.Debug|Any CPU.Build.0 = Debug|Any CPU
{8350A0AB-F0EE-48CF-9CA6-6019467101CF}.Release|Any CPU.ActiveCfg = Release|Any CPU
@ -151,10 +137,6 @@ Global
{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
{29478B10-FC53-4E93-ADEF-A775D9408131}.Release|Any CPU.Build.0 = Release|Any CPU
{C8728BFC-7F97-41BC-956B-690A57B634EC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{C8728BFC-7F97-41BC-956B-690A57B634EC}.Debug|Any CPU.Build.0 = Debug|Any CPU
{C8728BFC-7F97-41BC-956B-690A57B634EC}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C8728BFC-7F97-41BC-956B-690A57B634EC}.Release|Any CPU.Build.0 = Release|Any CPU
{152227AA-3165-4550-8997-6EA80C84516E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{152227AA-3165-4550-8997-6EA80C84516E}.Debug|Any CPU.Build.0 = Debug|Any CPU
{152227AA-3165-4550-8997-6EA80C84516E}.Release|Any CPU.ActiveCfg = Release|Any CPU
@ -163,12 +145,15 @@ Global
{E69E6BA5-68E7-4A4D-A38C-B2526AA66E96}.Debug|Any CPU.Build.0 = Debug|Any CPU
{E69E6BA5-68E7-4A4D-A38C-B2526AA66E96}.Release|Any CPU.ActiveCfg = Release|Any CPU
{E69E6BA5-68E7-4A4D-A38C-B2526AA66E96}.Release|Any CPU.Build.0 = Release|Any CPU
{906605A6-2CAB-4B29-B0DD-B735BF265E39}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{906605A6-2CAB-4B29-B0DD-B735BF265E39}.Debug|Any CPU.Build.0 = Debug|Any CPU
{906605A6-2CAB-4B29-B0DD-B735BF265E39}.Release|Any CPU.ActiveCfg = Release|Any CPU
{906605A6-2CAB-4B29-B0DD-B735BF265E39}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(NestedProjects) = preSolution
{4B742AB2-59DD-4792-8E0F-D80B5366B844} = {8E757FF7-F6D7-42B1-827F-26FA95D97803}
{2952B141-C54D-4E6F-8108-CAD735B0279F} = {8E757FF7-F6D7-42B1-827F-26FA95D97803}
{6504D430-AB4A-4DE3-AE76-0384591BEEE7} = {57996ADC-18C5-4991-9F95-58D58D442461}
{14CF8F01-570B-4B84-AB4A-E0C3EC117F89} = {57996ADC-18C5-4991-9F95-58D58D442461}
@ -176,7 +161,6 @@ Global
{A190C736-E95A-4BDA-AA80-6211226DFCAD} = {99380F97-AD1A-459F-8AB3-D404E1E6AD4F}
{43AA5DF3-49D5-4795-A290-D6511502B564} = {57996ADC-18C5-4991-9F95-58D58D442461}
{A6355E63-19AB-47EA-91FA-49B5E9B41F88} = {57996ADC-18C5-4991-9F95-58D58D442461}
{1AAA5A2C-EAA8-4C49-96A6-673EA1EEE831} = {57996ADC-18C5-4991-9F95-58D58D442461}
{8350A0AB-F0EE-48CF-9CA6-6019467101CF} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}
{F3B38753-06B4-4932-84B4-A07692AD802D} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}
{D24120D4-50A2-44B6-A4EA-6ADAAEBABA84} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}
@ -188,9 +172,9 @@ Global
{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}
{C8728BFC-7F97-41BC-956B-690A57B634EC} = {62002327-46B0-4B72-B95A-594CE7F8C80D}
{152227AA-3165-4550-8997-6EA80C84516E} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}
{E69E6BA5-68E7-4A4D-A38C-B2526AA66E96} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}
{906605A6-2CAB-4B29-B0DD-B735BF265E39} = {266B9B94-A4D2-41C2-860C-24A7C3B63B56}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {B09CC2BF-B2AF-4CB6-8728-5D1D8E5C50FA}

31
Tapeti/Connection/ITapetiClient.cs
View File

@ -78,13 +78,13 @@ namespace Tapeti.Connection
/// <param name="queueName"></param>
/// <param name="consumer">The consumer implementation which will receive the messages from the queue</param>
/// <returns>The consumer tag as returned by BasicConsume.</returns>
Task<string> Consume(CancellationToken cancellationToken, string queueName, IConsumer consumer);
Task<TapetiConsumerTag> Consume(CancellationToken cancellationToken, string queueName, IConsumer consumer);
/// <summary>
/// Stops the consumer with the specified tag.
/// </summary>
/// <param name="consumerTag">The consumer tag as returned by Consume.</param>
Task Cancel(string consumerTag);
Task Cancel(TapetiConsumerTag consumerTag);
/// <summary>
/// Creates a durable queue if it does not already exist, and updates the bindings.
@ -129,4 +129,31 @@ namespace Tapeti.Connection
/// </summary>
Task Close();
}
/// <summary>
/// Represents a consumer for a specific connection.
/// </summary>
public class TapetiConsumerTag
{
/// <summary>
/// The consumer tag as determined by the AMQP protocol.
/// </summary>
public string ConsumerTag { get; }
/// <summary>
/// An internal reference to the connection on which the consume was started.
/// </summary>
public long ConnectionReference { get;}
/// <summary>
/// Creates a new instance of the TapetiConsumerTag class.
/// </summary>
public TapetiConsumerTag(long connectionReference, string consumerTag)
{
ConnectionReference = connectionReference;
ConsumerTag = consumerTag;
}
}
}

19
Tapeti/Connection/TapetiBasicConsumer.cs
View File

@ -5,6 +5,15 @@ using Tapeti.Default;
namespace Tapeti.Connection
{
/// <summary>
/// Called to report the result of a consumed message back to RabbitMQ.
/// </summary>
/// <param name="expectedConnectionReference">The connection reference on which the consumed message was received</param>
/// <param name="deliveryTag">The delivery tag of the consumed message</param>
/// <param name="result">The result which should be sent back</param>
public delegate Task ResponseFunc(long expectedConnectionReference, ulong deliveryTag, ConsumeResult result);
/// <inheritdoc />
/// <summary>
/// Implements the bridge between the RabbitMQ Client consumer and a Tapeti Consumer
@ -12,13 +21,15 @@ namespace Tapeti.Connection
internal class TapetiBasicConsumer : DefaultBasicConsumer
{
private readonly IConsumer consumer;
private readonly Func<ulong, ConsumeResult, Task> onRespond;
private readonly long connectionReference;
private readonly ResponseFunc onRespond;
/// <inheritdoc />
public TapetiBasicConsumer(IConsumer consumer, Func<ulong, ConsumeResult, Task> onRespond)
public TapetiBasicConsumer(IConsumer consumer, long connectionReference, ResponseFunc onRespond)
{
this.consumer = consumer;
this.connectionReference = connectionReference;
this.onRespond = onRespond;
}
@ -45,11 +56,11 @@ namespace Tapeti.Connection
try
{
var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), bodyArray);
await onRespond(deliveryTag, response);
await onRespond(connectionReference, deliveryTag, response);
}
catch
{
await onRespond(deliveryTag, ConsumeResult.Error);
await onRespond(connectionReference, deliveryTag, ConsumeResult.Error);
}
});
}

301
Tapeti/Connection/TapetiClient.cs
View File

@ -51,6 +51,7 @@ namespace Tapeti.Connection
// These fields must be locked using connectionLock
private readonly object connectionLock = new();
private long connectionReference;
private RabbitMQ.Client.IConnection connection;
private IModel consumeChannelModel;
private IModel publishChannelModel;
@ -200,7 +201,7 @@ namespace Tapeti.Connection
/// <inheritdoc />
public async Task<string> Consume(CancellationToken cancellationToken, string queueName, IConsumer consumer)
public async Task<TapetiConsumerTag> Consume(CancellationToken cancellationToken, string queueName, IConsumer consumer)
{
if (deletedQueues.Contains(queueName))
return null;
@ -209,6 +210,7 @@ namespace Tapeti.Connection
throw new ArgumentNullException(nameof(queueName));
long capturedConnectionReference = -1;
string consumerTag = null;
await GetTapetiChannel(TapetiChannelType.Consume).QueueRetryable(channel =>
@ -216,33 +218,52 @@ namespace Tapeti.Connection
if (cancellationToken.IsCancellationRequested)
return;
var basicConsumer = new TapetiBasicConsumer(consumer, Respond);
capturedConnectionReference = Interlocked.Read(ref connectionReference);
var basicConsumer = new TapetiBasicConsumer(consumer, capturedConnectionReference, Respond);
consumerTag = channel.BasicConsume(queueName, false, basicConsumer);
});
return consumerTag;
return new TapetiConsumerTag(capturedConnectionReference, consumerTag);
}
/// <inheritdoc />
public async Task Cancel(string consumerTag)
public async Task Cancel(TapetiConsumerTag consumerTag)
{
if (isClosing || string.IsNullOrEmpty(consumerTag))
if (isClosing || string.IsNullOrEmpty(consumerTag.ConsumerTag))
return;
var capturedConnectionReference = Interlocked.Read(ref connectionReference);
// If the connection was re-established in the meantime, don't respond with an
// invalid deliveryTag. The message will be requeued.
if (capturedConnectionReference != consumerTag.ConnectionReference)
return;
// No need for a retryable channel here, if the connection is lost
// so is the consumer.
await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
{
channel.BasicCancel(consumerTag);
// Check again as a reconnect may have occured in the meantime
var currentConnectionReference = Interlocked.Read(ref connectionReference);
if (currentConnectionReference != consumerTag.ConnectionReference)
return;
channel.BasicCancel(consumerTag.ConsumerTag);
});
}
private async Task Respond(ulong deliveryTag, ConsumeResult result)
private async Task Respond(long expectedConnectionReference, ulong deliveryTag, ConsumeResult result)
{
await GetTapetiChannel(TapetiChannelType.Consume).Queue(channel =>
{
// If the connection was re-established in the meantime, don't respond with an
// invalid deliveryTag. The message will be requeued.
var currentConnectionReference = Interlocked.Read(ref connectionReference);
if (currentConnectionReference != connectionReference)
return;
// No need for a retryable channel here, if the connection is lost we can't
// use the deliveryTag anymore.
switch (result)
@ -487,8 +508,8 @@ namespace Tapeti.Connection
await publishChannel.Reset();
// No need to close the channels as the connection will be closed
capturedConsumeModel.Dispose();
capturedPublishModel.Dispose();
capturedConsumeModel?.Dispose();
capturedPublishModel?.Dispose();
// ReSharper disable once InvertIf
if (capturedConnection != null)
@ -695,56 +716,76 @@ namespace Tapeti.Connection
if (channel != null && channel.IsOpen)
return channel;
}
// If the Disconnect quickly follows the Connect (when an error occurs that is reported back by RabbitMQ
// not related to the connection), wait for a bit to avoid spamming the connection
if ((DateTime.UtcNow - connectedDateTime).TotalMilliseconds <= MinimumConnectedReconnectDelay)
Thread.Sleep(ReconnectDelay);
// If the Disconnect quickly follows the Connect (when an error occurs that is reported back by RabbitMQ
// not related to the connection), wait for a bit to avoid spamming the connection
if ((DateTime.UtcNow - connectedDateTime).TotalMilliseconds <= MinimumConnectedReconnectDelay)
Thread.Sleep(ReconnectDelay);
var connectionFactory = new ConnectionFactory
var connectionFactory = new ConnectionFactory
{
HostName = connectionParams.HostName,
Port = connectionParams.Port,
VirtualHost = connectionParams.VirtualHost,
UserName = connectionParams.Username,
Password = connectionParams.Password,
AutomaticRecoveryEnabled = false,
TopologyRecoveryEnabled = false,
RequestedHeartbeat = TimeSpan.FromSeconds(30)
};
if (connectionParams.ClientProperties != null)
foreach (var pair in connectionParams.ClientProperties)
{
HostName = connectionParams.HostName,
Port = connectionParams.Port,
VirtualHost = connectionParams.VirtualHost,
UserName = connectionParams.Username,
Password = connectionParams.Password,
AutomaticRecoveryEnabled = false,
TopologyRecoveryEnabled = false,
RequestedHeartbeat = TimeSpan.FromSeconds(30)
};
if (connectionFactory.ClientProperties.ContainsKey(pair.Key))
connectionFactory.ClientProperties[pair.Key] = Encoding.UTF8.GetBytes(pair.Value);
else
connectionFactory.ClientProperties.Add(pair.Key, Encoding.UTF8.GetBytes(pair.Value));
}
if (connectionParams.ClientProperties != null)
foreach (var pair in connectionParams.ClientProperties)
while (true)
{
try
{
RabbitMQ.Client.IConnection capturedConnection;
IModel capturedConsumeChannelModel;
IModel capturedPublishChannelModel;
lock (connectionLock)
{
if (connectionFactory.ClientProperties.ContainsKey(pair.Key))
connectionFactory.ClientProperties[pair.Key] = Encoding.UTF8.GetBytes(pair.Value);
else
connectionFactory.ClientProperties.Add(pair.Key, Encoding.UTF8.GetBytes(pair.Value));
capturedConnection = connection;
}
while (true)
{
try
if (capturedConnection != null)
{
if (connection != null)
try
{
try
{
if (connection.IsOpen)
connection.Close();
}
finally
{
connection.Dispose();
}
connection = null;
}
catch (AlreadyClosedException)
{
}
finally
{
connection.Dispose();
}
logger.Connect(new ConnectContext(connectionParams, isReconnect));
connection = null;
}
logger.Connect(new ConnectContext(connectionParams, isReconnect));
Interlocked.Increment(ref connectionReference);
lock (connectionLock)
{
connection = connectionFactory.CreateConnection();
capturedConnection = connection;
consumeChannelModel = connection.CreateModel();
if (consumeChannel == null)
throw new BrokerUnreachableException(null);
@ -753,98 +794,102 @@ namespace Tapeti.Connection
if (publishChannel == null)
throw new BrokerUnreachableException(null);
capturedConsumeChannelModel = consumeChannelModel;
capturedPublishChannelModel = publishChannelModel;
}
if (config.Features.PublisherConfirms)
if (config.Features.PublisherConfirms)
{
lastDeliveryTag = 0;
Monitor.Enter(confirmLock);
try
{
lastDeliveryTag = 0;
foreach (var pair in confirmMessages)
pair.Value.CompletionSource.SetCanceled();
Monitor.Enter(confirmLock);
try
{
foreach (var pair in confirmMessages)
pair.Value.CompletionSource.SetCanceled();
confirmMessages.Clear();
}
finally
{
Monitor.Exit(confirmLock);
}
publishChannelModel.ConfirmSelect();
confirmMessages.Clear();
}
finally
{
Monitor.Exit(confirmLock);
}
if (connectionParams.PrefetchCount > 0)
consumeChannelModel.BasicQos(0, connectionParams.PrefetchCount, false);
var capturedConsumeChannelModel = consumeChannelModel;
consumeChannelModel.ModelShutdown += (_, e) =>
{
lock (connectionLock)
{
if (consumeChannelModel == null || consumeChannelModel != capturedConsumeChannelModel)
return;
consumeChannelModel = null;
}
ConnectionEventListener?.Disconnected(new DisconnectedEventArgs
{
ReplyCode = e.ReplyCode,
ReplyText = e.ReplyText
});
logger.Disconnect(new DisconnectContext(connectionParams, e.ReplyCode, e.ReplyText));
// Reconnect if the disconnect was unexpected
if (!isClosing)
GetTapetiChannel(TapetiChannelType.Consume).QueueRetryable(_ => { });
};
var capturedPublishChannelModel = publishChannelModel;
publishChannelModel.ModelShutdown += (_, _) =>
{
lock (connectionLock)
{
if (publishChannelModel == null || publishChannelModel != capturedPublishChannelModel)
return;
publishChannelModel = null;
}
// No need to reconnect, the next Publish will
};
publishChannelModel.BasicReturn += HandleBasicReturn;
publishChannelModel.BasicAcks += HandleBasicAck;
publishChannelModel.BasicNacks += HandleBasicNack;
connectedDateTime = DateTime.UtcNow;
var connectedEventArgs = new ConnectedEventArgs
{
ConnectionParams = connectionParams,
LocalPort = connection.LocalPort
};
if (isReconnect)
ConnectionEventListener?.Reconnected(connectedEventArgs);
else
ConnectionEventListener?.Connected(connectedEventArgs);
logger.ConnectSuccess(new ConnectContext(connectionParams, isReconnect, connection.LocalPort));
isReconnect = true;
break;
capturedPublishChannelModel.ConfirmSelect();
}
catch (BrokerUnreachableException e)
if (connectionParams.PrefetchCount > 0)
capturedPublishChannelModel.BasicQos(0, connectionParams.PrefetchCount, false);
capturedPublishChannelModel.ModelShutdown += (_, e) =>
{
logger.ConnectFailed(new ConnectContext(connectionParams, isReconnect, exception: e));
Thread.Sleep(ReconnectDelay);
}
lock (connectionLock)
{
if (consumeChannelModel == null || consumeChannelModel != capturedConsumeChannelModel)
return;
consumeChannelModel = null;
}
ConnectionEventListener?.Disconnected(new DisconnectedEventArgs
{
ReplyCode = e.ReplyCode,
ReplyText = e.ReplyText
});
logger.Disconnect(new DisconnectContext(connectionParams, e.ReplyCode, e.ReplyText));
// Reconnect if the disconnect was unexpected
if (!isClosing)
GetTapetiChannel(TapetiChannelType.Consume).QueueRetryable(_ => { });
};
capturedPublishChannelModel.ModelShutdown += (_, _) =>
{
lock (connectionLock)
{
if (publishChannelModel == null || publishChannelModel != capturedPublishChannelModel)
return;
publishChannelModel = null;
}
// No need to reconnect, the next Publish will
};
capturedPublishChannelModel.BasicReturn += HandleBasicReturn;
capturedPublishChannelModel.BasicAcks += HandleBasicAck;
capturedPublishChannelModel.BasicNacks += HandleBasicNack;
connectedDateTime = DateTime.UtcNow;
var connectedEventArgs = new ConnectedEventArgs
{
ConnectionParams = connectionParams,
LocalPort = capturedConnection.LocalPort
};
if (isReconnect)
ConnectionEventListener?.Reconnected(connectedEventArgs);
else
ConnectionEventListener?.Connected(connectedEventArgs);
logger.ConnectSuccess(new ConnectContext(connectionParams, isReconnect, capturedConnection.LocalPort));
isReconnect = true;
break;
}
catch (BrokerUnreachableException e)
{
logger.ConnectFailed(new ConnectContext(connectionParams, isReconnect, exception: e));
Thread.Sleep(ReconnectDelay);
}
}
lock (connectionLock)
{
return channelType == TapetiChannelType.Publish
? publishChannelModel
: consumeChannelModel;

4
Tapeti/Connection/TapetiSubscriber.cs
View File

@ -1,4 +1,4 @@
using System;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
@ -13,7 +13,7 @@ namespace Tapeti.Connection
private readonly Func<ITapetiClient> clientFactory;
private readonly ITapetiConfig config;
private bool consuming;
private readonly List<string> consumerTags = new();
private readonly List<TapetiConsumerTag> consumerTags = new();
private CancellationTokenSource initializeCancellationTokenSource;

7
Tapeti/Tapeti.csproj
View File

@ -25,10 +25,6 @@
<PackageReference Include="System.Configuration.ConfigurationManager" Version="5.0.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Tapeti.Annotations\Tapeti.Annotations.csproj" />
</ItemGroup>
<ItemGroup>
<None Include="..\resources\icons\Tapeti.png">
<Pack>True</Pack>
@ -37,6 +33,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
<PackageReference Include="Tapeti.Annotations" Version="3.0.0" />
</ItemGroup>
</Project>

43
appveyor.yml
View File

@ -11,29 +11,20 @@ before_build:
after_build:
# Create NuGet packages
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti\Tapeti.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.Annotations\Tapeti.Annotations.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.DataAnnotations\Tapeti.DataAnnotations.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.DataAnnotations.Extensions\Tapeti.DataAnnotations.Extensions.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.Flow\Tapeti.Flow.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.Flow.SQL\Tapeti.Flow.SQL.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.Transient\Tapeti.Transient.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.Serilog\Tapeti.Serilog.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.SimpleInjector\Tapeti.SimpleInjector.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.Autofac\Tapeti.Autofac.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.CastleWindsor\Tapeti.CastleWindsor.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.Ninject\Tapeti.Ninject.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true --output output Tapeti.UnityContainer\Tapeti.UnityContainer.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
# Create Tapeti.Cmd release
- cmd: dotnet publish -c Release -r win-x64 --self-contained=true -o publish\x64\selfcontained Tapeti.Cmd\Tapeti.Cmd.csproj
- cmd: dotnet publish -c Release -r win-x64 --self-contained=false -o publish\x64 Tapeti.Cmd\Tapeti.Cmd.csproj
- cmd: copy publish\x64\selfcontained\Tapeti.Cmd.exe publish\x64
- cmd: rmdir /s /q publish\x64\selfcontained
- cmd: 7z a output\Tapeti.Cmd-x64-%GitVersion_NuGetVersion%.zip %APPVEYOR_BUILD_FOLDER%\publish\x64\*
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti\Tapeti.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.DataAnnotations\Tapeti.DataAnnotations.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.Flow\Tapeti.Flow.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.Flow.SQL\Tapeti.Flow.SQL.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.Transient\Tapeti.Transient.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.Serilog\Tapeti.Serilog.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.SimpleInjector\Tapeti.SimpleInjector.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.Autofac\Tapeti.Autofac.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.CastleWindsor\Tapeti.CastleWindsor.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.Ninject\Tapeti.Ninject.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
- cmd: dotnet pack -p:IncludeSymbols=true -p:SymbolPackageFormat=snupkg -p:PublishRepositoryUrl=true -p:EmbedUntrackedSources=true --output output Tapeti.UnityContainer\Tapeti.UnityContainer.csproj /p:Configuration=Release /p:Version=%GitVersion_NuGetVersion%
# Push artifacts
- ps: Get-ChildItem output\*.nupkg | % { Push-AppveyorArtifact $_.FullName -FileName $_.Name }
- ps: Get-ChildItem output\*.snupkg | % { Push-AppveyorArtifact $_.FullName -FileName $_.Name }
- ps: Get-ChildItem output\*.zip | % { Push-AppveyorArtifact $_.FullName -FileName $_.Name }
build:
project: Tapeti.sln
@ -47,15 +38,7 @@ configuration:
deploy:
- provider: NuGet
api_key:
secure: 3WCSZAzan66vEmHZ1q3XzfOfucuAQiA+SiCDChO/gswbxfIXUpiM1eMNASDa3qWH
secure: DWFz7brxvCOtPLGJWxMojaVzIJimMGd7BPdlW75MjbhHJuoHwG1kPxZHVdjN4tPe
skip_symbols: false
artifact: /.*(\.|\.s)nupkg/
- provider: GitHub
auth_token:
secure: dWOConKg3VTPvd9DmWOOKiX1SJCalaqKInuk9GlKQOZX2s+Bia49J7q+AHO8wFj7
artifact: /Tapeti.Cmd-.*\.zip/
draft: false
prerelease: false
on:
APPVEYOR_REPO_TAG: true
artifact: /.*(\.|\.s)nupkg/

351
docs/tapeticmd.rst
View File

@ -1,353 +1,4 @@
Tapeti.Cmd
==========
The Tapeti command-line tool provides various operations for managing messages and queues.
Some operations, like shovel, are compatible with all types of messages. However, commands like import and export can assume JSON messages, specifically those sent by Tapeti, so your results may vary.
Common parameters
-----------------
Most operations support the following parameters. All are optional.
-h <hostname>, --host <hostname>
Specifies the hostname of the RabbitMQ server. Default is localhost.
--port <port>
Specifies the AMQP port of the RabbitMQ server. Default is 5672.
-v <virtualhost>, --virtualhost <virtualhost>
Specifies the virtual host to use. Default is /.
-u <username>, --username <username>
Specifies the username to authenticate the connection. Default is guest.
-p <password>, --password <username>
Specifies the password to authenticate the connection. Default is guest.
Example:
::
.\Tapeti.Cmd.exe <operation> -h rabbitmq-server -u tapeti -p topsecret
Export
------
Fetches messages from a queue and writes them to disk.
-q <queue>, --queue <queue>
*Required*. The queue to read the messages from.
-o <target>, --output <target>
*Required*. Path or filename (depending on the chosen serialization method) where the messages will be output to.
-y, --overwrite
If the output exists, do not ask to overwrite.
-r, --remove
If specified messages are acknowledged and removed from the queue. If not messages are kept.
--skip <count>
Number of messages in the input to skip. Useful if a previous non-removing export was interrupted.
-n <count>, --maxcount <count>
Maximum number of messages to retrieve from the queue. If not specified all messages are exported.
-s <method>, --serialization <method>
The method used to serialize the message for import or export. Valid options: SingleFileJSON, EasyNetQHosepipe. Defaults to SingleFileJSON. See Serialization methods below for more information.
Example:
::
.\Tapeti.Cmd.exe export -q tapeti.example.01 -o dump.json
Import
------
Read messages from disk as previously exported and publish them to a queue.
-i <source>, --input <source>
Path or filename (depending on the chosen serialization method) where the messages will be read from.
-m <message>, --message <message>
Single message to be sent, in the same format as used for SingleFileJSON. Serialization argument has no effect when using this input. Be sure to quote the entire message, and escape quotes within the message with another quote.
-c, --pipe
Messages are read from the standard input pipe, in the same format as used for SingleFileJSON. Serialization argument has no effect when using this input.
-e, --exchange
If specified publishes to the originating exchange using the original routing key. By default these are ignored and the message is published directly to the originating queue.
--skip <count>
Number of messages in the input to skip. Useful if a previous import was interrupted.
-n <count>, --maxcount <count>
Maximum number of messages to import. If not specified all messages are imported.
-s <method>, --serialization <method>
The method used to serialize the message for import or export. Valid options: SingleFileJSON, EasyNetQHosepipe. Defaults to SingleFileJSON. See Serialization methods below for more information.
--maxrate <messages per second>
The maximum amount of messages per second to import.
--batchsize <messages per batch>
How many messages to import before pausing. Will wait for manual confirmation unless batchpausetime is specified.
--batchpausetime <seconds>
How many seconds to wait before starting the next batch if batchsize is specified.
Either input, message or pipe is required.
Example:
::
.\Tapeti.Cmd.exe import -i dump.json
Shovel
------
Reads messages from a queue and publishes them to another queue, optionally to another RabbitMQ server.
-q <queue>, --queue <queue>
*Required*. The queue to read the messages from.
-t <queue>, --targetqueue <queue>
The target queue to publish the messages to. Defaults to the source queue if a different target host, port or virtualhost is specified. Otherwise it must be different from the source queue.
-r, --remove
If specified messages are acknowledged and removed from the queue. If not messages are kept.
--skip <count>
Number of messages in the input to skip. Useful if a previous non-removing shovel was interrupted.
-n <count>, --maxcount <count>
Maximum number of messages to retrieve from the queue. If not specified all messages are exported.
--targethost <host>
Hostname of the target RabbitMQ server. Defaults to the source host. Note that you may still specify a different targetusername for example.
--targetport <port>
AMQP port of the target RabbitMQ server. Defaults to the source port.
--targetvirtualhost <virtualhost>
Virtual host used for the target RabbitMQ connection. Defaults to the source virtualhost.
--targetusername <username>
Username used to connect to the target RabbitMQ server. Defaults to the source username.
--targetpassword <password>
Password used to connect to the target RabbitMQ server. Defaults to the source password.
--maxrate <messages per second>
The maximum amount of messages per second to shovel.
--batchsize <messages per batch>
How many messages to shovel before pausing. Will wait for manual confirmation unless batchpausetime is specified.
--batchpausetime <seconds>
How many seconds to wait before starting the next batch if batchsize is specified.
Example:
::
.\Tapeti.Cmd.exe shovel -q tapeti.example.01 -t tapeti.example.06
Purge
-----
Removes all messages from a queue destructively.
-q <queue>, --queue <queue>
*Required*. The queue to purge.
--confirm
Confirms the purging of the specified queue. If not provided, an interactive prompt will ask for confirmation.
Example:
::
.\Tapeti.Cmd.exe purge -q tapeti.example.01
Declare queue
-------------
Declares a durable queue without arguments.
-q <queue>, --queue <queue>
*Required*. The queue to declare.
-b <bindings>, --bindings <bindings>
One or more bindings to add to the queue. Format: <exchange>:<routingKey>
Example:
::
.\Tapeti.Cmd.exe declarequeue -q tapeti.cmd.example -b myexchange:example.message myexchange:another.message
Bind queue
----------
Add a binding to an existing queue.
-q <queue>, --queue <queue>
*Required*. The name of the queue to add the binding(s) to.
-b <bindings>, --bindings <bindings>
One or more bindings to add to the queue. Format: <exchange>:<routingKey>
Example:
::
.\Tapeti.Cmd.exe bindqueue -q tapeti.cmd.example -b myexchange:example.message myexchange:another.message
Unbind queue
------------
Remove a binding from a queue.
-q <queue>, --queue <queue>
*Required*. The name of the queue to remove the binding(s) from.
-b <bindings>, --bindings <bindings>
One or more bindings to remove from the queue. Format: <exchange>:<routingKey>
Example:
::
.\Tapeti.Cmd.exe unbindqueue -q tapeti.cmd.example -b myexchange:example.message myexchange:another.message
Remove queue
------------
Removes a durable queue.
-q <queue>, --queue <queue>
*Required*. The name of the queue to remove.
--confirm
Confirms the removal of the specified queue. If not provided, an interactive prompt will ask for confirmation.
--confirmpurge
Confirms the removal of the specified queue even if there still are messages in the queue. If not provided, an interactive prompt will ask for confirmation.
Example:
::
.\Tapeti.Cmd.exe removequeue -q tapeti.cmd.example
Serialization methods
---------------------
For importing and exporting messages, Tapeti.Cmd supports two serialization methods.
SingleFileJSON
''''''''''''''
The default serialization method. All messages are contained in a single file, where each line is a JSON document describing the message properties and it's content.
An example message (formatted as multi-line to be more readable, but keep in mind that it **must be a single line** in the export file to be imported properly):
::
{
"DeliveryTag": 1,
"Redelivered": true,
"Exchange": "tapeti",
"RoutingKey": "quote.request",
"Queue": "tapeti.example.01",
"Properties": {
"AppId": null,
"ClusterId": null,
"ContentEncoding": null,
"ContentType": "application/json",
"CorrelationId": null,
"DeliveryMode": 2,
"Expiration": null,
"Headers": {
"classType": "Messaging.TapetiExample.QuoteRequestMessage:Messaging.TapetiExample"
},
"MessageId": null,
"Priority": null,
"ReplyTo": null,
"Timestamp": 1581600132,
"Type": null,
"UserId": null
},
"Body": {
"Amount": 2
},
"RawBody": "<JSON encoded byte array>"
}
The properties correspond to the RabbitMQ client's IBasicProperties and can be omitted if empty.
Either Body or RawBody is present. Body is used if the ContentType is set to application/json, and will contain the original message as an inline JSON object for easy manipulation. For other content types, the RawBody contains the original encoded body.
Below is a bare minimum example, assuming Tapeti style messages and the default direct-to-queue import (no --exchange parameter). Again, keep in mind that it **must be a single line** in the export file to be imported properly.
::
{
"Queue": "tapeti.example.01",
"Properties": {
"ContentType": "application/json",
"Headers": {
"classType": "Messaging.TapetiExample.QuoteRequestMessage:Messaging.TapetiExample"
}
},
"Body": {
"Amount": 2
}
}
Actual file contents will thus look like:
::
{ "Queue": "tapeti.example.01", "Properties": { "ContentType": "application/json", "Headers": { "classType": "Messaging.TapetiExample.QuoteRequestMessage:Messaging.TapetiExample" } }, "Body": { "Amount": 2 } }
EasyNetQHosepipe
''''''''''''''''
Provides compatibility with the EasyNetQ Hosepipe's dump/insert format. The source or target parameter must be a path. Each message consists of 3 files, ending in .message.txt, .properties.txt and .info.txt.
As this is only provided for emergency situations, see the source code if you want to know more about the format specification.
Generating an example
---------------------
The "example" operation is available to generate an example message in SingleFileJSON format.
::
.\Tapeti.Cmd.exe example
To save the output to a file:
::
.\Tapeti.Cmd.exe example > example.json
Tapeti.Cmd has been moved to it's own repository at https://github.com/MvRens/Tapeti.Cmd. Along with it, the documentation is now available at https://tapeticmd.readthedocs.io/.

BIN
resources/icons/Tapeti.Annotations.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 5.9 KiB