Skip to content

Commit 16ddcd6

Browse files
oskardudyczoskardudycz
committed
Added examples of process managers in HotelManagement example
1 parent 6e8d737 commit 16ddcd6

32 files changed

+1363
-43
lines changed

‎Core.Marten/Aggregates/DocumentSessionExtensions.cs‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
using Core.Aggregates;
2+
using Core.Exceptions;
3+
using Marten;
4+
5+
namespace Core.Marten.Aggregates;
6+
7+
/// <summary>
8+
/// This code assumes that Aggregate:
9+
/// - is event-driven but not fully event-sourced
10+
/// - streams have string identifiers
11+
/// - aggregate is versioned, so optimistic concurrency is applied
12+
/// </summary>
13+
public static class DocumentSessionExtensions
14+
{
15+
public static Task Add<T>(
16+
this IDocumentSession documentSession,
17+
string id,
18+
T aggregate,
19+
CancellationToken ct
20+
) where T : IAggregate
21+
{
22+
documentSession.Insert(aggregate);
23+
documentSession.Events.Append($"events-{id}", aggregate.DequeueUncommittedEvents());
24+
25+
return documentSession.SaveChangesAsync(token: ct);
26+
}
27+
28+
public static async Task GetAndUpdate<T>(
29+
this IDocumentSession documentSession,
30+
string id,
31+
Action<T> handle,
32+
CancellationToken ct
33+
) where T : IAggregate
34+
{
35+
var aggregate = await documentSession.LoadAsync<T>(id, ct).ConfigureAwait(false);
36+
37+
if (aggregate is null)
38+
throw AggregateNotFoundException.For<T>(id);
39+
40+
handle(aggregate);
41+
42+
documentSession.Update(aggregate);
43+
44+
documentSession.Events.Append($"events-{id}", aggregate.DequeueUncommittedEvents());
45+
46+
await documentSession.SaveChangesAsync(token: ct).ConfigureAwait(false);
47+
}
48+
}

‎Core.Marten/ProcessManagers/DocumentSessionExtensions.cs‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
using Core.Exceptions;
2+
using Core.ProcessManagers;
3+
using Marten;
4+
5+
namespace Core.Marten.ProcessManagers;
6+
7+
/// <summary>
8+
/// This code assumes that Process Manager:
9+
/// - is event-driven but not fully event-sourced
10+
/// - streams have string identifiers
11+
/// - process manager is versioned, so optimistic concurrency is applied
12+
/// </summary>
13+
public static class DocumentSessionExtensions
14+
{
15+
public static Task Add<T>(
16+
this IDocumentSession documentSession,
17+
string id,
18+
T processManager,
19+
CancellationToken ct
20+
) where T : IProcessManager
21+
{
22+
documentSession.Insert(processManager);
23+
EnqueueMessages(documentSession, id, processManager);
24+
25+
return documentSession.SaveChangesAsync(token: ct);
26+
}
27+
28+
public static async Task GetAndUpdate<T>(
29+
this IDocumentSession documentSession,
30+
string id,
31+
Action<T> handle,
32+
CancellationToken ct
33+
) where T : IProcessManager
34+
{
35+
var processManager = await documentSession.LoadAsync<T>(id, ct).ConfigureAwait(false);
36+
37+
if (processManager is null)
38+
throw AggregateNotFoundException.For<T>(id);
39+
40+
handle(processManager);
41+
42+
documentSession.Update(processManager);
43+
44+
EnqueueMessages(documentSession, id, processManager);
45+
await documentSession.SaveChangesAsync(token: ct).ConfigureAwait(false);
46+
}
47+
48+
private static void EnqueueMessages<T>(IDocumentSession documentSession, string id, T processManager) where T : IProcessManager
49+
{
50+
foreach (var message in processManager.DequeuePendingMessages())
51+
{
52+
message.Switch(
53+
@event => documentSession.Events.Append($"events-{id}", @event),
54+
command => documentSession.Events.Append($"commands-{id}", command)
55+
);
56+
}
57+
}
58+
}

‎Core/ProcessManagers/IProcessManager.cs‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
using Core.Projections;
2+
using Core.Structures;
3+
4+
namespace Core.ProcessManagers;
5+
6+
public interface IProcessManager: IProcessManager<Guid>
7+
{
8+
}
9+
10+
public interface IProcessManager<out T>: IProjection
11+
{
12+
T Id { get; }
13+
int Version { get; }
14+
15+
EventOrCommand[] DequeuePendingMessages();
16+
}
17+
18+
public class EventOrCommand: Either<object, object>
19+
{
20+
public static EventOrCommand Event(object @event) =>
21+
new(Maybe<object>.Of(@event), Maybe<object>.Empty);
22+
23+
24+
public static EventOrCommand Command(object @event) =>
25+
new(Maybe<object>.Empty, Maybe<object>.Of(@event));
26+
27+
private EventOrCommand(Maybe<object> left, Maybe<object> right): base(left, right)
28+
{
29+
}
30+
}

