Skip to content

Commit 32bc034

Browse files
oskardudyczoskardudycz
committed
Added example implementation of Process manager
1 parent bb16894 commit 32bc034

9 files changed

Lines changed: 481 additions & 0 deletions

File tree

‎samples/distributed-processes/src/main/java/io/eventdriven/distributedprocesses/core/processes/AbstractProcessManager.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
package io.eventdriven.distributedprocesses.core.processes;
2+
3+
import java.util.LinkedList;
4+
import java.util.Queue;
5+
6+
public abstract class AbstractProcessManager<Id> implements ProcessManager<Id> {
7+
protected Id id;
8+
protected long version = -1;
9+
10+
private final Queue<Object> scheduledCommands = new LinkedList<>();
11+
12+
public Id id() {
13+
return id;
14+
}
15+
16+
public void incrementVersion() {
17+
version++;
18+
}
19+
20+
public Object[] pollScheduledCommands() {
21+
var scheduledCommands = this.scheduledCommands.toArray();
22+
23+
this.scheduledCommands.clear();
24+
25+
return scheduledCommands;
26+
}
27+
28+
protected void schedule(Object command) {
29+
scheduledCommands.add(command);
30+
}
31+
}

‎samples/distributed-processes/src/main/java/io/eventdriven/distributedprocesses/core/processes/ProcessManager.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package io.eventdriven.distributedprocesses.core.processes;
2+
3+
public interface ProcessManager<Id> {
4+
Id id();
5+
6+
Object[] pollScheduledCommands();
7+
}

‎samples/distributed-processes/src/main/java/io/eventdriven/distributedprocesses/hotelmanagement/processmanager/groupcheckout/GroupCheckoutCommand.java‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
package io.eventdriven.distributedprocesses.hotelmanagement.processmanager.groupcheckout;
2+
3+
import java.time.OffsetDateTime;
4+
import java.util.UUID;
5+
6+
public sealed interface GroupCheckoutCommand {
7+
record InitiateGroupCheckout(
8+
UUID groupCheckoutId,
9+
UUID clerkId,
10+
UUID[] guestStayAccountIds,
11+
OffsetDateTime now
12+
) implements GroupCheckoutCommand {
13+
}
14+
15+
record RecordGuestCheckoutsInitiation(
16+
UUID groupCheckoutId,
17+
UUID[] guestStayAccountIds,
18+
OffsetDateTime now
19+
) implements GroupCheckoutCommand {
20+
}
21+
22+
record RecordGuestCheckoutCompletion(
23+
UUID groupCheckoutId,
24+
UUID guestStayAccountId,
25+
OffsetDateTime completedAt
26+
) implements GroupCheckoutCommand {
27+
}
28+
29+
record RecordGuestCheckoutFailure(
30+
UUID groupCheckoutId,
31+
UUID guestStayAccountId,
32+
OffsetDateTime failedAt
33+
) implements GroupCheckoutCommand {
34+
}
35+
}

‎samples/distributed-processes/src/main/java/io/eventdriven/distributedprocesses/hotelmanagement/processmanager/groupcheckout/GroupCheckoutEvent.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package io.eventdriven.distributedprocesses.hotelmanagement.processmanager.groupcheckout;
2+
3+
import java.time.OffsetDateTime;
4+
import java.util.UUID;
5+
6+
public sealed interface GroupCheckoutEvent {
7+
record GroupCheckoutInitiated(
8+
UUID groupCheckoutId,
9+
UUID clerkId,
10+
UUID[] guestStayAccountIds,
11+
OffsetDateTime initiatedAt
12+
) implements GroupCheckoutEvent {
13+
}
14+
15+
record GuestCheckoutsStarted(
16+
UUID groupCheckoutId,
17+
UUID[] guestStayAccountIds,
18+
OffsetDateTime initiatedAt
19+
) implements GroupCheckoutEvent {
20+
}
21+
22+
record GuestCheckoutCompleted(
23+
UUID groupCheckoutId,
24+
UUID guestStayAccountId,
25+
OffsetDateTime completedAt
26+
) implements GroupCheckoutEvent {
27+
}
28+
29+
record GuestCheckoutFailed(
30+
UUID groupCheckoutId,
31+
UUID guestStayAccountId,
32+
OffsetDateTime failedAt
33+
) implements GroupCheckoutEvent {
34+
}
35+
36+
record GroupCheckoutCompleted(
37+
UUID groupCheckoutId,
38+
UUID[] completedCheckouts,
39+
OffsetDateTime completedAt
40+
) implements GroupCheckoutEvent {
41+
}
42+
43+
record GroupCheckoutFailed(
44+
UUID groupCheckoutId,
45+
UUID[] completedCheckouts,
46+
UUID[] failedCheckouts,
47+
OffsetDateTime failedAt
48+
) implements GroupCheckoutEvent {
49+
}
50+
}

‎samples/distributed-processes/src/main/java/io/eventdriven/distributedprocesses/hotelmanagement/processmanager/groupcheckout/GroupCheckoutProcessManager.java‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
package io.eventdriven.distributedprocesses.hotelmanagement.processmanager.groupcheckout;
2+
3+
import io.eventdriven.distributedprocesses.core.processes.AbstractProcessManager;
4+
import io.eventdriven.distributedprocesses.hotelmanagement.processmanager.gueststayaccount.GuestStayAccountEvent;
5+
6+
import java.time.OffsetDateTime;
7+
import java.util.Arrays;
8+
import java.util.Map;
9+
import java.util.UUID;
10+
11+
import static io.eventdriven.distributedprocesses.core.collections.CollectionsExtensions.valueIs;
12+
import static io.eventdriven.distributedprocesses.hotelmanagement.processmanager.groupcheckout.GroupCheckoutEvent.GroupCheckoutInitiated;
13+
import static io.eventdriven.distributedprocesses.hotelmanagement.processmanager.gueststayaccount.GuestStayAccountCommand.CheckOutGuest;
14+
import static java.util.stream.Collectors.toMap;
15+
16+
public class GroupCheckoutProcessManager extends AbstractProcessManager<UUID> {
17+
private Map<UUID, CheckoutStatus> guestStayCheckouts;
18+
19+
private CheckoutStatus status;
20+
21+
private OffsetDateTime completedAt;
22+
private OffsetDateTime failedAt;
23+
24+
private GroupCheckoutProcessManager(
25+
UUID id,
26+
Map<UUID, CheckoutStatus> guestStayCheckouts,
27+
CheckoutStatus status
28+
) {
29+
this.id = id;
30+
this.guestStayCheckouts = guestStayCheckouts;
31+
this.status = status;
32+
}
33+
34+
enum CheckoutStatus {
35+
Initiated,
36+
InProgress,
37+
Completed,
38+
Failed
39+
}
40+
41+
public static GroupCheckoutProcessManager initiate(
42+
UUID id,
43+
UUID[] guestStayAccountIds
44+
) {
45+
return new GroupCheckoutProcessManager(id,
46+
Arrays.stream(guestStayAccountIds)
47+
.collect(toMap(key -> key, value -> CheckoutStatus.Initiated)),
48+
CheckoutStatus.Initiated
49+
);
50+
}
51+
52+
public void on(GroupCheckoutInitiated checkoutInitiated) {
53+
for (var guestAccountId : checkoutInitiated.guestStayAccountIds()) {
54+
schedule(
55+
new CheckOutGuest(guestAccountId, checkoutInitiated.groupCheckoutId(), checkoutInitiated.initiatedAt())
56+
);
57+
}
58+
59+
for (var guestStayAccountId : checkoutInitiated.guestStayAccountIds()) {
60+
if (valueIs(guestStayCheckouts, guestStayAccountId, CheckoutStatus.Initiated))
61+
guestStayCheckouts.replace(guestStayAccountId, CheckoutStatus.InProgress);
62+
}
63+
}
64+
65+
public void on(GuestStayAccountEvent.GuestCheckedOut checkoutCompleted) {
66+
if (checkoutCompleted.groupCheckoutId() == null
67+
|| status == CheckoutStatus.Completed
68+
|| status == CheckoutStatus.Failed
69+
)
70+
return;
71+
72+
var guestStayAccountId = checkoutCompleted.guestStayAccountId();
73+
74+
if (valueIs(guestStayCheckouts, guestStayAccountId, CheckoutStatus.Initiated, CheckoutStatus.InProgress))
75+
guestStayCheckouts.replace(guestStayAccountId, CheckoutStatus.Completed);
76+
77+
tryFinishCheckout(checkoutCompleted.completedAt());
78+
}
79+
80+
public void on(GuestStayAccountEvent.GuestCheckoutFailed checkoutFailed) {
81+
if (checkoutFailed.groupCheckoutId() == null
82+
|| status == CheckoutStatus.Completed
83+
|| status == CheckoutStatus.Failed
84+
)
85+
return;
86+
87+
var guestStayAccountId = checkoutFailed.guestStayAccountId();
88+
89+
if (valueIs(guestStayCheckouts, guestStayAccountId, CheckoutStatus.Initiated, CheckoutStatus.InProgress))
90+
guestStayCheckouts.replace(guestStayAccountId, CheckoutStatus.Completed);
91+
92+
tryFinishCheckout(checkoutFailed.failedAt());
93+
}
94+
95+
private void tryFinishCheckout(OffsetDateTime now) {
96+
if (areAnyOngoingCheckouts())
97+
return;
98+
99+
if(areAnyFailedCheckouts())
100+
{
101+
status = CheckoutStatus.Failed;
102+
failedAt = now;
103+
}
104+
else {
105+
status = CheckoutStatus.Completed;
106+
completedAt = now;
107+
}
108+
}
109+
110+
private boolean areAnyOngoingCheckouts() {
111+
return guestStayCheckouts.values().stream()
112+
.anyMatch(status -> status == CheckoutStatus.Initiated || status == CheckoutStatus.InProgress);
113+
}
114+
115+
private boolean areAnyFailedCheckouts() {
116+
return guestStayCheckouts.values().stream()
117+
.anyMatch(status -> status == CheckoutStatus.Failed);
118+
}
119+
}

