mirror of https://github.com/Squidex/squidex.git
Sebastian Stehle
9 years ago
8 changed files with 335 additions and 210 deletions
@ -1,124 +0,0 @@ |
|||
// ==========================================================================
|
|||
// EventReceiverBlock.cs
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex Group
|
|||
// All rights reserved.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.Log; |
|||
using System.Threading.Tasks.Dataflow; |
|||
|
|||
namespace Squidex.Infrastructure.CQRS.Events.Internal |
|||
{ |
|||
public abstract class EventReceiverBlock<TInput, TOutput> |
|||
{ |
|||
private long lastEventNumber = -1; |
|||
|
|||
protected ISemanticLog Log { get; } |
|||
|
|||
protected IEventConsumer EventConsumer { get; } |
|||
|
|||
protected IEventConsumerInfoRepository EventConsumerInfoRepository { get; } |
|||
|
|||
public ITargetBlock<TInput> Target { get; } |
|||
|
|||
public Task Completion |
|||
{ |
|||
get { return Target.Completion; } |
|||
} |
|||
|
|||
protected EventReceiverBlock(bool transform, IEventConsumer eventConsumer, IEventConsumerInfoRepository eventConsumerInfoRepository, ISemanticLog log) |
|||
{ |
|||
EventConsumer = eventConsumer; |
|||
EventConsumerInfoRepository = eventConsumerInfoRepository; |
|||
|
|||
Log = log; |
|||
|
|||
if (transform) |
|||
{ |
|||
var nullHandlerBlock = |
|||
new ActionBlock<TOutput>(_ => { }); |
|||
|
|||
var transformBlock = |
|||
new TransformBlock<TInput, TOutput>(new Func<TInput, Task<TOutput>>(HandleAsync), |
|||
new ExecutionDataflowBlockOptions { BoundedCapacity = 1 }); |
|||
transformBlock.LinkTo(nullHandlerBlock, new DataflowLinkOptions { PropagateCompletion = true }, x => x == null); |
|||
|
|||
Target = transformBlock; |
|||
} |
|||
else |
|||
{ |
|||
Target = |
|||
new ActionBlock<TInput>(HandleAsync, |
|||
new ExecutionDataflowBlockOptions { BoundedCapacity = 1 }); |
|||
} |
|||
} |
|||
|
|||
public Task NextAsync(TInput input) |
|||
{ |
|||
return Target.SendAsync(input); |
|||
} |
|||
|
|||
public void NextOrThrowAway(TInput input) |
|||
{ |
|||
Target.Post(input); |
|||
} |
|||
|
|||
public void Complete() |
|||
{ |
|||
Target.Complete(); |
|||
} |
|||
|
|||
public void Reset() |
|||
{ |
|||
lastEventNumber = -1; |
|||
} |
|||
|
|||
public void LinkTo(ITargetBlock<TOutput> other) |
|||
{ |
|||
if (Target is TransformBlock<TInput, TOutput> transformBlock) |
|||
{ |
|||
transformBlock.LinkTo(other, new DataflowLinkOptions { PropagateCompletion = true }, e => e != null); |
|||
} |
|||
} |
|||
|
|||
protected abstract Task<TOutput> On(TInput input); |
|||
|
|||
protected abstract long GetEventNumber(TInput input); |
|||
|
|||
private async Task<TOutput> HandleAsync(TInput input) |
|||
{ |
|||
try |
|||
{ |
|||
var eventNumber = GetEventNumber(input); |
|||
|
|||
if (eventNumber > lastEventNumber) |
|||
{ |
|||
var envelope = await On(input); |
|||
|
|||
lastEventNumber = eventNumber; |
|||
|
|||
return envelope; |
|||
} |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
Log.LogFatal(ex, w => w.WriteProperty("action", "EventHandlingFailed")); |
|||
|
|||
try |
|||
{ |
|||
await EventConsumerInfoRepository.StopAsync(EventConsumer.Name, ex.ToString()); |
|||
} |
|||
catch (Exception ex2) |
|||
{ |
|||
Log.LogFatal(ex2, w => w.WriteProperty("action", "EventHandlingFailed")); |
|||
} |
|||
} |
|||
|
|||
return default(TOutput); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,21 @@ |
|||
// ==========================================================================
|
|||
// IEventReceiverBlock.cs
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex Group
|
|||
// All rights reserved.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
|
|||
namespace Squidex.Infrastructure.CQRS.Events.Internal |
|||
{ |
|||
public interface IEventReceiverBlock |
|||
{ |
|||
Action<Exception> OnError { get; set; } |
|||
|
|||
void Reset(); |
|||
|
|||
void Stop(); |
|||
} |
|||
} |
|||
@ -1,33 +1,86 @@ |
|||
// ==========================================================================
|
|||
// UpdateStateBlock.cs
|
|||
// DispatchEventBlock.cs
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex Group
|
|||
// All rights reserved.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using System.Threading.Tasks.Dataflow; |
|||
using Squidex.Infrastructure.Log; |
|||
|
|||
namespace Squidex.Infrastructure.CQRS.Events.Internal |
|||
{ |
|||
public sealed class UpdateStateBlock : EventReceiverBlock<Envelope<IEvent>, Envelope<IEvent>> |
|||
internal sealed class UpdateStateBlock : IEventReceiverBlock |
|||
{ |
|||
public UpdateStateBlock(IEventConsumer eventConsumer, IEventConsumerInfoRepository eventConsumerInfoRepository, ISemanticLog log) |
|||
: base(false, eventConsumer, eventConsumerInfoRepository, log) |
|||
private readonly ISemanticLog log; |
|||
private readonly IEventConsumerInfoRepository eventConsumerInfoRepository; |
|||
private readonly IEventConsumer eventConsumer; |
|||
private readonly ActionBlock<Envelope<IEvent>> actionBlock; |
|||
private long lastReceivedEventNumber = -1; |
|||
private bool isRunning = true; |
|||
|
|||
public Action<Exception> OnError { get; set; } |
|||
|
|||
public ITargetBlock<Envelope<IEvent>> Target |
|||
{ |
|||
get { return actionBlock; } |
|||
} |
|||
|
|||
public Task Completion |
|||
{ |
|||
get { return actionBlock.Completion; } |
|||
} |
|||
|
|||
public UpdateStateBlock(IEventConsumerInfoRepository eventConsumerInfoRepository, IEventConsumer eventConsumer, ISemanticLog log) |
|||
{ |
|||
this.eventConsumerInfoRepository = eventConsumerInfoRepository; |
|||
this.eventConsumer = eventConsumer; |
|||
|
|||
this.log = log; |
|||
|
|||
actionBlock = |
|||
new ActionBlock<Envelope<IEvent>>(HandleAsync, |
|||
new ExecutionDataflowBlockOptions { BoundedCapacity = 1 }); |
|||
} |
|||
|
|||
public void Stop() |
|||
{ |
|||
isRunning = false; |
|||
} |
|||
|
|||
protected override async Task<Envelope<IEvent>> On(Envelope<IEvent> input) |
|||
public void Reset() |
|||
{ |
|||
await EventConsumerInfoRepository.SetLastHandledEventNumberAsync(EventConsumer.Name, input.Headers.EventNumber()); |
|||
isRunning = true; |
|||
|
|||
return input; |
|||
lastReceivedEventNumber = -1; |
|||
} |
|||
|
|||
protected override long GetEventNumber(Envelope<IEvent> input) |
|||
private async Task HandleAsync(Envelope<IEvent> input) |
|||
{ |
|||
return input.Headers.EventNumber(); |
|||
var eventNumber = input.Headers.EventNumber(); |
|||
|
|||
if (eventNumber <= lastReceivedEventNumber || !isRunning) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
try |
|||
{ |
|||
await eventConsumerInfoRepository.SetLastHandledEventNumberAsync(eventConsumer.Name, eventNumber); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
OnError?.Invoke(ex); |
|||
|
|||
log.LogFatal(ex, w => w |
|||
.WriteProperty("action", "UpdateState") |
|||
.WriteProperty("state", "Failed") |
|||
.WriteProperty("eventId", input.Headers.EventId().ToString()) |
|||
.WriteProperty("eventNumber", input.Headers.EventNumber())); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
Loading…
Reference in new issue