mirror of https://github.com/Squidex/squidex.git
Browse Source
* Bring event store back. * Progress with event store3. * Fix offset and position handling.pull/785/head
Sebastian Stehle
4 years ago
committed by
GitHub
GitHub
18 changed files with 491 additions and 322 deletions
@ -0,0 +1,79 @@ |
|||||
|
// ==========================================================================
|
||||
|
// Squidex Headless CMS
|
||||
|
// ==========================================================================
|
||||
|
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
||||
|
// All rights reserved. Licensed under the MIT license.
|
||||
|
// ==========================================================================
|
||||
|
|
||||
|
using System; |
||||
|
using System.Collections.Concurrent; |
||||
|
using System.Threading.Tasks; |
||||
|
using EventStore.Client; |
||||
|
using Squidex.Text; |
||||
|
|
||||
|
namespace Squidex.Infrastructure.EventSourcing |
||||
|
{ |
||||
|
public sealed class EventStoreProjectionClient |
||||
|
{ |
||||
|
private readonly ConcurrentDictionary<string, bool> projections = new ConcurrentDictionary<string, bool>(); |
||||
|
private readonly string projectionPrefix; |
||||
|
private readonly EventStoreProjectionManagementClient client; |
||||
|
|
||||
|
public EventStoreProjectionClient(EventStoreClientSettings settings, string projectionPrefix) |
||||
|
{ |
||||
|
client = new EventStoreProjectionManagementClient(settings); |
||||
|
|
||||
|
this.projectionPrefix = projectionPrefix; |
||||
|
} |
||||
|
|
||||
|
private string CreateFilterProjectionName(string filter) |
||||
|
{ |
||||
|
return $"by-{projectionPrefix.Slugify()}-{filter.Slugify()}"; |
||||
|
} |
||||
|
|
||||
|
public async Task<string> CreateProjectionAsync(string? streamFilter = null) |
||||
|
{ |
||||
|
if (!string.IsNullOrWhiteSpace(streamFilter) && streamFilter[0] != '^') |
||||
|
{ |
||||
|
return $"{projectionPrefix}-{streamFilter}"; |
||||
|
} |
||||
|
|
||||
|
streamFilter ??= ".*"; |
||||
|
|
||||
|
var name = CreateFilterProjectionName(streamFilter); |
||||
|
|
||||
|
var query = |
||||
|
$@"fromAll()
|
||||
|
.when({{ |
||||
|
$any: function (s, e) {{ |
||||
|
if (e.streamId.indexOf('{projectionPrefix}') === 0 && /{streamFilter}/.test(e.streamId.substring({projectionPrefix.Length + 1}))) {{ |
||||
|
linkTo('{name}', e); |
||||
|
}} |
||||
|
}} |
||||
|
}});";
|
||||
|
|
||||
|
await CreateProjectionAsync(name, query); |
||||
|
|
||||
|
return name; |
||||
|
} |
||||
|
|
||||
|
private async Task CreateProjectionAsync(string name, string query) |
||||
|
{ |
||||
|
if (projections.TryAdd(name, true)) |
||||
|
{ |
||||
|
try |
||||
|
{ |
||||
|
await client.CreateContinuousAsync(name, "fromAll().when()"); |
||||
|
await client.UpdateAsync(name, query, true); |
||||
|
} |
||||
|
catch (Exception ex) |
||||
|
{ |
||||
|
if (!ex.Is<InvalidOperationException>()) |
||||
|
{ |
||||
|
throw; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -1,142 +0,0 @@ |
|||||
// ==========================================================================
|
|
||||
// Squidex Headless CMS
|
|
||||
// ==========================================================================
|
|
||||
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
|
||||
// All rights reserved. Licensed under the MIT license.
|
|
||||
// ==========================================================================
|
|
||||
|
|
||||
using System; |
|
||||
using System.Collections.Concurrent; |
|
||||
using System.Linq; |
|
||||
using System.Net; |
|
||||
using System.Net.Http; |
|
||||
using System.Net.Sockets; |
|
||||
using System.Threading.Tasks; |
|
||||
using EventStore.ClientAPI; |
|
||||
using EventStore.ClientAPI.Exceptions; |
|
||||
using EventStore.ClientAPI.Projections; |
|
||||
using Squidex.Hosting.Configuration; |
|
||||
using Squidex.Text; |
|
||||
|
|
||||
namespace Squidex.Infrastructure.EventSourcing |
|
||||
{ |
|
||||
public sealed class ProjectionClient |
|
||||
{ |
|
||||
private readonly ConcurrentDictionary<string, bool> projections = new ConcurrentDictionary<string, bool>(); |
|
||||
private readonly IEventStoreConnection connection; |
|
||||
private readonly string projectionPrefix; |
|
||||
private readonly string projectionHost; |
|
||||
private ProjectionsManager projectionsManager; |
|
||||
|
|
||||
public ProjectionClient(IEventStoreConnection connection, string projectionPrefix, string projectionHost) |
|
||||
{ |
|
||||
this.connection = connection; |
|
||||
|
|
||||
this.projectionPrefix = projectionPrefix; |
|
||||
this.projectionHost = projectionHost; |
|
||||
} |
|
||||
|
|
||||
private string CreateFilterProjectionName(string filter) |
|
||||
{ |
|
||||
return $"by-{projectionPrefix.Slugify()}-{filter.Slugify()}"; |
|
||||
} |
|
||||
|
|
||||
public async Task<string> CreateProjectionAsync(string? streamFilter = null) |
|
||||
{ |
|
||||
streamFilter ??= ".*"; |
|
||||
|
|
||||
var name = CreateFilterProjectionName(streamFilter); |
|
||||
|
|
||||
var query = |
|
||||
$@"fromAll()
|
|
||||
.when({{ |
|
||||
$any: function (s, e) {{ |
|
||||
if (e.streamId.indexOf('{projectionPrefix}') === 0 && /{streamFilter}/.test(e.streamId.substring({projectionPrefix.Length + 1}))) {{ |
|
||||
linkTo('{name}', e); |
|
||||
}} |
|
||||
}} |
|
||||
}});";
|
|
||||
|
|
||||
await CreateProjectionAsync(name, query); |
|
||||
|
|
||||
return name; |
|
||||
} |
|
||||
|
|
||||
private async Task CreateProjectionAsync(string name, string query) |
|
||||
{ |
|
||||
if (projections.TryAdd(name, true)) |
|
||||
{ |
|
||||
try |
|
||||
{ |
|
||||
var credentials = connection.Settings.DefaultUserCredentials; |
|
||||
|
|
||||
await projectionsManager.CreateContinuousAsync(name, query, credentials); |
|
||||
} |
|
||||
catch (Exception ex) |
|
||||
{ |
|
||||
if (!ex.Is<ProjectionCommandConflictException>()) |
|
||||
{ |
|
||||
throw; |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
public async Task ConnectAsync() |
|
||||
{ |
|
||||
var addressParts = projectionHost.Split(':'); |
|
||||
|
|
||||
if (addressParts.Length < 2 || !int.TryParse(addressParts[1], out var port)) |
|
||||
{ |
|
||||
port = 2113; |
|
||||
} |
|
||||
|
|
||||
var endpoints = await Dns.GetHostAddressesAsync(addressParts[0]); |
|
||||
var endpoint = new IPEndPoint(endpoints.First(x => x.AddressFamily == AddressFamily.InterNetwork), port); |
|
||||
|
|
||||
async Task ConnectToSchemaAsync(string schema) |
|
||||
{ |
|
||||
projectionsManager = |
|
||||
new ProjectionsManager( |
|
||||
connection.Settings.Log, endpoint, |
|
||||
connection.Settings.OperationTimeout, |
|
||||
null, |
|
||||
schema); |
|
||||
|
|
||||
await projectionsManager.ListAllAsync(connection.Settings.DefaultUserCredentials); |
|
||||
} |
|
||||
|
|
||||
try |
|
||||
{ |
|
||||
try |
|
||||
{ |
|
||||
await ConnectToSchemaAsync("https"); |
|
||||
} |
|
||||
catch (HttpRequestException) |
|
||||
{ |
|
||||
await ConnectToSchemaAsync("http"); |
|
||||
} |
|
||||
catch (AggregateException ex) when (ex.Flatten().InnerException is HttpRequestException) |
|
||||
{ |
|
||||
await ConnectToSchemaAsync("http"); |
|
||||
} |
|
||||
} |
|
||||
catch (Exception ex) |
|
||||
{ |
|
||||
var error = new ConfigurationError($"GetEventStore cannot connect to event store projections: {projectionHost}."); |
|
||||
|
|
||||
throw new ConfigurationException(error, ex); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
public static long? ParsePositionOrNull(string? position) |
|
||||
{ |
|
||||
return long.TryParse(position, out var parsedPosition) ? (long?)parsedPosition : null; |
|
||||
} |
|
||||
|
|
||||
public static long ParsePosition(string? position) |
|
||||
{ |
|
||||
return long.TryParse(position, out var parsedPosition) ? parsedPosition + 1 : StreamPosition.Start; |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -0,0 +1,83 @@ |
|||||
|
// ==========================================================================
|
||||
|
// Squidex Headless CMS
|
||||
|
// ==========================================================================
|
||||
|
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
||||
|
// All rights reserved. Licensed under the MIT license.
|
||||
|
// ==========================================================================
|
||||
|
|
||||
|
using System.Collections.Generic; |
||||
|
using System.Globalization; |
||||
|
using System.Runtime.CompilerServices; |
||||
|
using System.Threading; |
||||
|
using EventStore.Client; |
||||
|
|
||||
|
namespace Squidex.Infrastructure.EventSourcing |
||||
|
{ |
||||
|
public static class Utils |
||||
|
{ |
||||
|
public static StreamRevision ToRevision(this long version) |
||||
|
{ |
||||
|
return StreamRevision.FromInt64(version); |
||||
|
} |
||||
|
|
||||
|
public static StreamPosition ToPosition(this long version) |
||||
|
{ |
||||
|
if (version <= 0) |
||||
|
{ |
||||
|
return StreamPosition.Start; |
||||
|
} |
||||
|
|
||||
|
return StreamPosition.FromInt64(version); |
||||
|
} |
||||
|
|
||||
|
public static StreamPosition ToPosition(this string? position, bool inclusive) |
||||
|
{ |
||||
|
if (string.IsNullOrWhiteSpace(position)) |
||||
|
{ |
||||
|
return StreamPosition.Start; |
||||
|
} |
||||
|
|
||||
|
if (long.TryParse(position, NumberStyles.Integer, CultureInfo.InvariantCulture, out var parsedPosition)) |
||||
|
{ |
||||
|
if (!inclusive) |
||||
|
{ |
||||
|
parsedPosition++; |
||||
|
} |
||||
|
|
||||
|
return StreamPosition.FromInt64(parsedPosition); |
||||
|
} |
||||
|
|
||||
|
return StreamPosition.Start; |
||||
|
} |
||||
|
|
||||
|
public static async IAsyncEnumerable<StoredEvent> IgnoreNotFound(this IAsyncEnumerable<StoredEvent> source, |
||||
|
[EnumeratorCancellation] CancellationToken ct = default) |
||||
|
{ |
||||
|
var enumerator = source.GetAsyncEnumerator(ct); |
||||
|
|
||||
|
bool resultFound; |
||||
|
try |
||||
|
{ |
||||
|
resultFound = await enumerator.MoveNextAsync(ct); |
||||
|
} |
||||
|
catch (StreamNotFoundException) |
||||
|
{ |
||||
|
resultFound = false; |
||||
|
} |
||||
|
|
||||
|
if (!resultFound) |
||||
|
{ |
||||
|
yield break; |
||||
|
} |
||||
|
|
||||
|
yield return enumerator.Current; |
||||
|
|
||||
|
while (await enumerator.MoveNextAsync(ct)) |
||||
|
{ |
||||
|
ct.ThrowIfCancellationRequested(); |
||||
|
|
||||
|
yield return enumerator.Current; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,51 @@ |
|||||
|
// ==========================================================================
|
||||
|
// Squidex Headless CMS
|
||||
|
// ==========================================================================
|
||||
|
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
||||
|
// All rights reserved. Licensed under the MIT license.
|
||||
|
// ==========================================================================
|
||||
|
|
||||
|
using System; |
||||
|
using System.Threading.Tasks; |
||||
|
using EventStore.Client; |
||||
|
using Squidex.Infrastructure.TestHelpers; |
||||
|
|
||||
|
namespace Squidex.Infrastructure.EventSourcing |
||||
|
{ |
||||
|
public sealed class GetEventStoreFixture : IDisposable |
||||
|
{ |
||||
|
private readonly EventStoreClientSettings settings; |
||||
|
|
||||
|
public GetEventStore EventStore { get; } |
||||
|
|
||||
|
public GetEventStoreFixture() |
||||
|
{ |
||||
|
AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true); |
||||
|
|
||||
|
settings = EventStoreClientSettings.Create("esdb://admin:changeit@127.0.0.1:2113?tls=false"); |
||||
|
|
||||
|
EventStore = new GetEventStore(settings, TestUtils.DefaultSerializer); |
||||
|
EventStore.InitializeAsync(default).Wait(); |
||||
|
} |
||||
|
|
||||
|
public void Dispose() |
||||
|
{ |
||||
|
CleanupAsync().Wait(); |
||||
|
} |
||||
|
|
||||
|
private async Task CleanupAsync() |
||||
|
{ |
||||
|
var projectionsManager = new EventStoreProjectionManagementClient(settings); |
||||
|
|
||||
|
await foreach (var projection in projectionsManager.ListAllAsync()) |
||||
|
{ |
||||
|
var name = projection.Name; |
||||
|
|
||||
|
if (name.StartsWith("by-squidex-test", StringComparison.OrdinalIgnoreCase)) |
||||
|
{ |
||||
|
await projectionsManager.DisableAsync(name); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,31 @@ |
|||||
|
// ==========================================================================
|
||||
|
// Squidex Headless CMS
|
||||
|
// ==========================================================================
|
||||
|
// Copyright (c) Squidex UG (haftungsbeschraenkt)
|
||||
|
// All rights reserved. Licensed under the MIT license.
|
||||
|
// ==========================================================================
|
||||
|
|
||||
|
using Xunit; |
||||
|
|
||||
|
#pragma warning disable SA1300 // Element should begin with upper-case letter
|
||||
|
|
||||
|
namespace Squidex.Infrastructure.EventSourcing |
||||
|
{ |
||||
|
[Trait("Category", "Dependencies")] |
||||
|
public class GetEventStoreTests : EventStoreTests<GetEventStore>, IClassFixture<GetEventStoreFixture> |
||||
|
{ |
||||
|
public GetEventStoreFixture _ { get; } |
||||
|
|
||||
|
protected override int SubscriptionDelayInMs { get; } = 1000; |
||||
|
|
||||
|
public GetEventStoreTests(GetEventStoreFixture fixture) |
||||
|
{ |
||||
|
_ = fixture; |
||||
|
} |
||||
|
|
||||
|
public override GetEventStore CreateStore() |
||||
|
{ |
||||
|
return _.EventStore; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue