mirror of https://github.com/Squidex/squidex.git
Sebastian Stehle
8 years ago
10 changed files with 273 additions and 53 deletions
@ -0,0 +1,16 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
namespace Squidex.Infrastructure.EventSourcing |
|||
{ |
|||
internal static class Constants |
|||
{ |
|||
public const string Collection = "Events"; |
|||
|
|||
public const string LeaseCollection = "Leases"; |
|||
} |
|||
} |
|||
@ -0,0 +1,150 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Text.RegularExpressions; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.Azure.Documents; |
|||
using Microsoft.Azure.Documents.ChangeFeedProcessor.FeedProcessing; |
|||
using Newtonsoft.Json; |
|||
using Builder = Microsoft.Azure.Documents.ChangeFeedProcessor.ChangeFeedProcessorBuilder; |
|||
using Collection = Microsoft.Azure.Documents.ChangeFeedProcessor.DocumentCollectionInfo; |
|||
using Options = Microsoft.Azure.Documents.ChangeFeedProcessor.ChangeFeedProcessorOptions; |
|||
|
|||
#pragma warning disable IDE0017 // Simplify object initialization
|
|||
|
|||
namespace Squidex.Infrastructure.EventSourcing |
|||
{ |
|||
internal sealed class CosmosDbSubscription : IEventSubscription, IChangeFeedObserverFactory, IChangeFeedObserver |
|||
{ |
|||
private readonly TaskCompletionSource<bool> processorStopRequested = new TaskCompletionSource<bool>(); |
|||
private readonly Task processorTask; |
|||
private readonly CosmosDbEventStore store; |
|||
private readonly Regex regex; |
|||
private readonly string hostName; |
|||
private readonly IEventSubscriber subscriber; |
|||
|
|||
public CosmosDbSubscription(CosmosDbEventStore store, IEventSubscriber subscriber, string streamFilter, string position = null) |
|||
{ |
|||
var fromBeginning = string.IsNullOrWhiteSpace(position); |
|||
|
|||
if (fromBeginning) |
|||
{ |
|||
hostName = $"squidex.{DateTime.UtcNow.Ticks.ToString()}"; |
|||
} |
|||
else |
|||
{ |
|||
hostName = position; |
|||
} |
|||
|
|||
if (!StreamFilter.IsAll(streamFilter)) |
|||
{ |
|||
regex = new Regex(streamFilter); |
|||
} |
|||
|
|||
this.store = store; |
|||
|
|||
this.subscriber = subscriber; |
|||
|
|||
processorTask = Task.Run(async () => |
|||
{ |
|||
try |
|||
{ |
|||
Collection CreateCollection(string name) |
|||
{ |
|||
var collection = new Collection(); |
|||
|
|||
collection.CollectionName = name; |
|||
collection.DatabaseName = store.DatabaseId; |
|||
collection.MasterKey = store.MasterKey; |
|||
collection.Uri = store.ServiceUri; |
|||
|
|||
return collection; |
|||
} |
|||
|
|||
var processor = |
|||
await new Builder() |
|||
.WithFeedCollection(CreateCollection(Constants.Collection)) |
|||
.WithLeaseCollection(CreateCollection(Constants.LeaseCollection)) |
|||
.WithHostName(hostName) |
|||
.WithProcessorOptions(new Options { StartFromBeginning = fromBeginning }) |
|||
.WithObserverFactory(this) |
|||
.BuildAsync(); |
|||
|
|||
await processor.StartAsync(); |
|||
await processorStopRequested.Task; |
|||
await processor.StopAsync(); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
await subscriber.OnErrorAsync(this, ex); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
public IChangeFeedObserver CreateObserver() |
|||
{ |
|||
return this; |
|||
} |
|||
|
|||
public async Task CloseAsync(IChangeFeedObserverContext context, ChangeFeedObserverCloseReason reason) |
|||
{ |
|||
if (reason == ChangeFeedObserverCloseReason.ObserverError) |
|||
{ |
|||
await subscriber.OnErrorAsync(this, new InvalidOperationException("Change feed observer failed.")); |
|||
} |
|||
} |
|||
|
|||
public Task OpenAsync(IChangeFeedObserverContext context) |
|||
{ |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public async Task ProcessChangesAsync(IChangeFeedObserverContext context, IReadOnlyList<Document> docs, CancellationToken cancellationToken) |
|||
{ |
|||
if (!processorStopRequested.Task.IsCompleted) |
|||
{ |
|||
foreach (var document in docs) |
|||
{ |
|||
if (!processorStopRequested.Task.IsCompleted) |
|||
{ |
|||
var streamName = document.GetPropertyValue<string>("eventStream"); |
|||
|
|||
if (regex == null || regex.IsMatch(streamName)) |
|||
{ |
|||
var commit = JsonConvert.DeserializeObject<CosmosDbEventCommit>(document.ToString(), store.SerializerSettings); |
|||
|
|||
var eventStreamOffset = (int)commit.EventStreamOffset; |
|||
|
|||
foreach (var @event in commit.Events) |
|||
{ |
|||
eventStreamOffset++; |
|||
|
|||
var eventData = @event.ToEventData(); |
|||
|
|||
await subscriber.OnEventAsync(this, new StoredEvent(commit.EventStream, hostName, eventStreamOffset, eventData)); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
public void WakeUp() |
|||
{ |
|||
} |
|||
|
|||
public Task StopAsync() |
|||
{ |
|||
processorStopRequested.SetResult(true); |
|||
|
|||
return processorTask; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,22 @@ |
|||
// ==========================================================================
|
|||
// Squidex Headless CMS
|
|||
// ==========================================================================
|
|||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|||
// All rights reserved. Licensed under the MIT license.
|
|||
// ==========================================================================
|
|||
|
|||
using System; |
|||
|
|||
namespace Squidex.Infrastructure.EventSourcing |
|||
{ |
|||
public static class StreamFilter |
|||
{ |
|||
public static bool IsAll(string filter) |
|||
{ |
|||
return string.IsNullOrWhiteSpace(filter) |
|||
|| string.Equals(filter, ".*", StringComparison.OrdinalIgnoreCase) |
|||
|| string.Equals(filter, "(.*)", StringComparison.OrdinalIgnoreCase) |
|||
|| string.Equals(filter, "(.*?)", StringComparison.OrdinalIgnoreCase); |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue