You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2023/04/22 00:26:48 UTC
[james-project] branch master updated: JAMES-3900 Snapshots for polled updates (#1533)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
The following commit(s) were added to refs/heads/master by this push:
new 334db08121 JAMES-3900 Snapshots for polled updates (#1533)
334db08121 is described below
commit 334db08121e42b5b04ee587858f05482a0f27a9c
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Sat Apr 22 07:26:41 2023 +0700
JAMES-3900 Snapshots for polled updates (#1533)
They fully capture the state of the aggregate.
This prevents loading the full history for looong
running tasks and prevents timeouts upon polling updates.
---
.../task/eventsourcing/DecisionProjection.scala | 9 +++++++--
.../apache/james/task/eventsourcing/Events.scala | 4 +++-
.../james/task/eventsourcing/TaskAggregate.scala | 21 ++++++++++++++++-----
3 files changed, 26 insertions(+), 8 deletions(-)
diff --git a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/DecisionProjection.scala b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/DecisionProjection.scala
index ce2f6d2208..d42fb9a4ca 100644
--- a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/DecisionProjection.scala
+++ b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/DecisionProjection.scala
@@ -32,7 +32,7 @@ case class DecisionProjection(status: Status, latestUpdateAdditionalInformationU
case event: Cancelled => DecisionProjection(Status.CANCELLED, event.additionalInformation.map(_.timestamp))
case event: Completed => DecisionProjection(Status.COMPLETED, event.additionalInformation.map(_.timestamp))
case event: Failed => DecisionProjection(Status.FAILED, event.additionalInformation.map(_.timestamp))
- case event: AdditionalInformationUpdated => DecisionProjection(status, Some(event.additionalInformation.timestamp))
+ case event: AdditionalInformationUpdated => DecisionProjection(Status.IN_PROGRESS, Some(event.additionalInformation.timestamp))
}
}
@@ -41,6 +41,11 @@ case class DecisionProjection(status: Status, latestUpdateAdditionalInformationU
}
object DecisionProjection {
- def initial(created : Created): DecisionProjection = DecisionProjection(Status.WAITING, None)
+ def initial(taskEvent : TaskEvent): DecisionProjection = {
+ taskEvent match {
+ case _: Created => DecisionProjection(Status.WAITING, None)
+ case updated: AdditionalInformationUpdated => DecisionProjection(Status.IN_PROGRESS, Some(updated.additionalInformation.timestamp))
+ }
+ }
}
diff --git a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/Events.scala b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
index 1f5ff6fbd6..45e5683f38 100644
--- a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
+++ b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
@@ -33,7 +33,9 @@ case class Created(aggregateId: TaskAggregateId, override val eventId: EventId,
case class Started(aggregateId: TaskAggregateId, override val eventId: EventId, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
-case class AdditionalInformationUpdated(aggregateId: TaskAggregateId, override val eventId: EventId, additionalInformation: AdditionalInformation) extends TaskEvent(aggregateId, eventId)
+case class AdditionalInformationUpdated(aggregateId: TaskAggregateId, override val eventId: EventId, additionalInformation: AdditionalInformation) extends TaskEvent(aggregateId, eventId) {
+ override val isASnapshot = true
+}
case class CancelRequested(aggregateId: TaskAggregateId, override val eventId: EventId, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
diff --git a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
index 04c36bfff4..197b700e6e 100644
--- a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
+++ b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
@@ -25,12 +25,23 @@ import org.apache.james.task.TaskExecutionDetails.AdditionalInformation
import org.apache.james.task.TaskManager.Status
import org.apache.james.task.{Hostname, Task}
-class TaskAggregate private(val aggregateId: TaskAggregateId, private val history: History) {
+import scala.util.{Failure, Success}
- val initialEvent = history.getEvents.headOption match {
- case Some(created @ Created(_, _, _, _)) => created
- case _ => throw new IllegalArgumentException("History must start with Created event")
- }
+class TaskAggregate private(val aggregateId: TaskAggregateId, private val history: History) {
+ private val initialEvent: TaskEvent = history.getEvents.headOption
+ .map(e => Success(initialEvent(e)))
+ .getOrElse(Failure(new IllegalArgumentException("History must start with Created event")))
+ .get
+
+ private def initialEvent(event: Event) =
+ event match {
+ case created@Created(_, _, _, _) if created.eventId.equals(EventId.first) => created
+ case any: TaskEvent => if (any.eventId.equals(EventId.first)) {
+ throw new IllegalArgumentException("History must start with Created event")
+ } else {
+ any
+ }
+ }
private val currentDecisionProjection: DecisionProjection = history
.getEvents
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org