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

RDB-136 Flow tabel wordt niet meer opgeruimd Bij een exceptie in Flow

Browse Source 
Refactoring om de actie van de Exceptionstrategie door te geven aan de cleanup stack
Megre bug geresolved
This commit is contained in:
Menno van Lavieren 2017-10-17 13:29:16 +02:00
parent 355c992dbf
commit be7637e784
13 changed files with 243 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Tapeti.Flow/Default/FlowCleanupMiddleware.cs
View File

@ -9,7 +9,7 @@ namespace Tapeti.Flow.Default
{
public class FlowCleanupMiddleware : ICleanupMiddleware
{
public async Task Handle(IMessageContext context, ConsumeResponse response)
public async Task Handle(IMessageContext context, HandlingResult handlingResult)
{
object flowContextObj;
if (!context.Items.TryGetValue(ContextItems.FlowContext, out flowContextObj))
@ -18,7 +18,8 @@ namespace Tapeti.Flow.Default
if (flowContext.FlowStateLock != null)
{
if (response == ConsumeResponse.Nack)
if (handlingResult.ConsumeResponse == ConsumeResponse.Nack
|| handlingResult.MessageAction == MessageAction.ErrorLog)
{
await flowContext.FlowStateLock.DeleteFlowState();
}

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

@ -52,7 +52,34 @@ namespace Tapeti.Flow.Default
};
var flowHandler = config.DependencyResolver.Resolve<IFlowHandler>();
HandlingResultBuilder handlingResult = new HandlingResultBuilder
{
ConsumeResponse = ConsumeResponse.Nack,
};
try
{
await flowHandler.Execute(context, yieldPoint);
handlingResult.ConsumeResponse = ConsumeResponse.Ack;
}
finally
{
await RunCleanup(context, handlingResult.ToHandlingResult());
}
}
private async Task RunCleanup(MessageContext context, HandlingResult handlingResult)
{
foreach (var handler in config.CleanupMiddleware)
{
try
{
await handler.Handle(context, handlingResult);
}
catch (Exception eCleanup)
{
logger.HandlerException(eCleanup);
}
}
}

2
Tapeti/Config/ICleanupMiddleware.cs
View File

@ -8,6 +8,6 @@ namespace Tapeti.Config
{
public interface ICleanupMiddleware
{
Task Handle(IMessageContext context, ConsumeResponse response);
Task Handle(IMessageContext context, HandlingResult handlingResult);
}
}

17
Tapeti/Config/IExceptionStrategyContext.cs Normal file
View File

@ -0,0 +1,17 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Tapeti.Config
{
public interface IExceptionStrategyContext
{
IMessageContext MessageContext { get; }
Exception Exception { get; }
HandlingResultBuilder HandlingResult { get; set; }
}
}

38
Tapeti/Connection/TapetiConsumer.cs
View File

@ -44,7 +44,7 @@ namespace Tapeti.Connection
{
ExceptionDispatchInfo exception = null;
MessageContext context = null;
ConsumeResponse response = ConsumeResponse.Nack;
HandlingResult handlingResult = null;
try
{
try
@ -59,7 +59,11 @@ namespace Tapeti.Connection
await DispatchMesage(context, body);
response = ConsumeResponse.Ack;
handlingResult = new HandlingResult
{
ConsumeResponse = ConsumeResponse.Ack,
MessageAction = MessageAction.None
};
}
catch (Exception eDispatch)
{
@ -67,7 +71,11 @@ namespace Tapeti.Connection
logger.HandlerException(eDispatch);
try
{
response = exceptionStrategy.HandleException(null, exception.SourceException);
var exceptionStrategyContext = new ExceptionStrategyContext(context, exception.SourceException);
exceptionStrategy.HandleException(exceptionStrategyContext);
handlingResult = exceptionStrategyContext.HandlingResult.ToHandlingResult();
}
catch (Exception eStrategy)
{
@ -76,7 +84,15 @@ namespace Tapeti.Connection
}
try
{
await RunCleanup(context, response);
if (handlingResult == null)
{
handlingResult = new HandlingResult
{
ConsumeResponse = ConsumeResponse.Nack,
MessageAction = MessageAction.None
};
}
await RunCleanup(context, handlingResult);
}
catch (Exception eCleanup)
{
@ -87,7 +103,15 @@ namespace Tapeti.Connection
{
try
{
await worker.Respond(deliveryTag, response);
if (handlingResult == null)
{
handlingResult = new HandlingResult
{
ConsumeResponse = ConsumeResponse.Nack,
MessageAction = MessageAction.None
};
}
await worker.Respond(deliveryTag, handlingResult.ConsumeResponse);
}
catch (Exception eRespond)
{
@ -108,13 +132,13 @@ namespace Tapeti.Connection
});
}
private async Task RunCleanup(MessageContext context, ConsumeResponse response)
private async Task RunCleanup(MessageContext context, HandlingResult handlingResult)
{
foreach(var handler in cleanupMiddleware)
{
try
{
await handler.Handle(context, response);
await handler.Handle(context, handlingResult);
}
catch (Exception eCleanup)
{

40
Tapeti/Default/ExceptionStrategyContext.cs Normal file
View File

@ -0,0 +1,40 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Tapeti.Config;
namespace Tapeti.Default
{
public class ExceptionStrategyContext : IExceptionStrategyContext
{
internal ExceptionStrategyContext(IMessageContext messageContext, Exception exception)
{
MessageContext = messageContext;
Exception = exception;
}
public IMessageContext MessageContext { get; }
public Exception Exception { get; }
private HandlingResultBuilder handlingResult;
public HandlingResultBuilder HandlingResult
{
get
{
if (handlingResult == null)
{
handlingResult = new HandlingResultBuilder();
}
return handlingResult;
}
set
{
handlingResult = value;
}
}
}
}

17
Tapeti/Default/NackExceptionStrategy.cs Normal file
View File

@ -0,0 +1,17 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Tapeti.Config;
namespace Tapeti.Default
{
public class NackExceptionStrategy : IExceptionStrategy
{
public void HandleException(IExceptionStrategyContext context)
{
context.HandlingResult.ConsumeResponse = ConsumeResponse.Nack;
}
}
}

5
Tapeti/Default/RequeueExceptionStrategy.cs
View File

@ -5,10 +5,9 @@ namespace Tapeti.Default
{
public class RequeueExceptionStrategy : IExceptionStrategy
{
public ConsumeResponse HandleException(IMessageContext context, Exception exception)
public void HandleException(IExceptionStrategyContext context)
{
// TODO log exception
return ConsumeResponse.Requeue;
context.HandlingResult.ConsumeResponse = ConsumeResponse.Requeue;
}
}
}

79
Tapeti/HandlingResult.cs Normal file
View File

@ -0,0 +1,79 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Tapeti
{
public class HandlingResult
{
public HandlingResult()
{
ConsumeResponse = ConsumeResponse.Nack;
MessageAction = MessageAction.None;
}
/// <summary>
/// Determines which response will be given to the message bus from where the message originates.
/// </summary>
public ConsumeResponse ConsumeResponse { get; internal set; }
/// <summary>
/// Registers which action the Exception strategy has taken or will take to handle the error condition
/// on the message. This is important to know for cleanup handlers registered by middleware.
/// </summary>
public MessageAction MessageAction { get; internal set; }
}
public class HandlingResultBuilder
{
private static readonly HandlingResult Default = new HandlingResult();
private HandlingResult data = Default;
public ConsumeResponse ConsumeResponse {
get
{
return data.ConsumeResponse;
}
set
{
GetWritableData().ConsumeResponse = value;
}
}
public MessageAction MessageAction
{
get
{
return data.MessageAction;
}
set
{
GetWritableData().MessageAction = value;
}
}
public HandlingResult ToHandlingResult()
{
if (data == Default)
{
return new HandlingResult();
}
var result = GetWritableData();
data = Default;
return result;
}
private HandlingResult GetWritableData()
{
if (data == Default)
{
data = new HandlingResult();
}
return data;
}
}
}

8
Tapeti/IExceptionStrategy.cs
View File

@ -8,9 +8,9 @@ namespace Tapeti
/// <summary>
/// Called when an exception occurs while handling a message.
/// </summary>
/// <param name="context">The message context if available. May be null!</param>
/// <param name="exception">The exception instance</param>
/// <returns>The ConsumeResponse to determine whether to requeue, dead-letter (nack) or simply ack the message.</returns>
ConsumeResponse HandleException(IMessageContext context, Exception exception);
/// <param name="context">The exception strategy context containing the necessary data including the message context and the thrown exception.
/// Also the response to the message can be set.
/// If there is any other handling of the message than the expected default than HandlingResult.MessageFutureAction must be set accordingly. </param>
void HandleException(IExceptionStrategyContext context);
}
}

15
Tapeti/MessageFutureAction.cs Normal file
View File

@ -0,0 +1,15 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Tapeti
{
public enum MessageAction
{
None = 1,
ErrorLog = 2,
Retry = 3,
}
}

5
Tapeti/Tapeti.csproj
View File

@ -53,6 +53,7 @@
<Reference Include="System.Xml" />
</ItemGroup>
<ItemGroup>
<Compile Include="Config\IExceptionStrategyContext.cs" />
<Compile Include="Config\ICleanupMiddleware.cs" />
<Compile Include="Config\IPublishContext.cs" />
<Compile Include="Config\IMessageFilterMiddleware.cs" />
@ -64,6 +65,9 @@
<Compile Include="Connection\TapetiWorker.cs" />
<Compile Include="Default\ConsoleLogger.cs" />
<Compile Include="Default\DevNullLogger.cs" />
<Compile Include="Default\ExceptionStrategyContext.cs" />
<Compile Include="Default\NackExceptionStrategy.cs" />
<Compile Include="HandlingResult.cs" />
<Compile Include="Default\JsonMessageSerializer.cs" />
<Compile Include="Default\MessageContext.cs" />
<Compile Include="Default\PublishResultBinding.cs" />
@ -84,6 +88,7 @@
<Compile Include="Config\IConfig.cs" />
<Compile Include="MessageController.cs" />
<Compile Include="Config\IBindingMiddleware.cs" />
<Compile Include="MessageFutureAction.cs" />
<Compile Include="TapetiAppSettingsConnectionParams.cs" />
<Compile Include="TapetiConnectionParams.cs" />
<Compile Include="TapetiConfig.cs" />

2
Tapeti/TapetiConfig.cs
View File

@ -143,7 +143,7 @@ namespace Tapeti
container.RegisterDefault<IMessageSerializer, JsonMessageSerializer>();
container.RegisterDefault<IExchangeStrategy, NamespaceMatchExchangeStrategy>();
container.RegisterDefault<IRoutingKeyStrategy, TypeNameRoutingKeyStrategy>();
container.RegisterDefault<IExceptionStrategy, RequeueExceptionStrategy>();
container.RegisterDefault<IExceptionStrategy, NackExceptionStrategy>();
}