‎Core/ProcessManagers/ProcessManager.cs‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
namespace Core.ProcessManagers;
2+
3+
public abstract class ProcessManager: ProcessManager<Guid>, IProcessManager
4+
{
5+
}
6+
7+
public abstract class ProcessManager<T>: IProcessManager<T> where T : notnull
8+
{
9+
public T Id { get; protected set; } = default!;
10+
11+
public int Version { get; protected set; }
12+
13+
[NonSerialized] private readonly Queue<EventOrCommand> scheduledCommands = new();
14+
15+
public EventOrCommand[] DequeuePendingMessages()
16+
{
17+
var dequeuedEvents = scheduledCommands.ToArray();
18+
19+
scheduledCommands.Clear();
20+
21+
return dequeuedEvents;
22+
}
23+
24+
protected void EnqueueEvent(object @event) =>
25+
scheduledCommands.Enqueue(EventOrCommand.Event(@event));
26+
27+
protected void ScheduleCommand(object @event) =>
28+
scheduledCommands.Enqueue(EventOrCommand.Command(@event));
29+
30+
public virtual void When(object @event)
31+
{
32+
}
33+
}

‎Core/Structures/Either.cs‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
namespace Core.Structures;
2+
3+
public class Either<TLeft, TRight>
4+
{
5+
public Maybe<TLeft> Left { get; }
6+
public Maybe<TRight> Right { get; }
7+
8+
public Either(TLeft value)
9+
{
10+
Left = Maybe<TLeft>.Of(value);
11+
Right = Maybe<TRight>.Empty;
12+
}
13+
14+
public Either(TRight value)
15+
{
16+
Left = Maybe<TLeft>.Empty;
17+
Right = Maybe<TRight>.Of(value);
18+
}
19+
20+
public Either(Maybe<TLeft> left, Maybe<TRight> right)
21+
{
22+
if (!left.IsPresent && !right.IsPresent)
23+
throw new ArgumentOutOfRangeException(nameof(right));
24+
25+
Left = left;
26+
Right = right;
27+
}
28+
29+
public TMapped Map<TMapped>(
30+
Func<TLeft, TMapped> mapLeft,
31+
Func<TRight, TMapped> mapRight
32+
)
33+
{
34+
if (Left.IsPresent)
35+
return mapLeft(Left.GetOrThrow());
36+
37+
if (Right.IsPresent)
38+
return mapRight(Right.GetOrThrow());
39+
40+
throw new Exception("That should never happen!");
41+
}
42+
43+
public void Switch(
44+
Action<TLeft> onLeft,
45+
Action<TRight> onRight
46+
)
47+
{
48+
if (Left.IsPresent)
49+
{
50+
onLeft(Left.GetOrThrow());
51+
return;
52+
}
53+
54+
if (Right.IsPresent)
55+
{
56+
onRight(Right.GetOrThrow());
57+
return;
58+
}
59+
60+
throw new Exception("That should never happen!");
61+
}
62+
}
63+
64+
public static class EitherExtensions
65+
{
66+
public static (TLeft? Left, TRight? Right) AssertAnyDefined<TLeft, TRight>(
67+
this (TLeft? Left, TRight? Right) value
68+
)
69+
{
70+
if (value.Left == null && value.Right == null)
71+
throw new ArgumentOutOfRangeException(nameof(value), "One of values needs to be set");
72+
73+
return value;
74+
}
75+
76+
public static TMapped Map<TLeft, TRight, TMapped>(
77+
this (TLeft? Left, TRight? Right) value,
78+
Func<TLeft, TMapped> mapLeft,
79+
Func<TRight, TMapped> mapRight
80+
)
81+
where TLeft: struct
82+
where TRight: struct
83+
{
84+
var (left, right) = value.AssertAnyDefined();
85+
86+
if (left.HasValue)
87+
return mapLeft(left.Value);
88+
89+
if (right.HasValue)
90+
return mapRight(right.Value);
91+
92+
throw new Exception("That should never happen!");
93+
}
94+
95+
public static TMapped Map<TLeft, TRight, TMapped>(
96+
this (TLeft? Left, TRight? Right) value,
97+
Func<TLeft, TMapped> mapT1,
98+
Func<TRight, TMapped> mapT2
99+
)
100+
{
101+
value.AssertAnyDefined();
102+
103+
var either = value.Left != null
104+
? new Either<TLeft, TRight>(value.Left!)
105+
: new Either<TLeft, TRight>(value.Right!);
106+
107+
return either.Map(mapT1, mapT2);
108+
}
109+
110+
public static void Switch<TLeft, TRight>(
111+
this (TLeft? Left, TRight? Right) value,
112+
Action<TLeft> onT1,
113+
Action<TRight> onT2
114+
)
115+
{
116+
value.AssertAnyDefined();
117+
118+
var either = value.Left != null
119+
? new Either<TLeft, TRight>(value.Left!)
120+
: new Either<TLeft, TRight>(value.Right!);
121+
122+
either.Switch(onT1, onT2);
123+
}
124+
125+
public static (TLeft?, TRight?) Either<TLeft, TRight>(
126+
TLeft? left = default
127+
) => (left, default);
128+
129+
public static (TLeft?, TRight?) Either<TLeft, TRight>(
130+
TRight? right = default
131+
) => (default, right);
132+
}

0 commit comments

Comments
 (0)