mirror of https://github.com/Squidex/squidex.git
Sebastian Stehle
9 years ago
11 changed files with 488 additions and 201 deletions
@ -0,0 +1,70 @@ |
|||
// ==========================================================================
|
|||
// DispatchEventBlock.cs
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex Group
|
|||
// All rights reserved.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.Log; |
|||
|
|||
namespace Squidex.Infrastructure.CQRS.Events.Internal |
|||
{ |
|||
internal sealed class DispatchEventBlock : EventReceiverBlock<Envelope<IEvent>, Envelope<IEvent>> |
|||
{ |
|||
public DispatchEventBlock(IEventConsumer eventConsumer, IEventConsumerInfoRepository eventConsumerInfoRepository, ISemanticLog log) |
|||
: base(true, eventConsumer, eventConsumerInfoRepository, log) |
|||
{ |
|||
} |
|||
|
|||
protected override async Task<Envelope<IEvent>> On(Envelope<IEvent> input) |
|||
{ |
|||
var consumerName = EventConsumer.Name; |
|||
|
|||
var eventId = input.Headers.EventId().ToString(); |
|||
var eventType = input.Payload.GetType().Name; |
|||
try |
|||
{ |
|||
Log.LogInformation(w => w |
|||
.WriteProperty("action", "HandleEvent") |
|||
.WriteProperty("actionId", eventId) |
|||
.WriteProperty("state", "Started") |
|||
.WriteProperty("eventId", eventId) |
|||
.WriteProperty("eventType", eventType) |
|||
.WriteProperty("eventConsumer", consumerName)); |
|||
|
|||
await EventConsumer.On(input); |
|||
|
|||
Log.LogInformation(w => w |
|||
.WriteProperty("action", "HandleEvent") |
|||
.WriteProperty("actionId", eventId) |
|||
.WriteProperty("state", "Completed") |
|||
.WriteProperty("eventId", eventId) |
|||
.WriteProperty("eventType", eventType) |
|||
.WriteProperty("eventConsumer", consumerName)); |
|||
|
|||
return input; |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
Log.LogError(ex, w => w |
|||
.WriteProperty("action", "HandleEvent") |
|||
.WriteProperty("actionId", eventId) |
|||
.WriteProperty("state", "Started") |
|||
.WriteProperty("eventId", eventId) |
|||
.WriteProperty("eventType", eventType) |
|||
.WriteProperty("eventConsumer", consumerName)); |
|||
|
|||
} |
|||
|
|||
return null; |
|||
} |
|||
|
|||
protected override long GetEventNumber(Envelope<IEvent> input) |
|||
{ |
|||
return input.Headers.EventNumber(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,118 @@ |
|||
// ==========================================================================
|
|||
// 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) |
|||
{ |
|||
Target = |
|||
new TransformBlock<TInput, TOutput>(new Func<TInput, Task<TOutput>>(HandleAsync), |
|||
new ExecutionDataflowBlockOptions { BoundedCapacity = 1 }); |
|||
} |
|||
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, 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,52 @@ |
|||
// ==========================================================================
|
|||
// ParseEventBlock.cs
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex Group
|
|||
// All rights reserved.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.Log; |
|||
|
|||
namespace Squidex.Infrastructure.CQRS.Events.Internal |
|||
{ |
|||
internal sealed class ParseEventBlock : EventReceiverBlock<StoredEvent, Envelope<IEvent>> |
|||
{ |
|||
private readonly EventDataFormatter formatter; |
|||
|
|||
public ParseEventBlock(IEventConsumer eventConsumer, IEventConsumerInfoRepository eventConsumerInfoRepository, ISemanticLog log, EventDataFormatter formatter) |
|||
: base(true, eventConsumer, eventConsumerInfoRepository, log) |
|||
{ |
|||
this.formatter = formatter; |
|||
} |
|||
|
|||
protected override Task<Envelope<IEvent>> On(StoredEvent input) |
|||
{ |
|||
Envelope<IEvent> result = null; |
|||
try |
|||
{ |
|||
result = formatter.Parse(input.Data); |
|||
|
|||
result.SetEventNumber(input.EventNumber); |
|||
result.SetEventStreamNumber(input.EventStreamNumber); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
Log.LogFatal(ex, w => w |
|||
.WriteProperty("action", "ParseEvent") |
|||
.WriteProperty("state", "Failed") |
|||
.WriteProperty("eventId", input.Data.EventId.ToString()) |
|||
.WriteProperty("eventNumber", input.EventNumber)); |
|||
} |
|||
|
|||
return Task.FromResult(result); |
|||
} |
|||
|
|||
protected override long GetEventNumber(StoredEvent input) |
|||
{ |
|||
return input.EventNumber; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,102 @@ |
|||
// ==========================================================================
|
|||
// QueryEventsBlock.cs
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex Group
|
|||
// All rights reserved.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using System.Threading.Tasks.Dataflow; |
|||
using Squidex.Infrastructure.Log; |
|||
|
|||
// ReSharper disable InvertIf
|
|||
|
|||
namespace Squidex.Infrastructure.CQRS.Events.Internal |
|||
{ |
|||
internal sealed class QueryEventsBlock : EventReceiverBlock<object, object> |
|||
{ |
|||
private readonly IEventStore eventStore; |
|||
private bool isStarted; |
|||
private long handled; |
|||
|
|||
public Func<StoredEvent, Task> OnEvent { get; set; } |
|||
|
|||
public Action OnReset { get; set; } |
|||
|
|||
public QueryEventsBlock(IEventConsumer eventConsumer, IEventConsumerInfoRepository eventConsumerInfoRepository, ISemanticLog log, IEventStore eventStore) |
|||
: base(false, eventConsumer, eventConsumerInfoRepository, log) |
|||
{ |
|||
this.eventStore = eventStore; |
|||
} |
|||
|
|||
protected override async Task<object> On(object input) |
|||
{ |
|||
if (!isStarted) |
|||
{ |
|||
await EventConsumerInfoRepository.CreateAsync(EventConsumer.Name); |
|||
|
|||
isStarted = true; |
|||
} |
|||
|
|||
var status = await EventConsumerInfoRepository.FindAsync(EventConsumer.Name); |
|||
|
|||
var lastReceivedEventNumber = status.LastHandledEventNumber; |
|||
|
|||
if (status.IsResetting) |
|||
{ |
|||
await ResetAsync(); |
|||
} |
|||
|
|||
if (!status.IsStopped) |
|||
{ |
|||
var ct = CancellationToken.None; |
|||
|
|||
await eventStore.GetEventsAsync(storedEvent => OnEvent?.Invoke(storedEvent), ct, null, lastReceivedEventNumber); |
|||
} |
|||
|
|||
return null; |
|||
} |
|||
|
|||
private async Task ResetAsync() |
|||
{ |
|||
var consumerName = EventConsumer.Name; |
|||
|
|||
var actionId = Guid.NewGuid().ToString(); |
|||
try |
|||
{ |
|||
Log.LogInformation(w => w |
|||
.WriteProperty("action", "EventConsumerReset") |
|||
.WriteProperty("actionId", actionId) |
|||
.WriteProperty("state", "Started") |
|||
.WriteProperty("eventConsumer", consumerName)); |
|||
|
|||
await EventConsumer.ClearAsync(); |
|||
await EventConsumerInfoRepository.SetLastHandledEventNumberAsync(consumerName, -1); |
|||
|
|||
Log.LogInformation(w => w |
|||
.WriteProperty("action", "EventConsumerReset") |
|||
.WriteProperty("actionId", actionId) |
|||
.WriteProperty("state", "Completed") |
|||
.WriteProperty("eventConsumer", consumerName)); |
|||
|
|||
OnReset?.Invoke(); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
Log.LogFatal(ex, w => w |
|||
.WriteProperty("action", "EventConsumerReset") |
|||
.WriteProperty("actionId", actionId) |
|||
.WriteProperty("state", "Completed") |
|||
.WriteProperty("eventConsumer", consumerName)); |
|||
} |
|||
} |
|||
|
|||
protected override long GetEventNumber(object input) |
|||
{ |
|||
return handled++; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
// ==========================================================================
|
|||
// UpdateStateBlock.cs
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex Group
|
|||
// All rights reserved.
|
|||
// ==========================================================================
|
|||
|
|||
using System.Threading.Tasks; |
|||
using Squidex.Infrastructure.Log; |
|||
|
|||
namespace Squidex.Infrastructure.CQRS.Events.Internal |
|||
{ |
|||
public sealed class UpdateStateBlock : EventReceiverBlock<Envelope<IEvent>, Envelope<IEvent>> |
|||
{ |
|||
public UpdateStateBlock(IEventConsumer eventConsumer, IEventConsumerInfoRepository eventConsumerInfoRepository, ISemanticLog log) |
|||
: base(false, eventConsumer, eventConsumerInfoRepository, log) |
|||
{ |
|||
} |
|||
|
|||
protected override async Task<Envelope<IEvent>> On(Envelope<IEvent> input) |
|||
{ |
|||
await EventConsumerInfoRepository.SetLastHandledEventNumberAsync(EventConsumer.Name, input.Headers.EventNumber()); |
|||
|
|||
return input; |
|||
} |
|||
|
|||
protected override long GetEventNumber(Envelope<IEvent> input) |
|||
{ |
|||
return input.Headers.EventNumber(); |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue