-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
#7 continue to implement the category saga
- Loading branch information
1 parent
0ce71d3
commit 2514ff5
Showing
9 changed files
with
119 additions
and
68 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
123 changes: 90 additions & 33 deletions
123
src/Cik.Magazine.CategoryService/Sagas/CategoryProcessManager.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,94 +1,151 @@ | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using Akka; | ||
| using Akka.Event; | ||
| using Akka.Persistence; | ||
| using Akka.Persistence.Fsm; | ||
| using Cik.Magazine.Shared; | ||
| using Cik.Magazine.Shared.Messages.Category; | ||
| using Status = Cik.Magazine.Shared.Messages.Category.Status; | ||
| using MongoDB.Bson.Serialization; | ||
|
|
||
| namespace Cik.Magazine.CategoryService.Sagas | ||
| { | ||
| public class CategoryData | ||
| { | ||
| public Guid Id { get; set; } | ||
| public Status Status { get; set; } | ||
| public Status Status { get; set; } | ||
| } | ||
|
|
||
| public class CategoryProcessManager : PersistentFSM<Status, CategoryData, Event> | ||
| /// <summary> | ||
| /// CategoryProcessManager workflow for this case as | ||
| /// 1. [CategoryAggreation] submits CreateCategory command to [CategoryProcessManager] (status is Draft) | ||
| /// 2. [CategoryProcessManager] sends an email to [Admin] for an approval process (status is WaitingForApproval) | ||
| /// 2.1 This process will run periodic if [Admin] doesn't do the approval process (Quatz) | ||
| /// 3. [Admin] knows about it, then submits the ApproveCategory command to the [CategoryAggreation] | ||
| /// 4. [CategoryAggreation] updates itself status to Published and tells [CategoryProcessManager] about it | ||
| /// 5. [CategoryProcessManager] sends an email to notify with [Admin], and triggers CategoryApproved event for | ||
| /// persistence it into the storage | ||
| /// </summary> | ||
| public class CategoryProcessManager : PersistentFSM<Status, List<CategoryData>, Event> | ||
| { | ||
| private readonly List<CategoryData> _data = new List<CategoryData>(); | ||
| private readonly Guid _id; | ||
| private readonly CategoryData _data = new CategoryData(); | ||
| private readonly ILoggingAdapter _log; | ||
| private long LastSnapshottedVersion { get; set; } | ||
|
|
||
| public CategoryProcessManager(Guid id) | ||
| { | ||
| _id = id; | ||
| _log = Context.GetLogger(); | ||
|
|
||
| StartWith(Status.Reviewing, _data); | ||
| When(Status.Reviewing, (e, state) => | ||
| if (!BsonClassMap.IsClassMapRegistered(typeof(StateChangeEvent))) | ||
| BsonClassMap.RegisterClassMap<StateChangeEvent>(); | ||
|
|
||
| StartWith(Status.Draft, _data); | ||
|
|
||
| When(Status.Draft, (e, state) => | ||
| { | ||
| if (e.FsmEvent is CategoryCreated) | ||
| if (e.FsmEvent is CreateCategory) | ||
| { | ||
| var oldEvent = (CategoryCreated)e.FsmEvent; | ||
| return GoTo(Status.Published) | ||
| .Applying(new CategoryStatusUpdated(oldEvent.AggregateId, Status.Published)); | ||
| var oldEvent = (CreateCategory) e.FsmEvent; | ||
| return GoTo(Status.WaitingForApproval) | ||
| .Applying(new CategoryStatusUpdated(oldEvent.AggregateId, Status.WaitingForApproval)); | ||
| } | ||
|
|
||
| return state; | ||
| }); | ||
| When(Status.Published, (e, state) => | ||
|
|
||
| When(Status.Draft, (e, state) => | ||
| { | ||
| // TODO: 1. handle e.FsmEvent is ApproveCategory | ||
| // TODO: 2. GoTo(Status.Published) | ||
| // TODO: 3. Applying(new CategoryApproved(oldEvent.AggregateId, Status.Published)) | ||
|
|
||
| return state; | ||
| }); | ||
|
|
||
| When(Status.Published, (e, state) => { return state; }); | ||
| } | ||
|
|
||
| public override string PersistenceId => $"category-process-manager-{_id}"; | ||
| private long LastSnapshottedVersion { get; set; } | ||
|
|
||
| protected override bool ReceiveCommand(object message) | ||
| { | ||
| return message.Match() | ||
| .With<IEvent>(@event => | ||
| { | ||
| Persist(@event, e => | ||
| { | ||
| }); | ||
| }).WasHandled; | ||
| } | ||
| public override string PersistenceId => $"category-process-manager-{_id}"; | ||
|
|
||
| protected override bool ReceiveRecover(object message) | ||
| { | ||
| return message.Match() | ||
| .With<CategoryStatusUpdated>(@event => | ||
| { | ||
| _data.Id = @event.AggregateId; | ||
| _data.Status = @event.Status; | ||
| if (!_data.Any()) | ||
| { | ||
| _data.Add(new CategoryData | ||
| { | ||
| Id = @event.AggregateId, | ||
| Status = @event.Status | ||
| }); | ||
| } | ||
| else | ||
| { | ||
| var temps = _data.ConvertAll(x => x); | ||
| foreach (var categoryData in temps) | ||
| if (categoryData.Id.Equals(@event.AggregateId)) | ||
| categoryData.Status = @event.Status; | ||
| else | ||
| _data.Add(new CategoryData | ||
| { | ||
| Id = @event.AggregateId, | ||
| Status = @event.Status | ||
| }); | ||
| } | ||
| }) | ||
| .With<RecoveryCompleted>(() => | ||
| { | ||
| _log.Debug("[PM] Recovered state to version {0}", LastSequenceNr); | ||
| OnRecoveryCompleted(); | ||
| }) | ||
| .With<SnapshotOffer>(offer => | ||
| { | ||
| LastSnapshottedVersion = offer.Metadata.SequenceNr; | ||
| }).WasHandled; | ||
| .With<SnapshotOffer>(offer => { LastSnapshottedVersion = offer.Metadata.SequenceNr; }).WasHandled; | ||
| } | ||
|
|
||
| protected override CategoryData ApplyEvent(Event e, CategoryData data) | ||
| protected override List<CategoryData> ApplyEvent(Event e, List<CategoryData> data) | ||
| { | ||
| if (e is CategoryStatusUpdated) | ||
| { | ||
| // TODO: send notification to sys-admin for approve the category | ||
| var evented = (CategoryStatusUpdated) e; | ||
| // TODO: refactor later | ||
| if (!_data.Any()) | ||
| { | ||
| _data.Add(new CategoryData | ||
| { | ||
| Id = evented.AggregateId, | ||
| Status = evented.Status | ||
| }); | ||
| } | ||
| else | ||
| { | ||
| var temps = _data.ConvertAll(x => x); | ||
| foreach (var categoryData in temps) | ||
| if (categoryData.Id.Equals(evented.AggregateId)) | ||
| categoryData.Status = evented.Status; | ||
| else | ||
| _data.Add(new CategoryData | ||
| { | ||
| Id = evented.AggregateId, | ||
| Status = evented.Status | ||
| }); | ||
| } | ||
|
|
||
| return data; | ||
| } | ||
| return data; | ||
| } | ||
|
|
||
| protected override void OnRecoveryCompleted() | ||
| { | ||
|
|
||
| foreach (var categoryData in _data) | ||
| if (categoryData.Status == Status.WaitingForApproval) | ||
| { | ||
| // TODO: [side-effects] send notification to sys-admin for approve the category | ||
| // TODO: e.g EmailActor.Tell(SendEmail()) | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,7 +2,8 @@ | |
| { | ||
| public enum Status | ||
| { | ||
| Reviewing, | ||
| Draft, | ||
| WaitingForApproval, | ||
| Published | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters