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

Merge branch 'release/2.0.4'

Browse Source
This commit is contained in:
Mark van Renswoude 2020-01-24 14:53:17 +01:00
commit 97a7742840
8 changed files with 361 additions and 318 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

586
Tapeti.Cmd/Program.cs
View File

@ -1,293 +1,293 @@
using System; using System;
using System.Diagnostics; using System.Diagnostics;
using CommandLine; using CommandLine;
using RabbitMQ.Client; using RabbitMQ.Client;
using Tapeti.Cmd.Commands; using Tapeti.Cmd.Commands;
using Tapeti.Cmd.Serialization; using Tapeti.Cmd.Serialization;
namespace Tapeti.Cmd namespace Tapeti.Cmd
{ {
public class Program public class Program
{ {
public class CommonOptions public class CommonOptions
{ {
[Option('h', "host", HelpText = "Hostname of the RabbitMQ server.", Default = "localhost")] [Option('h', "host", HelpText = "Hostname of the RabbitMQ server.", Default = "localhost")]
public string Host { get; set; } public string Host { get; set; }
[Option('p', "port", HelpText = "AMQP port of the RabbitMQ server.", Default = 5672)] [Option("port", HelpText = "AMQP port of the RabbitMQ server.", Default = 5672)]
public int Port { get; set; } public int Port { get; set; }
[Option('v', "virtualhost", HelpText = "Virtual host used for the RabbitMQ connection.", Default = "/")] [Option('v', "virtualhost", HelpText = "Virtual host used for the RabbitMQ connection.", Default = "/")]
public string VirtualHost { get; set; } public string VirtualHost { get; set; }
[Option('u', "username", HelpText = "Username used to connect to the RabbitMQ server.", Default = "guest")] [Option('u', "username", HelpText = "Username used to connect to the RabbitMQ server.", Default = "guest")]
public string Username { get; set; } public string Username { get; set; }
[Option('p', "password", HelpText = "Password used to connect to the RabbitMQ server.", Default = "guest")] [Option('p', "password", HelpText = "Password used to connect to the RabbitMQ server.", Default = "guest")]
public string Password { get; set; } public string Password { get; set; }
} }
public enum SerializationMethod public enum SerializationMethod
{ {
SingleFileJSON, SingleFileJSON,
EasyNetQHosepipe EasyNetQHosepipe
} }
public class MessageSerializerOptions : CommonOptions public class MessageSerializerOptions : CommonOptions
{ {
[Option('s', "serialization", HelpText = "The method used to serialize the message for import or export. Valid options: SingleFileJSON, EasyNetQHosepipe.", Default = SerializationMethod.SingleFileJSON)] [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; } public SerializationMethod SerializationMethod { get; set; }
} }
[Verb("export", HelpText = "Fetch messages from a queue and write it to disk.")] [Verb("export", HelpText = "Fetch messages from a queue and write it to disk.")]
public class ExportOptions : MessageSerializerOptions public class ExportOptions : MessageSerializerOptions
{ {
[Option('q', "queue", Required = true, HelpText = "The queue to read the messages from.")] [Option('q', "queue", Required = true, HelpText = "The queue to read the messages from.")]
public string QueueName { get; set; } 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.")] [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; } public string OutputPath { get; set; }
[Option('r', "remove", HelpText = "If specified messages are acknowledged and removed from the queue. If not messages are kept.")] [Option('r', "remove", HelpText = "If specified messages are acknowledged and removed from the queue. If not messages are kept.")]
public bool RemoveMessages { get; set; } public bool RemoveMessages { get; set; }
[Option('n', "maxcount", HelpText = "(Default: all) Maximum number of messages to retrieve from the queue.")] [Option('n', "maxcount", HelpText = "(Default: all) Maximum number of messages to retrieve from the queue.")]
public int? MaxCount { get; set; } public int? MaxCount { get; set; }
} }
[Verb("import", HelpText = "Read messages from disk as previously exported and publish them to a queue.")] [Verb("import", HelpText = "Read messages from disk as previously exported and publish them to a queue.")]
public class ImportOptions : MessageSerializerOptions public class ImportOptions : MessageSerializerOptions
{ {
[Option('i', "input", Required = true, HelpText = "Path or filename (depending on the chosen serialization method) where the messages will be read from.")] [Option('i', "input", Required = true, HelpText = "Path or filename (depending on the chosen serialization method) where the messages will be read from.")]
public string Input { get; set; } public string Input { 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.")] [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; } public bool PublishToExchange { get; set; }
} }
[Verb("shovel", HelpText = "Reads messages from a queue and publishes them to another queue, optionally to another RabbitMQ server.")] [Verb("shovel", HelpText = "Reads messages from a queue and publishes them to another queue, optionally to another RabbitMQ server.")]
public class ShovelOptions : CommonOptions public class ShovelOptions : CommonOptions
{ {
[Option('q', "queue", Required = true, HelpText = "The queue to read the messages from.")] [Option('q', "queue", Required = true, HelpText = "The queue to read the messages from.")]
public string QueueName { get; set; } 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.")] [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; } 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.")] [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; } public bool RemoveMessages { get; set; }
[Option('n', "maxcount", HelpText = "(Default: all) Maximum number of messages to retrieve from the queue.")] [Option('n', "maxcount", HelpText = "(Default: all) Maximum number of messages to retrieve from the queue.")]
public int? MaxCount { get; set; } 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.")] [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; } public string TargetHost { get; set; }
[Option("targetport", HelpText = "AMQP port of the target RabbitMQ server. Defaults to the source port.")] [Option("targetport", HelpText = "AMQP port of the target RabbitMQ server. Defaults to the source port.")]
public int? TargetPort { get; set; } public int? TargetPort { get; set; }
[Option("targetvirtualhost", HelpText = "Virtual host used for the target RabbitMQ connection. Defaults to the source virtualhost.")] [Option("targetvirtualhost", HelpText = "Virtual host used for the target RabbitMQ connection. Defaults to the source virtualhost.")]
public string TargetVirtualHost { get; set; } public string TargetVirtualHost { get; set; }
[Option("targetusername", HelpText = "Username used to connect to the target RabbitMQ server. Defaults to the source username.")] [Option("targetusername", HelpText = "Username used to connect to the target RabbitMQ server. Defaults to the source username.")]
public string TargetUsername { get; set; } public string TargetUsername { get; set; }
[Option("targetpassword", HelpText = "Password used to connect to the target RabbitMQ server. Defaults to the source password.")] [Option("targetpassword", HelpText = "Password used to connect to the target RabbitMQ server. Defaults to the source password.")]
public string TargetPassword { get; set; } public string TargetPassword { get; set; }
} }
public static int Main(string[] args) public static int Main(string[] args)
{ {
return Parser.Default.ParseArguments<ExportOptions, ImportOptions, ShovelOptions>(args) return Parser.Default.ParseArguments<ExportOptions, ImportOptions, ShovelOptions>(args)
.MapResult( .MapResult(
(ExportOptions o) => ExecuteVerb(o, RunExport), (ExportOptions o) => ExecuteVerb(o, RunExport),
(ImportOptions o) => ExecuteVerb(o, RunImport), (ImportOptions o) => ExecuteVerb(o, RunImport),
(ShovelOptions o) => ExecuteVerb(o, RunShovel), (ShovelOptions o) => ExecuteVerb(o, RunShovel),
errs => errs =>
{ {
if (!Debugger.IsAttached) if (!Debugger.IsAttached)
return 1; return 1;
Console.WriteLine("Press any Enter key to continue..."); Console.WriteLine("Press any Enter key to continue...");
Console.ReadLine(); Console.ReadLine();
return 1; return 1;
} }
); );
} }
private static int ExecuteVerb<T>(T options, Action<T> execute) where T : class private static int ExecuteVerb<T>(T options, Action<T> execute) where T : class
{ {
try try
{ {
execute(options); execute(options);
return 0; return 0;
} }
catch (Exception e) catch (Exception e)
{ {
Console.WriteLine(e.Message); Console.WriteLine(e.Message);
return 1; return 1;
} }
} }
private static IConnection GetConnection(CommonOptions options) private static IConnection GetConnection(CommonOptions options)
{ {
var factory = new ConnectionFactory var factory = new ConnectionFactory
{ {
HostName = options.Host, HostName = options.Host,
Port = options.Port, Port = options.Port,
VirtualHost = options.VirtualHost, VirtualHost = options.VirtualHost,
UserName = options.Username, UserName = options.Username,
Password = options.Password Password = options.Password
}; };
return factory.CreateConnection(); return factory.CreateConnection();
} }
private static IMessageSerializer GetMessageSerializer(MessageSerializerOptions options, string path) private static IMessageSerializer GetMessageSerializer(MessageSerializerOptions options, string path)
{ {
switch (options.SerializationMethod) switch (options.SerializationMethod)
{ {
case SerializationMethod.SingleFileJSON: case SerializationMethod.SingleFileJSON:
return new SingleFileJSONMessageSerializer(path); return new SingleFileJSONMessageSerializer(path);
case SerializationMethod.EasyNetQHosepipe: case SerializationMethod.EasyNetQHosepipe:
return new EasyNetQMessageSerializer(path); return new EasyNetQMessageSerializer(path);
default: default:
throw new ArgumentOutOfRangeException(nameof(options.SerializationMethod), options.SerializationMethod, "Invalid SerializationMethod"); throw new ArgumentOutOfRangeException(nameof(options.SerializationMethod), options.SerializationMethod, "Invalid SerializationMethod");
} }
} }
private static void RunExport(ExportOptions options) private static void RunExport(ExportOptions options)
{ {
int messageCount; int messageCount;
using (var messageSerializer = GetMessageSerializer(options, options.OutputPath)) using (var messageSerializer = GetMessageSerializer(options, options.OutputPath))
using (var connection = GetConnection(options)) using (var connection = GetConnection(options))
using (var channel = connection.CreateModel()) using (var channel = connection.CreateModel())
{ {
messageCount = new ExportCommand messageCount = new ExportCommand
{ {
MessageSerializer = messageSerializer, MessageSerializer = messageSerializer,
QueueName = options.QueueName, QueueName = options.QueueName,
RemoveMessages = options.RemoveMessages, RemoveMessages = options.RemoveMessages,
MaxCount = options.MaxCount MaxCount = options.MaxCount
}.Execute(channel); }.Execute(channel);
} }
Console.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} exported."); Console.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} exported.");
} }
private static void RunImport(ImportOptions options) private static void RunImport(ImportOptions options)
{ {
int messageCount; int messageCount;
using (var messageSerializer = GetMessageSerializer(options, options.Input)) using (var messageSerializer = GetMessageSerializer(options, options.Input))
using (var connection = GetConnection(options)) using (var connection = GetConnection(options))
using (var channel = connection.CreateModel()) using (var channel = connection.CreateModel())
{ {
messageCount = new ImportCommand messageCount = new ImportCommand
{ {
MessageSerializer = messageSerializer, MessageSerializer = messageSerializer,
DirectToQueue = !options.PublishToExchange DirectToQueue = !options.PublishToExchange
}.Execute(channel); }.Execute(channel);
} }
Console.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} published."); Console.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} published.");
} }
private static void RunShovel(ShovelOptions options) private static void RunShovel(ShovelOptions options)
{ {
int messageCount; int messageCount;
using (var sourceConnection = GetConnection(options)) using (var sourceConnection = GetConnection(options))
using (var sourceChannel = sourceConnection.CreateModel()) using (var sourceChannel = sourceConnection.CreateModel())
{ {
var shovelCommand = new ShovelCommand var shovelCommand = new ShovelCommand
{ {
QueueName = options.QueueName, QueueName = options.QueueName,
TargetQueueName = !string.IsNullOrEmpty(options.TargetQueueName) ? options.TargetQueueName : options.QueueName, TargetQueueName = !string.IsNullOrEmpty(options.TargetQueueName) ? options.TargetQueueName : options.QueueName,
RemoveMessages = options.RemoveMessages, RemoveMessages = options.RemoveMessages,
MaxCount = options.MaxCount MaxCount = options.MaxCount
}; };
if (RequiresSecondConnection(options)) if (RequiresSecondConnection(options))
{ {
using (var targetConnection = GetTargetConnection(options)) using (var targetConnection = GetTargetConnection(options))
using (var targetChannel = targetConnection.CreateModel()) using (var targetChannel = targetConnection.CreateModel())
{ {
messageCount = shovelCommand.Execute(sourceChannel, targetChannel); messageCount = shovelCommand.Execute(sourceChannel, targetChannel);
} }
} }
else else
messageCount = shovelCommand.Execute(sourceChannel, sourceChannel); messageCount = shovelCommand.Execute(sourceChannel, sourceChannel);
} }
Console.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} shoveled."); Console.WriteLine($"{messageCount} message{(messageCount != 1 ? "s" : "")} shoveled.");
} }
private static bool RequiresSecondConnection(ShovelOptions options) private static bool RequiresSecondConnection(ShovelOptions options)
{ {
if (!string.IsNullOrEmpty(options.TargetHost) && options.TargetHost != options.Host) if (!string.IsNullOrEmpty(options.TargetHost) && options.TargetHost != options.Host)
return true; return true;
if (options.TargetPort.HasValue && options.TargetPort.Value != options.Port) if (options.TargetPort.HasValue && options.TargetPort.Value != options.Port)
return true; return true;
if (!string.IsNullOrEmpty(options.TargetVirtualHost) && options.TargetVirtualHost != options.VirtualHost) if (!string.IsNullOrEmpty(options.TargetVirtualHost) && options.TargetVirtualHost != options.VirtualHost)
return true; return true;
// All relevant target host parameters are either omitted or the same. This means the queue must be different // All relevant target host parameters are either omitted or the same. This means the queue must be different
// to prevent an infinite loop. // to prevent an infinite loop.
if (string.IsNullOrEmpty(options.TargetQueueName) || options.TargetQueueName == options.QueueName) 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"); 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) if (!string.IsNullOrEmpty(options.TargetUsername) && options.TargetUsername != options.Username)
return true; return true;
// ReSharper disable once ConvertIfStatementToReturnStatement // ReSharper disable once ConvertIfStatementToReturnStatement
if (!string.IsNullOrEmpty(options.TargetPassword) && options.TargetPassword != options.Password) if (!string.IsNullOrEmpty(options.TargetPassword) && options.TargetPassword != options.Password)
return true; return true;
// Everything's the same, we can use the same channel // Everything's the same, we can use the same channel
return false; return false;
} }
private static IConnection GetTargetConnection(ShovelOptions options) private static IConnection GetTargetConnection(ShovelOptions options)
{ {
var factory = new ConnectionFactory var factory = new ConnectionFactory
{ {
HostName = !string.IsNullOrEmpty(options.TargetHost) ? options.TargetHost : options.Host, HostName = !string.IsNullOrEmpty(options.TargetHost) ? options.TargetHost : options.Host,
Port = options.TargetPort ?? options.Port, Port = options.TargetPort ?? options.Port,
VirtualHost = !string.IsNullOrEmpty(options.TargetVirtualHost) ? options.TargetVirtualHost : options.VirtualHost, VirtualHost = !string.IsNullOrEmpty(options.TargetVirtualHost) ? options.TargetVirtualHost : options.VirtualHost,
UserName = !string.IsNullOrEmpty(options.TargetUsername) ? options.TargetUsername : options.Username, UserName = !string.IsNullOrEmpty(options.TargetUsername) ? options.TargetUsername : options.Username,
Password = !string.IsNullOrEmpty(options.TargetPassword) ? options.TargetPassword : options.Password, Password = !string.IsNullOrEmpty(options.TargetPassword) ? options.TargetPassword : options.Password,
}; };
return factory.CreateConnection(); return factory.CreateConnection();
} }
} }
} }

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

