Added base code for consistency exercise

Workshops/EventDrivenArchitecture/05-Consistency/05-Consistency.csproj

@@ -0,0 +1,26 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net8.0</TargetFramework>
+        <RootNamespace>Consistency</RootNamespace>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="Bogus" Version="35.6.1"/>
+        <PackageReference Include="FluentAssertions" Version="6.12.0"/>
+        <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.11.0"/>
+        <PackageReference Include="Ogooreck" Version="0.8.2"/>
+        <PackageReference Include="GitHubActionsTestLogger" Version="2.4.1">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageReference>
+        <PackageReference Include="xunit" Version="2.9.0"/>
+        <PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
+        </PackageReference>
+        <PackageReference Include="Microsoft.AspNetCore.TestHost" Version="8.0.8"/>
+    </ItemGroup>
+</Project>

Workshops/EventDrivenArchitecture/05-Consistency/Core/CommandBus.cs

@@ -0,0 +1,33 @@
+using System.Collections.Concurrent;
+
+namespace Consistency.Core;
+
+public class CommandBus
+{
+    public async ValueTask Send(object[] commands, CancellationToken ct)
+    {
+        foreach (var command in commands)
+        {
+            if (!commandHandlers.TryGetValue(command.GetType(), out var handler))
+                continue;
+
+            foreach (var middleware in middlewares)
+                middleware(command);
+
+            await handler(command, ct);
+        }
+    }
+
+    public CommandBus Handle<T>(Func<T, CancellationToken, ValueTask> eventHandler)
+    {
+        commandHandlers[typeof(T)] = (command, ct) => eventHandler((T)command, ct);
+
+        return this;
+    }
+
+    public void Use(Action<object> middleware) =>
+        middlewares.Add(middleware);
+
+    private readonly ConcurrentDictionary<Type, Func<object, CancellationToken, ValueTask>> commandHandlers = new();
+    private readonly List<Action<object>> middlewares = [];
+}

Workshops/EventDrivenArchitecture/05-Consistency/Core/Database.cs

@@ -0,0 +1,66 @@
+using System.Text.Json;
+
+namespace Consistency.Core;
+
+public class Database
+{
+    private Dictionary<string, object> storage = new();
+
+    public ValueTask Store<T>(Guid id, T obj, CancellationToken _) where T : class
+    {
+        Monitor.Enter(storage);
+        try
+        {
+            if (DateTimeOffset.Now.Ticks % 2 == 0)
+                storage[GetId<T>(id)] = obj;
+
+            return ValueTask.CompletedTask;
+        }
+        finally
+        {
+            Monitor.Exit(storage);
+        }
+    }
+
+    public ValueTask Delete<T>(Guid id, CancellationToken _)
+    {
+        Monitor.Enter(storage);
+        try
+        {
+            if (DateTimeOffset.Now.Ticks % 2 == 0)
+                storage.Remove(GetId<T>(id));
+
+            return ValueTask.CompletedTask;
+        }
+        finally
+        {
+            Monitor.Exit(storage);
+        }
+    }
+
+    public ValueTask<T?> Get<T>(Guid id, CancellationToken _) where T : class =>
+        ValueTask.FromResult(
+            storage.TryGetValue(GetId<T>(id), out var result)
+                ?
+                // Clone to simulate getting new instance on loading
+                JsonSerializer.Deserialize<T>(JsonSerializer.Serialize((T)result))
+                : null
+        );
+
+    public async ValueTask Transaction(Func<Database, ValueTask> action)
+    {
+        Monitor.Enter(storage);
+        try
+        {
+            var serialisedDatabase = new Database();
+
+            await action(serialisedDatabase);
+        }
+        finally
+        {
+            Monitor.Exit(storage);
+        }
+    }
+
+    private static string GetId<T>(Guid id) => $"{typeof(T).Name}-{id}";
+}

Workshops/EventDrivenArchitecture/05-Consistency/Core/EventBus.cs

@@ -0,0 +1,47 @@
+using System.Collections.Concurrent;
+
+namespace Consistency.Core;
+
+public class EventBus
+{
+    public async ValueTask Publish(object[] events, CancellationToken ct)
+    {
+        if (DateTimeOffset.Now.Ticks % 2 == 0)
+            return;
+
+        foreach (var @event in events)
+        {
+            foreach (var middleware in middlewares)
+                middleware(@event);
+
+            if (!eventHandlers.TryGetValue(@event.GetType(), out var handlers))
+                continue;
+
+            foreach (var handler in handlers)
+                await handler(@event, ct);
+        }
+    }
+
+    public EventBus Subscribe<T>(Func<T, CancellationToken, ValueTask> eventHandler)
+    {
+        Func<object, CancellationToken, ValueTask> handler = (@event, ct) => eventHandler((T)@event, ct);
+
+        eventHandlers.AddOrUpdate(
+            typeof(T),
+            _ => [handler],
+            (_, handlers) =>
+            {
+                handlers.Add(handler);
+                return handlers;
+            }
+        );
+
+        return this;
+    }
+
+    public void Use(Action<object> middleware) =>
+        middlewares.Add(middleware);
+
+    private readonly ConcurrentDictionary<Type, List<Func<object, CancellationToken, ValueTask>>> eventHandlers = new();
+    private readonly List<Action<object>> middlewares = [];
+}

Workshops/EventDrivenArchitecture/05-Consistency/Core/MessageCatcher.cs

@@ -0,0 +1,26 @@
+using FluentAssertions;
+
+namespace Consistency.Core;
+
+public class MessageCatcher
+{
+    public List<object> Published { get; } = [];
+
+    public void Catch(object message) =>
+        Published.Add(message);
+
+    public void Reset() => Published.Clear();
+
+    public void ShouldNotReceiveAnyMessage() =>
+        Published.Should().BeEmpty();
+
+    public void ShouldReceiveSingleMessage<T>(T message)
+    {
+        Published.Should().HaveCount(1);
+        Published.OfType<T>().Should().HaveCount(1);
+        Published.Single().Should().Be(message);
+    }
+
+    public void ShouldReceiveMessages(object[] messages) =>
+        Published.Should().BeEquivalentTo(messages);
+}