‎samples/distributed-processes/src/main/java/io/eventdriven/distributedprocesses/hotelmanagement/processmanager/gueststayaccount/GuestStayAccount.java‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
package io.eventdriven.distributedprocesses.hotelmanagement.processmanager.gueststayaccount;
2+
3+
import io.eventdriven.distributedprocesses.core.aggregates.AbstractAggregate;
4+
import org.springframework.lang.Nullable;
5+
6+
import java.time.OffsetDateTime;
7+
import java.util.UUID;
8+
9+
import static io.eventdriven.distributedprocesses.hotelmanagement.processmanager.gueststayaccount.GuestStayAccountEvent.*;
10+
11+
public class GuestStayAccount extends AbstractAggregate<GuestStayAccountEvent, UUID> {
12+
private enum Status {
13+
Open,
14+
Checkout
15+
}
16+
17+
private Status status;
18+
private double balance;
19+
20+
public static GuestStayAccount open(UUID guestStayAccountId, OffsetDateTime openedAt) {
21+
return new GuestStayAccount(
22+
guestStayAccountId,
23+
openedAt
24+
);
25+
}
26+
27+
private GuestStayAccount(
28+
UUID guestStayAccountId,
29+
OffsetDateTime openedAt
30+
) {
31+
enqueue(new GuestCheckedIn(guestStayAccountId, openedAt));
32+
}
33+
34+
public void recordCharge(double amount, OffsetDateTime now) {
35+
if (status != Status.Open)
36+
throw new RuntimeException("Cannot record charge if status is other than Open");
37+
38+
enqueue(new ChargeRecorded(id(), amount, now));
39+
}
40+
41+
public void recordPayment(double amount, OffsetDateTime now) {
42+
if (status != Status.Open)
43+
throw new RuntimeException("Cannot record payment if status is other than Open");
44+
45+
enqueue(new PaymentRecorded(id(), amount, now));
46+
}
47+
48+
public void checkout(@Nullable UUID groupCheckoutId, OffsetDateTime now) {
49+
if (status != Status.Open || balance != 0) {
50+
enqueue(
51+
new GuestCheckoutFailed(id(),
52+
balance != 0 ? GuestCheckoutFailed.Reason.BalanceNotSettled : GuestCheckoutFailed.Reason.InvalidState,
53+
groupCheckoutId,
54+
OffsetDateTime.now()
55+
)
56+
);
57+
return;
58+
}
59+
enqueue(new GuestCheckedOut(id(), groupCheckoutId, now));
60+
}
61+
62+
@Override
63+
public void when(GuestStayAccountEvent event) {
64+
switch (event) {
65+
case GuestCheckedIn opened -> {
66+
id = opened.guestStayAccountId();
67+
balance = 0;
68+
status = Status.Open;
69+
}
70+
case ChargeRecorded chargeRecorded -> balance -= chargeRecorded.amount();
71+
case PaymentRecorded paymentRecorded ->
72+
balance += paymentRecorded.amount();
73+
case GuestCheckedOut ignored ->
74+
status = Status.Checkout;
75+
case GuestCheckoutFailed ignored -> {
76+
}
77+
}
78+
}
79+
}
80+

‎samples/distributed-processes/src/main/java/io/eventdriven/distributedprocesses/hotelmanagement/processmanager/gueststayaccount/GuestStayAccountCommand.java‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
package io.eventdriven.distributedprocesses.hotelmanagement.processmanager.gueststayaccount;
2+
3+
import io.eventdriven.distributedprocesses.core.http.ETag;
4+
import org.springframework.lang.Nullable;
5+
6+
import java.time.OffsetDateTime;
7+
import java.util.UUID;
8+
9+
public sealed interface GuestStayAccountCommand {
10+
record CheckInGuest(
11+
UUID guestStayAccountId,
12+
ETag expectedRevision,
13+
OffsetDateTime now
14+
) implements GuestStayAccountCommand {
15+
}
16+
17+
record RecordCharge(
18+
UUID guestStayAccountId,
19+
double amount,
20+
ETag expectedRevision,
21+
OffsetDateTime now
22+
) implements GuestStayAccountCommand {
23+
}
24+
25+
record RecordPayment(
26+
UUID guestStayAccountId,
27+
double amount,
28+
ETag expectedRevision,
29+
OffsetDateTime now
30+
) implements GuestStayAccountCommand {
31+
}
32+
33+
record CheckOutGuest(
34+
UUID guestStayAccountId,
35+
@Nullable
36+
UUID groupCheckoutId,
37+
OffsetDateTime now
38+
) implements GuestStayAccountCommand {
39+
}
40+
}

0 commit comments

Comments
 (0)