@ -1,22 +1,30 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework> <TargetFramework>netstandard2.0</TargetFramework>
<GenerateDocumentationFile>true</GenerateDocumentationFile> <GenerateDocumentationFile>true</GenerateDocumentationFile>
<Version>2.0.0</Version> <Version>2.0.0</Version>
</PropertyGroup> </PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'"> <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'">
<NoWarn>1701;1702</NoWarn> <NoWarn>1701;1702</NoWarn>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="System.Data.SqlClient" Version="4.6.1" /> <None Remove="scripts\Flow table.sql" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<ProjectReference Include="..\Tapeti.Flow\Tapeti.Flow.csproj" /> <EmbeddedResource Include="scripts\Flow table.sql" />
<ProjectReference Include="..\Tapeti\Tapeti.csproj" /> </ItemGroup>
</ItemGroup>
<ItemGroup>
</Project> <PackageReference Include="System.Data.SqlClient" Version="4.6.1" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Tapeti.Flow\Tapeti.Flow.csproj" />
<ProjectReference Include="..\Tapeti\Tapeti.csproj" />
</ItemGroup>
</Project>

13
Tapeti.Flow.SQL/scripts/Flow table.sql Normal file
View File

@ -0,0 +1,13 @@
/*
This script is embedded in the Tapeti.Flow.SQL package so it can be used with, for example, DbUp
*/
create table Flow
(
FlowID uniqueidentifier not null,
CreationTime datetime2(3) not null,
StateJson nvarchar(max) null,
constraint PK_Flow primary key clustered(FlowID)
);

