Skip to content

Commit 9be3d3b

Browse files
oskardudyczoskardudycz
committed
Added examples of choreography
1 parent 16ddcd6 commit 9be3d3b

File tree

8 files changed

+169
-68
lines changed

8 files changed

+169
-68
lines changed

‎Core.Marten/Extensions/DocumentSessionExtensions.cs‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
using Core.ProcessManagers;
12
using Core.Structures;
23
using Marten;
34

@@ -16,6 +17,17 @@ CancellationToken ct
1617
return documentSession.SaveChangesAsync(token: ct);
1718
}
1819

20+
public static Task Add<T>(
21+
this IDocumentSession documentSession,
22+
string id,
23+
object @event,
24+
CancellationToken ct
25+
) where T : class
26+
{
27+
documentSession.Events.StartStream<T>(id, @event);
28+
return documentSession.SaveChangesAsync(token: ct);
29+
}
30+
1931
public static Task GetAndUpdate<T>(
2032
this IDocumentSession documentSession,
2133
Guid id, int version,
@@ -25,6 +37,55 @@ CancellationToken ct
2537
documentSession.Events.WriteToAggregate<T>(id, version, stream =>
2638
stream.AppendOne(handle(stream.Aggregate)), ct);
2739

40+
public static Task GetAndUpdate<T>(
41+
this IDocumentSession documentSession,
42+
string id,
43+
int version,
44+
Func<T, IEnumerable<EventOrCommand>> handle,
45+
CancellationToken ct
46+
) where T : class =>
47+
documentSession.Events.WriteToAggregate<T>(id, version, stream =>
48+
{
49+
var messages = handle(stream.Aggregate);
50+
51+
foreach (var message in messages)
52+
{
53+
message.Switch(
54+
stream.AppendOne,
55+
command => documentSession.Events.Append($"commands-{id}", command)
56+
);
57+
}
58+
}, ct);
59+
60+
61+
public static Task GetAndUpdate<T>(
62+
this IDocumentSession documentSession,
63+
string id,
64+
Func<T, IEnumerable<EventOrCommand>> handle,
65+
CancellationToken ct
66+
) where T : class =>
67+
documentSession.Events.WriteToAggregate<T>(id, stream =>
68+
{
69+
var messages = handle(stream.Aggregate);
70+
71+
foreach (var message in messages)
72+
{
73+
message.Switch(
74+
stream.AppendOne,
75+
command => documentSession.Events.Append($"commands-{id}", command)
76+
);
77+
}
78+
}, ct);
79+
80+
public static Task GetAndUpdate<T>(
81+
this IDocumentSession documentSession,
82+
string id,
83+
Func<T, IEnumerable<object>> handle,
84+
CancellationToken ct
85+
) where T : class =>
86+
documentSession.Events.WriteToAggregate<T>(id,
87+
stream => stream.AppendMany(handle(stream.Aggregate)), ct);
88+
2889
public static Task GetAndUpdate<T>(
2990
this IDocumentSession documentSession,
3091
Guid id,

‎Core/ProcessManagers/IProcessManager.cs‎

Lines changed: 0 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -14,17 +14,3 @@ public interface IProcessManager<out T>: IProjection
1414

1515
EventOrCommand[] DequeuePendingMessages();
1616
}
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: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
using Core.Structures;
2+
13
namespace Core.ProcessManagers;
24

35
public abstract class ProcessManager: ProcessManager<Guid>, IProcessManager

‎Core/Structures/EventOrCommand.cs‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
namespace Core.Structures;
2+
3+
public class EventOrCommand: Either<object, object>
4+
{
5+
public static EventOrCommand Event(object @event) =>
6+
new(Maybe<object>.Of(@event), Maybe<object>.Empty);
7+
8+
public static IEnumerable<EventOrCommand> Events(params object[] events) =>
9+
events.Select(Event);
10+
11+
public static IEnumerable<EventOrCommand> Events(IEnumerable<object> events) =>
12+
events.Select(Event);
13+
14+
public static EventOrCommand Command(object @event) =>
15+
new(Maybe<object>.Empty, Maybe<object>.Of(@event));
16+
17+
private EventOrCommand(Maybe<object> left, Maybe<object> right): base(left, right)
18+
{
19+
}
20+
}

‎Sample/HotelManagement/HotelManagement/Choreography/GroupCheckouts/GroupCheckout.cs‎

Lines changed: 34 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
namespace HotelManagement.Choreography.GroupCheckouts;
55

66
public record GroupCheckoutInitiated(
7-
Guid GroupCheckoutId,
7+
Guid GroupCheckOutId,
88
Guid ClerkId,
99
Guid[] GuestStayIds,
1010
DateTimeOffset InitiatedAt
@@ -55,48 +55,50 @@ DateTimeOffset initiatedAt
5555
) =>
5656
new GroupCheckoutInitiated(groupCheckoutId, clerkId, guestStayIds, initiatedAt);
5757

58-
public Maybe<GuestCheckoutsInitiated> RecordGuestCheckoutsInitiation(
58+
public GuestCheckoutsInitiated? RecordGuestCheckoutsInitiation(
5959
Guid[] initiatedGuestStayIds,
6060
DateTimeOffset now
61-
) =>
62-
Maybe.If(
63-
Status == CheckoutStatus.Initiated,
64-
() => new GuestCheckoutsInitiated(Id, initiatedGuestStayIds, now)
65-
);
61+
)
62+
{
63+
if (Status == CheckoutStatus.Initiated)
64+
return null;
6665

67-
public Maybe<object[]> RecordGuestCheckoutCompletion(
66+
return new GuestCheckoutsInitiated(Id, initiatedGuestStayIds, now);
67+
}
68+
69+
public object[] RecordGuestCheckoutCompletion(
6870
Guid guestStayId,
6971
DateTimeOffset now
70-
) =>
71-
Maybe.If(
72-
Status == CheckoutStatus.Initiated && GuestStayCheckouts[guestStayId] != CheckoutStatus.Completed,
73-
() =>
74-
{
75-
var guestCheckoutCompleted = new GuestCheckoutCompleted(Id, guestStayId, now);
72+
)
73+
{
74+
if (Status == CheckoutStatus.Initiated && GuestStayCheckouts[guestStayId] != CheckoutStatus.Completed)
75+
return Array.Empty<object>();
76+
77+
var guestCheckoutCompleted = new GuestCheckoutCompleted(Id, guestStayId, now);
7678

77-
var guestStayCheckouts = GuestStayCheckouts.With(guestStayId, CheckoutStatus.Completed);
79+
var guestStayCheckouts = GuestStayCheckouts.With(guestStayId, CheckoutStatus.Completed);
7880

79-
return AreAnyOngoingCheckouts(guestStayCheckouts)
80-
? new object[] { guestCheckoutCompleted }
81-
: new[] { guestCheckoutCompleted, Finalize(guestStayCheckouts, now) };
82-
});
81+
return AreAnyOngoingCheckouts(guestStayCheckouts)
82+
? new object[] { guestCheckoutCompleted }
83+
: new[] { guestCheckoutCompleted, Finalize(guestStayCheckouts, now) };
84+
}
8385

84-
public Maybe<object[]> RecordGuestCheckoutFailure(
86+
public object[] RecordGuestCheckoutFailure(
8587
Guid guestStayId,
8688
DateTimeOffset now
87-
) =>
88-
Maybe.If(
89-
Status == CheckoutStatus.Initiated && GuestStayCheckouts[guestStayId] != CheckoutStatus.Failed,
90-
() =>
91-
{
92-
var guestCheckoutFailed = new GuestCheckoutFailed(Id, guestStayId, now);
89+
)
90+
{
91+
if(Status == CheckoutStatus.Initiated && GuestStayCheckouts[guestStayId] != CheckoutStatus.Failed)
92+
return Array.Empty<object>();
93+
94+
var guestCheckoutFailed = new GuestCheckoutFailed(Id, guestStayId, now);
9395

94-
var guestStayCheckouts = GuestStayCheckouts.With(guestStayId, CheckoutStatus.Failed);
96+
var guestStayCheckouts = GuestStayCheckouts.With(guestStayId, CheckoutStatus.Failed);
9597

96-
return AreAnyOngoingCheckouts(guestStayCheckouts)
97-
? new object[] { guestCheckoutFailed }
98-
: new[] { guestCheckoutFailed, Finalize(guestStayCheckouts, now) };
99-
});
98+
return AreAnyOngoingCheckouts(guestStayCheckouts)
99+
? new object[] { guestCheckoutFailed }
100+
: new[] { guestCheckoutFailed, Finalize(guestStayCheckouts, now) };
101+
}
100102

101103
private object Finalize(Dictionary<Guid, CheckoutStatus> guestStayCheckouts, DateTimeOffset now) =>
102104
!AreAnyFailedCheckouts(guestStayCheckouts)
@@ -128,7 +130,7 @@ private static Guid[] CheckoutsWith(Dictionary<Guid, CheckoutStatus> guestStayCh
128130

129131
public static GroupCheckout Create(GroupCheckoutInitiated @event) =>
130132
new GroupCheckout(
131-
@event.GroupCheckoutId,
133+
@event.GroupCheckOutId,
132134
@event.GuestStayIds.ToDictionary(id => id, _ => CheckoutStatus.Pending)
133135
);
134136

‎Sample/HotelManagement/HotelManagement/Choreography/GroupCheckouts/GroupCheckoutDomainService.cs‎

Lines changed: 49 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
using Core.Commands;
2+
using Core.Events;
23
using Core.Marten.Extensions;
4+
using Core.Structures;
5+
using HotelManagement.Choreography.GuestStayAccounts;
36
using Marten;
7+
using static Core.Structures.EventOrCommand;
48

59
namespace HotelManagement.Choreography.GroupCheckouts;
610

@@ -29,18 +33,22 @@ DateTimeOffset FailedAt
2933

3034
public class GuestStayDomainService:
3135
ICommandHandler<InitiateGroupCheckout>,
32-
ICommandHandler<RecordGuestCheckoutsInitiation>,
33-
ICommandHandler<RecordGuestCheckoutCompletion>,
34-
ICommandHandler<RecordGuestCheckoutFailure>
36+
IEventHandler<GroupCheckoutInitiated>,
37+
IEventHandler<GuestStayAccounts.GuestCheckedOut>,
38+
IEventHandler<GuestStayAccounts.GuestCheckoutFailed>
3539
{
3640
private readonly IDocumentSession documentSession;
41+
private readonly IAsyncCommandBus commandBus;
3742

38-
public GuestStayDomainService(IDocumentSession documentSession) =>
43+
public GuestStayDomainService(IDocumentSession documentSession, IAsyncCommandBus commandBus)
44+
{
3945
this.documentSession = documentSession;
46+
this.commandBus = commandBus;
47+
}
4048

4149
public Task Handle(InitiateGroupCheckout command, CancellationToken ct) =>
4250
documentSession.Add<GroupCheckout>(
43-
command.GroupCheckoutId,
51+
command.GroupCheckoutId.ToString(),
4452
GroupCheckout.Initiate(
4553
command.GroupCheckoutId,
4654
command.ClerkId,
@@ -50,24 +58,46 @@ public Task Handle(InitiateGroupCheckout command, CancellationToken ct) =>
5058
ct
5159
);
5260

53-
public Task Handle(RecordGuestCheckoutsInitiation command, CancellationToken ct) =>
54-
documentSession.GetAndUpdate(
55-
command.GroupCheckoutId,
56-
(GroupCheckout state) => state.RecordGuestCheckoutsInitiation(command.InitiatedGuestStayIds, DateTimeOffset.UtcNow),
57-
ct
58-
);
61+
public Task Handle(GroupCheckoutInitiated @event, CancellationToken ct)
62+
{
63+
IEnumerable<EventOrCommand> OnInitiated(GroupCheckout groupCheckout)
64+
{
65+
var result = groupCheckout.RecordGuestCheckoutsInitiation(@event.GuestStayIds, @event.InitiatedAt);
66+
67+
if (result is not null)
68+
{
69+
foreach (var guestAccountId in @event.GuestStayIds)
70+
{
71+
yield return Command(new CheckOutGuest(guestAccountId, @event.GroupCheckOutId));
72+
}
73+
74+
yield return Event(result);
75+
}
76+
}
5977

60-
public Task Handle(RecordGuestCheckoutCompletion command, CancellationToken ct) =>
61-
documentSession.GetAndUpdate(
62-
command.GuestStayId,
63-
(GroupCheckout state) => state.RecordGuestCheckoutCompletion(command.GuestStayId, command.CompletedAt),
78+
return documentSession.GetAndUpdate<GroupCheckout>(@event.GroupCheckOutId.ToString(), OnInitiated, ct);
79+
}
80+
81+
public Task Handle(GuestStayAccounts.GuestCheckedOut @event, CancellationToken ct)
82+
{
83+
if (!@event.GroupCheckOutId.HasValue)
84+
return Task.CompletedTask;
85+
86+
return documentSession.GetAndUpdate<GroupCheckout>(@event.GroupCheckOutId.Value.ToString(),
87+
groupCheckout => groupCheckout.RecordGuestCheckoutCompletion(@event.GuestStayId, @event.CheckedOutAt),
6488
ct
6589
);
90+
}
91+
92+
public Task Handle(Choreography.GuestStayAccounts.GuestCheckoutFailed @event, CancellationToken ct)
93+
{
94+
if (!@event.GroupCheckOutId.HasValue)
95+
return Task.CompletedTask;
6696

67-
public Task Handle(RecordGuestCheckoutFailure command, CancellationToken ct) =>
68-
documentSession.GetAndUpdate(
69-
command.GuestStayId,
70-
(GroupCheckout state) => state.RecordGuestCheckoutFailure(command.GuestStayId, command.FailedAt),
97+
return documentSession.GetAndUpdate<GroupCheckout>(@event.GroupCheckOutId.Value.ToString(),
98+
groupCheckout =>
99+
groupCheckout.RecordGuestCheckoutFailure(@event.GuestStayId, @event.FailedAt),
71100
ct
72101
);
102+
}
73103
}

‎Sample/HotelManagement/HotelManagement/Choreography/GroupCheckouts/GroupCheckoutSaga.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,14 @@ public async Task Handle(GroupCheckoutInitiated @event, CancellationToken ct)
1919
foreach (var guestAccountId in @event.GuestStayIds)
2020
{
2121
await commandBus.Schedule(
22-
new CheckOutGuest(guestAccountId, @event.GroupCheckoutId),
22+
new CheckOutGuest(guestAccountId, @event.GroupCheckOutId),
2323
ct
2424
);
2525
}
2626

2727
await commandBus.Schedule(
2828
new RecordGuestCheckoutsInitiation(
29-
@event.GroupCheckoutId,
29+
@event.GroupCheckOutId,
3030
@event.GuestStayIds
3131
),
3232
ct

‎Sample/HotelManagement/README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ It was modelled and explained in detail in the [Implementing Distributed Process
99
<a href="https://www.architecture-weekly.com/p/webinar-3-implementing-distributed" target="_blank"><img src="https://substackcdn.com/image/fetch/w_1920,h_1080,c_fill,f_auto,q_auto:good,fl_progressive:steep/https%3A%2F%2Fsubstack-video.s3.amazonaws.com%2Fvideo_upload%2Fpost%2F69413446%2F526b9100-7271-4482-99e7-9559416e9848%2Ftranscoded-00624.png" alt="How to deal with privacy and GDPR in Event-Sourced systems" width="640" border="10" /></a>
1010

1111
It shows how to:
12-
- orchestrate and coordinate a business workflow spanning across multiple aggregates using the [Saga pattern](https://event-driven.io/en/saga_process_manager_distributed_transactions/),
12+
- orchestrate and coordinate a business workflow spanning across multiple aggregates using the [Saga pattern](./HotelManagement/Sagas), [Choreography](./HotelManagement/Choreography) and [Process Managers](./HotelManagement/ProcessManagers),
1313
- handle distributed processing both for asynchronous command scheduling and event publishing,
1414
- getting at-least-once delivery guarantee,
1515
- implementing command store and outbox pattern on top of Marten and EventStoreDB,

0 commit comments

Comments
 (0)