james-server-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From matth...@apache.org
Subject [james-project] 03/04: JAMES-2873 Add Hostname into CancelRequested event
Date Mon, 02 Sep 2019 08:16:00 GMT
This is an automated email from the ASF dual-hosted git repository.

matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git

commit bc64574d21e7f703a2044744d79ce2a15ae45b0a
Author: Gautier DI FOLCO <gdifolco@linagora.com>
AuthorDate: Wed Aug 28 11:34:04 2019 +0200

    JAMES-2873 Add Hostname into CancelRequested event
---
 .../CassandraTaskExecutionDetailsProjectionDAO.scala    |  3 +++
 .../CassandraTaskExecutionDetailsProjectionModule.scala |  2 ++
 .../task/eventsourcing/distributed/TaskEventDTO.scala   |  7 ++++---
 .../distributed/TaskEventsSerializationTest.java        |  4 ++--
 .../java/org/apache/james/task/MemoryTaskManager.java   |  2 +-
 .../org/apache/james/task/TaskExecutionDetails.scala    | 17 +++++++++++++----
 .../james/task/eventsourcing/CommandHandlers.scala      |  7 ++++---
 .../task/eventsourcing/EventSourcingTaskManager.scala   |  4 ++--
 .../org/apache/james/task/eventsourcing/Events.scala    |  2 +-
 .../apache/james/task/eventsourcing/TaskAggregate.scala |  4 ++--
 .../eventsourcing/TaskExecutionDetailsProjection.scala  |  2 +-
 .../eventsourcing/EventSourcingTaskManagerTest.java     | 16 ++++++++++++++++
 12 files changed, 51 insertions(+), 19 deletions(-)

diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
index e7af442..f598aa5 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
@@ -47,6 +47,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
     .value(RAN_NODE, bindMarker(RAN_NODE))
     .value(COMPLETED_DATE, bindMarker(COMPLETED_DATE))
     .value(CANCELED_DATE, bindMarker(CANCELED_DATE))
+    .value(CANCEL_REQUESTED_NODE, bindMarker(CANCEL_REQUESTED_NODE))
     .value(FAILED_DATE, bindMarker(FAILED_DATE)))
 
   private val selectStatement = session.prepare(select().from(TABLE_NAME)
@@ -65,6 +66,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
       .setString(RAN_NODE, details.getRanNode.map[String](_.asString).orElse(null))
       .setUDTValue(COMPLETED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getCompletedDate).orElse(null))
       .setUDTValue(CANCELED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getCanceledDate).orElse(null))
+      .setString(CANCEL_REQUESTED_NODE, details.getCancelRequestedNode.map[String](_.asString).orElse(null))
       .setUDTValue(FAILED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getStartedDate).orElse(null)))
 
   def readDetails(taskId: TaskId): Mono[TaskExecutionDetails] = cassandraAsyncExecutor
@@ -85,6 +87,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
     ranNode = Optional.ofNullable(row.getString(RAN_NODE)).map(Hostname(_)),
     completedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(COMPLETED_DATE)),
     canceledDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(CANCELED_DATE)),
+    cancelRequestedNode = Optional.ofNullable(row.getString(CANCEL_REQUESTED_NODE)).map(Hostname(_)),
     failedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(FAILED_DATE)),
     additionalInformation = Optional.empty)
 }
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
index 1ee0554..4baaa0b 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
@@ -36,6 +36,7 @@ object CassandraTaskExecutionDetailsProjectionTable {
   val RAN_NODE: String = "ranNode"
   val COMPLETED_DATE: String = "completedDate"
   val CANCELED_DATE: String = "canceledDate"
+  val CANCEL_REQUESTED_NODE: String = "cancelRequestedNode"
   val FAILED_DATE: String = "failedDate"
 }
 