7
Tapeti.Flow/ContextItems.cs
View File

@ -9,5 +9,12 @@
/// Key given to the FlowContext object as stored in the message context. /// Key given to the FlowContext object as stored in the message context.
/// </summary> /// </summary>
public const string FlowContext = "Tapeti.Flow.FlowContext"; public const string FlowContext = "Tapeti.Flow.FlowContext";
/// <summary>
/// Indicates if the current message handler is the last one to be called before a
/// parallel flow is done and the convergeMethod will be called.
/// Temporarily disables storing the flow state.
/// </summary>
public const string FlowIsConverging = "Tapeti.Flow.IsConverging";
} }
} }

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

@ -83,6 +83,9 @@ namespace Tapeti.Flow.Default
private static Task HandleParallelResponse(IControllerMessageContext context) private static Task HandleParallelResponse(IControllerMessageContext context)
{ {
if (context.Get<object>(ContextItems.FlowIsConverging, out _))
return Task.CompletedTask;
var flowHandler = context.Config.DependencyResolver.Resolve<IFlowHandler>(); var flowHandler = context.Config.DependencyResolver.Resolve<IFlowHandler>();
return flowHandler.Execute(new FlowHandlerContext(context), new DelegateYieldPoint(async flowContext => return flowHandler.Execute(new FlowHandlerContext(context), new DelegateYieldPoint(async flowContext =>
{ {

7
Tapeti.Flow/Default/FlowContext.cs
View File

@ -36,9 +36,14 @@ namespace Tapeti.Flow.Default
await FlowStateLock.DeleteFlowState(); await FlowStateLock.DeleteFlowState();
} }
public bool IsStoredOrDeleted()
{
return storeCalled || deleteCalled;
}
public void EnsureStoreOrDeleteIsCalled() public void EnsureStoreOrDeleteIsCalled()
{ {
if (!storeCalled && !deleteCalled) if (!IsStoredOrDeleted())
throw new InvalidProgramException("Neither Store nor Delete are called for the state of the current flow. FlowID = " + FlowStateLock?.FlowID); throw new InvalidProgramException("Neither Store nor Delete are called for the state of the current flow. FlowID = " + FlowStateLock?.FlowID);
} }

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

@ -36,6 +36,10 @@ namespace Tapeti.Flow.Default
var converge = flowContext.FlowState.Continuations.Count == 0 && var converge = flowContext.FlowState.Continuations.Count == 0 &&
flowContext.ContinuationMetadata.ConvergeMethodName != null; flowContext.ContinuationMetadata.ConvergeMethodName != null;
if (converge)
// Indicate to the FlowBindingMiddleware that the state must not to be stored
context.Store(ContextItems.FlowIsConverging, null);
await next(); await next();
if (converge) if (converge)
@ -57,7 +61,10 @@ namespace Tapeti.Flow.Default
if (flowContext?.FlowStateLock != null) if (flowContext?.FlowStateLock != null)
{ {
if (consumeResult == ConsumeResult.Error) // TODO do not call when the controller method was filtered, if the same message has two methods
if (!flowContext.IsStoredOrDeleted())
// The exception strategy can set the consume result to Success. Instead, check if the yield point
// was handled. The flow provider ensures we only end up here in case of an exception.
await flowContext.FlowStateLock.DeleteFlowState(); await flowContext.FlowStateLock.DeleteFlowState();
flowContext.FlowStateLock.Dispose(); flowContext.FlowStateLock.Dispose();

2
Tapeti/Default/ControllerMethodBinding.cs
View File

@ -184,7 +184,7 @@ namespace Tapeti.Default
{ {
await MiddlewareHelper.GoAsync( await MiddlewareHelper.GoAsync(
bindingInfo.CleanupMiddleware, bindingInfo.CleanupMiddleware,
async (handler, next) => await handler.Cleanup(context, ConsumeResult.Success, next), async (handler, next) => await handler.Cleanup(context, consumeResult, next),
() => Task.CompletedTask); () => Task.CompletedTask);
} }