@@ -58,6 +59,7 @@ object CassandraTaskExecutionDetailsProjectionModule {
       .addColumn(CassandraTaskExecutionDetailsProjectionTable.RAN_NODE, text)
       .addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.COMPLETED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
       .addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.CANCELED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
+      .addColumn(CassandraTaskExecutionDetailsProjectionTable.CANCEL_REQUESTED_NODE, text)
       .addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.FAILED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME)))
     .build
 }
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
index 4e13454..8ff3093 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
@@ -60,14 +60,15 @@ object StartedDTO {
 
 case class CancelRequestedDTO(@JsonProperty("type") typeName: String,
                               @JsonProperty("aggregate") aggregateId: String,
-                              @JsonProperty("event") eventId: Int)
+                              @JsonProperty("event") eventId: Int,
+                              @JsonProperty("hostname") getHostname: String)
   extends TaskEventDTO(typeName, aggregateId, eventId) {
-  def toDomainObject: CancelRequested = CancelRequested(domainAggregateId, domainEventId)
+  def toDomainObject: CancelRequested = CancelRequested(domainAggregateId, domainEventId,
Hostname(getHostname))
 }
 
 object CancelRequestedDTO {
   def fromDomainObject(event: CancelRequested, typeName: String): CancelRequestedDTO =
-    CancelRequestedDTO(typeName, event.aggregateId.taskId.asString(), event.eventId.serialize())
+    CancelRequestedDTO(typeName, event.aggregateId.taskId.asString(), event.eventId.serialize(),
event.hostname.asString)
 }
 
 case class CompletedDTO(@JsonProperty("type") typeName: String,
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
index b41c762..eaf03ec 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
@@ -81,7 +81,7 @@ class TaskEventsSerializationTest {
         return Stream.of(
             Arguments.of(new Created(AGGREGATE_ID, EVENT_ID, TASK, HOSTNAME), "{\"task\":\"{\\\"type\\\":\\\"completed-task\\\"}\",\"type\":\"task-manager-created\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"hostname\":\"foo\"}\n"),
             Arguments.of(new Started(AGGREGATE_ID, EVENT_ID, HOSTNAME), "{\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-started\",\"hostname\":\"foo\"}"),
-            Arguments.of(new CancelRequested(AGGREGATE_ID, EVENT_ID), "{\"type\":\"task-manager-cancel-requested\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42}\n"),
+            Arguments.of(new CancelRequested(AGGREGATE_ID, EVENT_ID, HOSTNAME), "{\"type\":\"task-manager-cancel-requested\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"hostname\":\"foo\"}\n"),
             Arguments.of(new Completed(AGGREGATE_ID, EVENT_ID, Task.Result.COMPLETED), "{\"result\":\"COMPLETED\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-completed\"}"),
             Arguments.of(new Completed(AGGREGATE_ID, EVENT_ID, Task.Result.PARTIAL), "{\"result\":\"PARTIAL\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-completed\"}"),
             Arguments.of(new Failed(AGGREGATE_ID, EVENT_ID), "{\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-failed\"}"),
@@ -89,4 +89,4 @@ class TaskEventsSerializationTest {
         );
     }
 
-}
+}
\ No newline at end of file
diff --git a/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java b/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
index b065edd..90e88c0 100644
--- a/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
+++ b/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
@@ -133,7 +133,7 @@ public class MemoryTaskManager implements TaskManager {
     @Override
     public void cancel(TaskId id) {
         Optional.ofNullable(idToExecutionDetails.get(id)).ifPresent(details -> {
-                updateDetails(id).accept(TaskExecutionDetails::cancelRequested);
+                updateDetails(id).accept(taskExecutionDetails -> taskExecutionDetails.cancelRequested(hostname));
                 workQueue.cancel(id);
             }
         );
diff --git a/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala b/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
index 95b0d99..d08d879 100644
--- a/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
+++ b/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
@@ -44,6 +44,7 @@ class TaskExecutionDetails(val taskId: TaskId,
                            private val ranNode: Optional[Hostname] = Optional.empty(),
                            private val completedDate: Optional[ZonedDateTime] = Optional.empty(),
                            private val canceledDate: Optional[ZonedDateTime] = Optional.empty(),
+                           private val cancelRequestedNode: Optional[Hostname] = Optional.empty(),
                            private val failedDate: Optional[ZonedDateTime] = Optional.empty())
{
   def getTaskId: TaskId = taskId
 
@@ -65,6 +66,8 @@ class TaskExecutionDetails(val taskId: TaskId,
 
   def getCanceledDate: Optional[ZonedDateTime] = canceledDate
 
+  def getCancelRequestedNode: Optional[Hostname] = cancelRequestedNode
+
   def getFailedDate: Optional[ZonedDateTime] = failedDate
 
   def started(hostname: Hostname): TaskExecutionDetails = status match {
@@ -86,8 +89,8 @@ class TaskExecutionDetails(val taskId: TaskId,
   }
 
   def cancelRequested(hostname: Hostname): TaskExecutionDetails = status match {
-    case IN_PROGRESS => requestCancel
-    case WAITING => requestCancel
+    case IN_PROGRESS => requestCancel(hostname)
+    case WAITING => requestCancel(hostname)
     case _ => this
   }
 
@@ -113,12 +116,13 @@ class TaskExecutionDetails(val taskId: TaskId,
         Objects.equals(ranNode, that.ranNode) &&
         Objects.equals(completedDate, that.completedDate) &&
         Objects.equals(canceledDate, that.canceledDate) &&
+        Objects.equals(cancelRequestedNode, that.cancelRequestedNode) &&
         Objects.equals(failedDate, that.failedDate)
     case _ => false
   }
 
   override def hashCode(): Int =
-    Objects.hash(taskId, `type`, additionalInformation(), status, submittedDate, submittedNode,
startedDate, ranNode, completedDate, canceledDate, failedDate)
+    Objects.hash(taskId, `type`, additionalInformation(), status, submittedDate, submittedNode,
startedDate, ranNode, completedDate, canceledDate, cancelRequestedNode, failedDate)
 
   override def toString: String =
     MoreObjects.toStringHelper(this)
@@ -132,6 +136,7 @@ class TaskExecutionDetails(val taskId: TaskId,
       .add("ranNode", ranNode)
       .add("completedDate", completedDate)
       .add("canceledDate", canceledDate)
+      .add("cancelRequestedNode", cancelRequestedNode)
       .add("failedDate", failedDate)
       .toString
 
@@ -147,6 +152,7 @@ class TaskExecutionDetails(val taskId: TaskId,
     additionalInformation = additionalInformation,
     startedDate = startedDate,
     ranNode = ranNode,
+    cancelRequestedNode = cancelRequestedNode,
     completedDate = Optional.of(ZonedDateTime.now))
   private def fail = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.FAILED,
     submittedDate = submittedDate,
@@ -154,13 +160,15 @@ class TaskExecutionDetails(val taskId: TaskId,
     additionalInformation = additionalInformation,
     startedDate = startedDate,
     ranNode = ranNode,
+    cancelRequestedNode = cancelRequestedNode,
     failedDate = Optional.of(ZonedDateTime.now))
-  private def requestCancel = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCEL_REQUESTED,
+  private def requestCancel(hostname: Hostname) = new TaskExecutionDetails(taskId, `type`,
TaskManager.Status.CANCEL_REQUESTED,
     submittedDate = submittedDate,
     submittedNode = submittedNode,
     additionalInformation = additionalInformation,
     startedDate = startedDate,
     ranNode = ranNode,
+    cancelRequestedNode = Optional.of(hostname),
     canceledDate = Optional.of(ZonedDateTime.now))
   private def cancel = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCELLED,
     submittedDate = submittedDate,
@@ -168,5 +176,6 @@ class TaskExecutionDetails(val taskId: TaskId,
     additionalInformation = additionalInformation,
     startedDate = startedDate,
     ranNode = ranNode,
+    cancelRequestedNode = cancelRequestedNode,
     canceledDate = Optional.of(ZonedDateTime.now))
 }
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
b/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
index cdc241e..d585e74 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
@@ -50,11 +50,12 @@ class StartCommandHandler(private val loadHistory: TaskAggregateId =>
History,
   }
 }
 
-class RequestCancelCommandHandler(private val loadHistory: TaskAggregateId => History)
extends TaskCommandHandler[RequestCancel] {
+class RequestCancelCommandHandler(private val loadHistory: TaskAggregateId => History,
+                                  private val hostname: Hostname) extends TaskCommandHandler[RequestCancel]
{
   override def handledClass: Class[RequestCancel] = classOf[RequestCancel]
 
   override def handle(command: RequestCancel): util.List[_ <: Event] = {
-    loadAggregate(loadHistory, command.id).requestCancel()
+    loadAggregate(loadHistory, command.id).requestCancel(hostname)
   }
 }
 
@@ -80,4 +81,4 @@ class FailCommandHandler(private val loadHistory: TaskAggregateId =>
History) ex
   override def handle(command: Fail): util.List[_ <: Event] = {
     loadAggregate(loadHistory, command.id).fail()
   }
-}
+}
\ No newline at end of file
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
b/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
index 7302b4b..f57c947 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
@@ -42,7 +42,7 @@ class EventSourcingTaskManager @Inject @VisibleForTesting private[eventsourcing]
     case Created(aggregateId, _, task, _) =>
       val taskWithId = new TaskWithId(aggregateId.taskId, task)
       workQueue.submit(taskWithId)
-    case CancelRequested(aggregateId, _) =>
+    case CancelRequested(aggregateId, _, _) =>
       workQueue.cancel(aggregateId.taskId)
     case _ =>
   }
@@ -54,7 +54,7 @@ class EventSourcingTaskManager @Inject @VisibleForTesting private[eventsourcing]
     handlers = Set(
       new CreateCommandHandler(loadHistory, hostname),
       new StartCommandHandler(loadHistory, hostname),
-      new RequestCancelCommandHandler(loadHistory),
+      new RequestCancelCommandHandler(loadHistory, hostname),
       new CompleteCommandHandler(loadHistory),
       new CancelCommandHandler(loadHistory),
       new FailCommandHandler(loadHistory)),
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
index 61d7d8f..fa0607e 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
@@ -34,7 +34,7 @@ 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 CancelRequested(aggregateId: TaskAggregateId, override val eventId: EventId) extends
TaskEvent(aggregateId, eventId)
+case class CancelRequested(aggregateId: TaskAggregateId, override val eventId: EventId, hostname:
Hostname) extends TaskEvent(aggregateId, eventId)
 
 case class Completed(aggregateId: TaskAggregateId, override val eventId: EventId, result:
Result) extends TaskEvent(aggregateId, eventId)
 
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
index 24aef85..ae86433 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
@@ -50,9 +50,9 @@ class TaskAggregate private(val aggregateId: TaskAggregateId, private val
histor
     }
   }
 
-  def requestCancel(): util.List[Event] = {
+  def requestCancel(hostname: Hostname): util.List[Event] = {
     currentStatus match {
-      case Some(status) if !status.isFinished => createEventWithId(CancelRequested(aggregateId,
_))
+      case Some(status) if !status.isFinished => createEventWithId(CancelRequested(aggregateId,
_, hostname))
       case _ => Nil.asJava
     }
   }
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
index 5654e0f..a6ffff2 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
@@ -29,7 +29,7 @@ trait TaskExecutionDetailsProjection {
     case created: Created =>
       update(TaskExecutionDetails.from(created.task, created.aggregateId.taskId, created.hostname))
     case cancelRequested: CancelRequested =>
-      update(cancelRequested.aggregateId.taskId)(_.cancelRequested)
+      update(cancelRequested.aggregateId.taskId)(_.cancelRequested(hostname))
     case started: Started =>
       update(started.aggregateId.taskId)(_.started(hostname))
     case completed: Completed =>
diff --git a/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
b/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
index b8773a8..281e217 100644
--- a/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
+++ b/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
@@ -95,4 +95,20 @@ class EventSourcingTaskManagerTest implements TaskManagerContract {
                 .extracting("hostname")
                 .containsOnly(HOSTNAME));
     }
+
+    @Test
+    void cancelRequestedTaskShouldKeepOriginHostname() {
+        TaskId taskId = taskManager.submit(() -> {
+            Thread.sleep(100);
+            return Task.Result.COMPLETED;
+        });
+        taskManager.cancel(taskId);
+
+        TaskAggregateId aggregateId = new TaskAggregateId(taskId);
+        CALMLY_AWAIT.untilAsserted(() ->
+            assertThat(eventStore.getEventsOfAggregate(aggregateId).getEvents())
+                .filteredOn(event -> event instanceof CancelRequested)
+                .extracting("hostname")
+                .containsOnly(HOSTNAME));
+    }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org


Mime
View raw message